#!/usr/bin/env python3 # SPDX-License-Identifier: Apache-2.0 """End to end: the 3D printer 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 three and a half 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 heat, move or pause the printer through the API (403 capability-not-allowed); * every staged fault shows on the owner's device page with its standard fault code; * no temperature, job name or other measurement 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 printer3dSimulator/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 = 8807, 43307, 210 e2e.SITE_PORT, e2e.SITE = SITE_PORT, f"http://127.0.0.1:{SITE_PORT}" STAGED = ["system.main-function-degraded", "performance.too-hot", "sensors.stuck", "connectivity.link-lost", "actuators.position-mismatch", "system.main-function-lost", "sensors.out-of-range"] NEVER_ON_SITE = ["Cel", ".gcode", "gift-box", "hotend", "filament", "stall", "thermistor"] 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 answer(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: try: return e.code, json.loads(e.read() or b"{}").get("code", "") except ValueError: return e.code, "" def status_of(method, path, key=None): return answer(method, path, key)[0] def main(): work = pathlib.Path(tempfile.mkdtemp(prefix="heart-printer3d-")) 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 / "printer3dSimulator.py"), "--monitor-url", e2e.SITE, "--scenario", "all", "--speed", "5", "--seed", "7", "--duration", str(RUN_SEC), "--port", str(API_PORT), "--state-file", str(work / "printer3d-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": "3D printer 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 print jobs (file names) need the owner key", status_of("GET", "/heart/v1/jobs") == 403 and status_of("GET", "/heart/v1/jobs", key) == 200) private = [e["data"]["name"] for e in get("/heart/v1/properties", key)["items"] if e["data"]["privacy"] == "private"] check("the job name is private", private == ["org.heartaispec.demo.printer-3d.job_name"], str(private)) tries = [(m_, p_) for m_ in ("PUT", "POST", "PATCH", "DELETE") for p_ in ("/heart/v1/properties", "/heart/v1/properties/org.heartaispec.demo.printer-3d.hotend_target", "/heart/v1/invoke/pause", "/heart/v1/jobs")] refused = [answer(m_, p_, key) for m_, p_ in tries] check("nothing can heat, move or pause the printer through the API (403 capability-not-allowed)", all(r == (403, "capability-not-allowed") for r in refused), str(set(refused))) shown = set(re.findall(r"(?:system|sensors|actuators|performance|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 job name 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())