#!/usr/bin/env python3 # SPDX-License-Identifier: Apache-2.0 """End to end: the EV charger 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 the charger through the API (every change is 403 capability-not-allowed); * every staged fault shows on the owner's device page with its standard fault code; * no measurement, charging session or connector state ever reaches the site. Ports 8806 (site) and 43306 (local API), so it can run beside the other simulator tests. Needs: Node 20+, Python 3 with jsonschema and referencing (pip install "jsonschema[format]>=4.18" referencing). Usage, from the project's top folder: python3 evChargerSimulator/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 = 8806, 43306, 255 e2e.SITE_PORT, e2e.SITE = SITE_PORT, f"http://127.0.0.1:{SITE_PORT}" STAGED = ["sensors.stuck", "performance.too-hot", "power.supply-unstable", "power.supply-lost", "connectivity.link-lost", "system.main-function-lost", "actuators.not-responding"] NEVER_ON_SITE = ["kW.h", "kWh", "Cel", "°C", "connector_state", "session_energy", "suspended-ev", "grid_voltage", "mirror contact", "peakA"] 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, json.loads(r.read() or b"{}") except urllib.error.HTTPError as e: return e.code, json.loads(e.read() or b"{}") def status_of(method, path, key=None): return answer(method, path, key)[0] def main(): work = pathlib.Path(tempfile.mkdtemp(prefix="heart-ev-charger-")) 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 / "evChargerSimulator.py"), "--monitor-url", e2e.SITE, "--scenario", "all", "--speed", "5", "--seed", "7", "--duration", str(RUN_SEC), "--port", str(API_PORT), "--state-file", str(work / "ev-charger-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": "EV charger 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, names = [], [], [], set() 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])) names |= {e["data"]["name"] for e in get("/heart/v1/properties")["items"]} 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("public properties are only the in-service state", names == {"org.heartaispec.demo.ev-charger.charger_state"}, str(sorted(names))) check("private charging sessions need the owner key", status_of("GET", "/heart/v1/sessions") == 403 and status_of("GET", "/heart/v1/sessions", key) == 200) sessions = get("/heart/v1/sessions", key)["items"] check("charging sessions hold times and energy only: no card or vehicle ID", bool(sessions) and all(set(s) <= {"start", "end", "kWh", "peakA"} for s in sessions), str(sessions[:2])) refused = [answer(m_, p_, k_) for m_ in ("PUT", "POST", "PATCH", "DELETE") for p_ in ("/heart/v1/properties", "/heart/v1/sessions", "/heart/v1/capabilities/start") for k_ in (None, key)] check("nothing can change the charger through the API (403 capability-not-allowed, even with the owner key)", all(s_ == 403 and b_.get("code") == "capability-not-allowed" for s_, b_ in refused), str(refused[:1])) shown = set(re.findall(r"(?:system|sensors|power|connectivity|performance|actuators)\.[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, session or connector state 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())