#!/usr/bin/env python3 # SPDX-License-Identifier: Apache-2.0 """End to end: the pump station 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; * every change request is refused (403 capability-not-allowed): the PLC runs the station, HEART only watches; * every staged fault shows on the owner's device page with its standard fault code; * no level, flow, pressure, motor reading or PLC event 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 pumpStationSimulator/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 = 8805, 43305, 290 e2e.SITE_PORT, e2e.SITE = SITE_PORT, f"http://127.0.0.1:{SITE_PORT}" STAGED = ["sensors.stuck", "connectivity.link-lost", "system.main-function-degraded", "performance.too-hot", "system.main-function-lost", "power.supply-lost"] NEVER_ON_SITE = ["m3/h", "mm/s", " Cel", "reservoir", "sump", "thermal-trip", "high-pressure", "winding", "UPS"] 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 refusal(method, path, key=None): req = urllib.request.Request(f"http://127.0.0.1:{API_PORT}{path}", data=b"{}", method=method, headers={"Authorization": f"Bearer {key}"} if key else {}) try: with urllib.request.urlopen(req, timeout=10) as r: return r.status, None except urllib.error.HTTPError as e: return e.code, json.loads(e.read() or b"{}").get("code") def main(): work = pathlib.Path(tempfile.mkdtemp(prefix="heart-pump-station-")) 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 / "pumpStationSimulator.py"), "--monitor-url", e2e.SITE, "--scenario", "all", "--speed", "5", "--seed", "7", "--duration", str(RUN_SEC), "--port", str(API_PORT), "--state-file", str(work / "pump-station-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": "Pump station 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("the PLC event log needs the owner key", status_of("GET", "/heart/v1/plc-events") == 403 and status_of("GET", "/heart/v1/plc-events", key) == 200) check("measurements need the owner key", get("/heart/v1/properties")["items"] == [] and len(get("/heart/v1/properties", key)["items"]) > 10) refused = [(m_, p_, refusal(m_, p_, k_)) for m_ in ("PUT", "POST", "PATCH", "DELETE") for p_ in ("/heart/v1/properties", "/heart/v1/actions", "/heart/v1/invoke") for k_ in (None, key)] check("every change request is refused with 403 capability-not-allowed, even with the owner key", all(r_ == (403, "capability-not-allowed") for _, _, r_ in refused), str([x for x in refused if x[2] != (403, "capability-not-allowed")][:3])) shown = set(re.findall(r"(?:system|sensors|power|connectivity|performance)\.[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 PLC event 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())