#!/usr/bin/env python3 # SPDX-License-Identifier: Apache-2.0 """End to end: the pacemaker simulator reports to a HEART Monitor site running on this machine. Starts the monitor site (heart-monitor/test/serve.mjs, with an in-memory database and an email outbox), runs the simulator with every staged fault at 5x speed, signs in with an emailed code, registers the device through the site's own form, and checks for about four minutes that: * every message the simulator serves is valid against the official schemas (spec/schemas/1.0); * the public views hold nothing but public data, and private views need the owner key; * nothing can change pacing through the API; * every staged fault shows on the owner's device page with its standard fault code; * no measurement, heart rate or rhythm episode ever reaches the site. Needs: Node 20+, Python 3 with jsonschema and referencing (pip install "jsonschema[format]>=4.18" referencing). Usage, from the project's top folder: python3 pacemakerSimulator/test_e2e.py """ import json import pathlib import re import subprocess import sys import tempfile import time import urllib.error import urllib.request HERE = pathlib.Path(__file__).resolve().parent ROOT = HERE.parent sys.path.insert(0, str(ROOT / "heart-monitor" / "test")) import e2e # noqa: E402 the monitor site's own end-to-end helpers (sign in, forms, outbox) import jsonschema # noqa: E402 from referencing import Registry, Resource # noqa: E402 SITE_PORT, API_PORT, RUN_SEC = 8791, 43299, 230 e2e.SITE_PORT, e2e.SITE = SITE_PORT, f"http://127.0.0.1:{SITE_PORT}" STAGED = ["sensors.noisy", "connectivity.link-lost", "sensors.out-of-range", "system.main-function-lost", "power.battery-low"] NEVER_ON_SITE = ["ohm", "Ohm", "/min", "af.started", "impedance", "threshold", "mV"] registry = Registry() for p in (ROOT / "spec" / "schemas" / "1.0").glob("*.json"): d = json.loads(p.read_text()) registry = registry.with_resource(d["$id"], Resource.from_contents(d)) validator = jsonschema.Draft202012Validator(json.loads((ROOT / "spec/schemas/1.0/envelope.schema.json").read_text()), registry=registry) results = [] def check(name, ok, detail=""): results.append(ok) print(f"{'PASS' if ok else 'FAIL'} {name}{'' if ok else ': ' + detail}", flush=True) def get(path, key=None): req = urllib.request.Request(f"http://127.0.0.1:{API_PORT}{path}", headers={"Authorization": f"Bearer {key}"} if key else {}) with urllib.request.urlopen(req, timeout=10) as r: return json.loads(r.read()) def status_of(method, path, key=None): req = urllib.request.Request(f"http://127.0.0.1:{API_PORT}{path}", data=b"{}" if method != "GET" else None, method=method, headers={"Authorization": f"Bearer {key}"} if key else {}) try: with urllib.request.urlopen(req, timeout=10) as r: return r.status except urllib.error.HTTPError as e: return e.code def main(): work = pathlib.Path(tempfile.mkdtemp(prefix="heart-pacemaker-")) site = subprocess.Popen(["node", str(ROOT / "heart-monitor/test/serve.mjs"), str(SITE_PORT)], stderr=subprocess.DEVNULL) log_path = work / "simulator.log" sim = None try: def site_up(): try: return e2e.http("GET", "/monitor")[0] == 200 except OSError: return False e2e.until(site_up, 20, "the monitor site") with open(log_path, "w") as log: sim = subprocess.Popen([sys.executable, str(HERE / "pacemakerSimulator.py"), "--monitor-url", e2e.SITE, "--scenario", "all", "--speed", "5", "--seed", "7", "--duration", str(RUN_SEC), "--port", str(API_PORT), "--state-file", str(work / "pacemaker-monitor.json")], stdout=log, stderr=log, cwd=work) text = log_path.read_text m = e2e.until(lambda: re.search(r"register\?id=([0-9a-f]{16})\n\s+3\. Enter the claim code\s+([0-9A-Z-]{23})", text()), 20, "the claim code on the console") dev, code = m.group(1), m.group(2) owner = e2e.login("owner@example.com") s = e2e.http("POST", "/register", {"id": dev, "code": code.lower(), "name": "Pacemaker demo"}, cookie=owner)[0] check("the owner registers the device with its claim code", s in (200, 303), f"status {s}") key = e2e.until(lambda: re.search(r"owner key for private views: (\S+)\)", text()), 10, "owner key").group(1) invalid, leaks, pages = [], [], [] end = time.time() + RUN_SEC - 15 while time.time() < end and sim.poll() is None: try: msgs = ([get("/.well-known/heart-card.json"), get("/heart/v1/pulse"), get("/heart/v1/pulse", key)] + get("/heart/v1/arrhythmias")["items"] + get("/heart/v1/arrhythmias", key)["items"] + get("/heart/v1/properties")["items"] + get("/heart/v1/properties", key)["items"]) except OSError: break for e in msgs: errs = [x.message[:160] for x in validator.iter_errors(e)] if errs: invalid.append((e.get("type"), errs[0])) public = get("/heart/v1/properties")["items"] + get("/heart/v1/arrhythmias")["items"] + [get("/heart/v1/pulse")] leaks += [e["type"] for e in public if e["heartprivacy"] != "public"] leaks += [k for e in get("/heart/v1/arrhythmias")["items"] for k in e["data"] if k not in ("id", "code", "severity", "status", "detectedAt", "resolvedAt")] pages.append(e2e.http("GET", f"/device/{dev}", cookie=owner)[2]) time.sleep(10) check("every served message is valid against the official schemas", not invalid, str(invalid[:3])) check("public views carry only public data", not leaks, str(sorted(set(leaks)))) check("private rhythm episodes need the owner key", status_of("GET", "/heart/v1/rhythm") == 403 and status_of("GET", "/heart/v1/rhythm", key) == 200) check("nothing can change pacing through the API", all(status_of(m_, "/heart/v1/properties", key) == 403 for m_ in ("PUT", "POST", "PATCH", "DELETE"))) shown = set(re.findall(r"(?:system|sensors|power|connectivity)\.[a-z-]+", " ".join(pages))) missing = [c for c in STAGED if c not in shown] check("every staged fault shows on the owner's page with its fault code", not missing, f"missing {missing}") found = [w for w in NEVER_ON_SITE if any(w in p for p in pages)] check("no measurement or rhythm data reaches the site", not found, f"found {found}") public_page = e2e.http("GET", f"/device/{dev}")[0] check("the device page is private until shared", public_page in (403, 404), f"status {public_page}") finally: if sim: try: sim.wait(timeout=RUN_SEC + 60) except subprocess.TimeoutExpired: sim.kill() site.terminate() site.wait(timeout=10) print(f"\n{sum(results)}/{len(results)} passed (simulator log: {log_path})") return 0 if results and all(results) else 1 if __name__ == "__main__": sys.exit(main())