#!/usr/bin/env python3 # SPDX-License-Identifier: Apache-2.0 """End to end: the dog collar 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 event at 5x speed (simulated clock from 09:00), signs in with an emailed code, registers the device through the site's own form, and checks for about four and a half minutes that: * every message the simulator serves is valid against the official schemas (spec/schemas/1.0), the private safe-zone events (heart.rhythm) and the owner's Actions (heart.action) included; * the public views hold nothing but public data, and no location of any kind: no position, no distance, no safe-zone state, no capabilities; the track, the safe-zone events and the Actions need the owner key; * the owner turns lost mode on and off at once with the owner key (not physical), and its Action shows the battery it costs; * an LED change and a buzzer (physical: they act on the dog) wait: an approval over the network with only the key is refused with 403 presence-required, and the change runs only after the button file is touched; * the buzzer's welfare guard refuses a loud, long or too-soon buzz (409 guard-failed), and the hardware range is enforced (422); every other change is refused (403 capability-not-allowed); * the dog wandering out of the safe zone is a private event (zone.left, alone) and never an alert; * every staged fault shows on the owner's device page with its standard fault code; * no position, coordinate, safe-zone event, activity or lost mode ever reaches the site; * the lifecycle (sleep and restart), on a second simulator with no staged events: the Card lists them with wake sources timer and button and no wake; every refusal of API Standard 13.2 rule 3 (404, 501, 400); a sleep stops answering at all (the connection is refused) until its stated time, its Action carries nextReachableAt and then says succeeded; the button wakes it early; a restart comes back; and, on the main simulator, the collar refuses to sleep (409 conflict) while lost mode is on or while the dog is outside the safe zone. Needs: Node 20+, Python 3 with jsonschema and referencing (pip install "jsonschema[format]>=4.18" referencing). Usage, from the project's top folder: python3 dogCollarSimulator/test_e2e.py """ import calendar 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 = 8808, 43308, 280 LIFE_PORT = 43309 # the second simulator, for the lifecycle e2e.SITE_PORT, e2e.SITE = SITE_PORT, f"http://127.0.0.1:{SITE_PORT}" STAGED = ["power.battery-low", "connectivity.link-lost", "system.main-function-degraded", "performance.too-hot", "sensors.stuck", "sensors.no-reading"] CODES = STAGED + ["system.main-function-lost"] PUBLIC_PROPS = {"battery_pct", "battery_status"} # Words a page could only show if a reading leaked ("position" alone is not here: the page's CSS has position:sticky). NEVER_ON_SITE = ["eastM", "northM", "east_m", "north_m", "position_", "distance", "safe zone", "safe_zone", "zone.left", "zone.returned", "heat.risk", "withOwner", "lost_mode", "lost mode", "home_base", "home base", "activity", "walking", "running", "resting", "Cel", "dBm", "hdop", "satellites"] 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 call(method, path, key=None, body=None, port=None): data = json.dumps(body).encode() if body is not None else (b"{}" if method != "GET" else None) req = urllib.request.Request(f"http://127.0.0.1:{port or API_PORT}{path}", data=data, method=method, headers={"Content-Type": "application/json", **({"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: try: return e.code, json.loads(e.read() or b"{}") except ValueError: return e.code, {} def get(path, key=None): s, b = call("GET", path, key) if s != 200: raise OSError(f"GET {path} answered {s}") return b def prop(name, key): return next(p["data"]["value"] for p in get("/heart/v1/properties", key)["items"] if p["data"]["name"].endswith("." + name)) def invoke(cap, args, key): return call("POST", f"/heart/v1/capabilities/{cap}/invoke", key, {"arguments": args}) def answers(port): """True if the simulator answers HTTP at all; False if the connection is refused (asleep or restarting).""" try: call("GET", "/heart/v1/pulse", port=port) return True except (urllib.error.URLError, OSError): return False def stamp(text): return calendar.timegm(time.strptime(text, "%Y-%m-%dT%H:%M:%SZ")) def lifecycle_tests(work): """Sleep and restart, on a simulator of its own (no staged events, the dog resting at home at noon), so the main run's staged events keep their timing. Speed 5: a sleep of 20 simulated seconds lasts 4 real seconds.""" life = work / "lifecycle" life.mkdir() log_path = life / "simulator.log" with open(log_path, "w") as log: sim = subprocess.Popen([sys.executable, str(HERE / "dogCollarSimulator.py"), "--no-monitor", "--speed", "5", "--seed", "3", "--clock", "12:00", "--duration", "90", "--port", str(LIFE_PORT)], stdout=log, stderr=log, cwd=life) try: key = e2e.until(lambda: re.search(r"owner key for private views: (\S+)\)", log_path.read_text()), 15, "the lifecycle simulator").group(1) msgs = [] def post(op, body, k=key): s_, b_ = call("POST", f"/heart/v1/lifecycle/{op}", k, body, LIFE_PORT) return s_, b_ def lget(path, k=key): s_, b_ = call("GET", path, k, port=LIFE_PORT) msgs.append(b_) return b_ card, pub_card = lget("/.well-known/heart-card.json"), lget("/.well-known/heart-card.json", None) lc = card["data"].get("lifecycle", {}) check("the Card lists the lifecycle module: sleep and restart, woken by timer or button, and no wake (rule 3.3)", "lifecycle" in card["data"]["modules"] and set(lc.get("operations", [])) == {"sleep", "restart"} and lc.get("wakeSources") == ["timer", "button"] and "lifecycle" not in pub_card["data"], str(lc)) view, pub_view = lget("/heart/v1/lifecycle"), lget("/heart/v1/lifecycle", None) check("GET /heart/v1/lifecycle: running; the public view holds only the phase", view["data"]["phase"] == "running" and "sleep" in view["data"]["allowed"] and view["data"]["wakeSources"] == ["timer", "button"] and pub_view["data"] == {"phase": "running"} and pub_view["heartprivacy"] == "public", f"{view['data']} {pub_view['data']}") refusals = { "an unknown operation (404 not-found)": (post("frobnicate", {}), 404, "not-found"), "wake, not offered (501 not-implemented)": (post("wake", {}), 501, "not-implemented"), "shutdown, not offered (501 not-implemented)": (post("shutdown", {}), 501, "not-implemented"), "no owner key (401)": (post("sleep", {"durationSec": 5}, None), 401, "unauthenticated"), "an unknown field (400)": (post("sleep", {"durationSec": 5, "colour": "red"}), 400, "invalid-request"), "delaySec above 0, no schedules module (400)": (post("sleep", {"durationSec": 5, "delaySec": 10}), 400, "invalid-request"), "both durationSec and wakeAt (400)": (post("sleep", {"durationSec": 5, "wakeAt": "2099-01-01T00:00:00Z"}), 400, "invalid-request"), "neither durationSec nor wakeAt (400)": (post("sleep", {}), 400, "invalid-request"), "depth light (400)": (post("sleep", {"durationSec": 5, "depth": "light"}), 400, "invalid-request"), "depth hibernate (400)": (post("sleep", {"durationSec": 5, "depth": "hibernate"}), 400, "invalid-request"), "durationSec 0 or above 604800 (400)": (post("sleep", {"durationSec": 604801}), 400, "invalid-request"), "wakeAt in the past (400)": (post("sleep", {"wakeAt": "2020-01-01T00:00:00Z"}), 400, "invalid-request"), "operation in the body unlike the path (400)": (post("sleep", {"operation": "restart", "durationSec": 5}), 400, "invalid-request"), "a force restart over the network (403 presence-required)": (post("restart", {"mode": "force"}), 403, "presence-required"), } bad = [f"{k}: {r[0][0]} {r[0][1].get('code')}" for k, r in refusals.items() if (r[0][0], r[0][1].get("code")) != r[1:]] bad += ["durationSec 0"] if post("sleep", {"durationSec": 0})[0] != 400 else [] check("lifecycle refusals per rule 3: unknown operation 404, wake and shutdown 501, unknown field, delaySec, " "both or neither of durationSec and wakeAt, depth other than deep: 400", not bad and answers(LIFE_PORT), str(bad)) # Sleep for 20 simulated seconds: 4 real seconds at 5x. t0 = time.time() s, a = post("sleep", {"durationSec": 20, "depth": "deep", "reason": "test"}) msgs.append(a) d = a.get("data", {}) nra = stamp(d["nextReachableAt"]) if d.get("nextReachableAt") else 0 check("sleep answers 202 with a running write Action that says when it answers again (nextReachableAt)", s == 202 and d["status"] == "running" and d["risk"] == "write" and d["physical"] is False and d["operation"] == "POST /heart/v1/lifecycle/sleep" and d["cancellable"] is False and t0 + 4 <= nra <= t0 + 12, f"{s} {d}") down = e2e.until(lambda: not answers(LIFE_PORT), 3, "the collar to fall asleep") asleep = [] for _ in range(3): time.sleep(0.4) asleep.append(answers(LIFE_PORT)) check("asleep, the collar does not answer at all: the connection is refused", down and not any(asleep), str(asleep)) e2e.until(lambda: answers(LIFE_PORT), 15, "the collar to wake") woke = time.time() act = lget(f"/heart/v1/actions/{d['id']}")["data"] check("it wakes by its timer at the stated time, and the Action says succeeded", t0 + 3.5 <= woke <= nra + 2 and act["status"] == "succeeded" and act["result"]["wokeBy"] == "timer" and act["result"]["sleptSimulatedSec"] >= 18, f"after {woke - t0:.1f} s {act}") text = log_path.read_text() check("while asleep the simulated sensor work stops (the console says so)", "asleep (deep)" in text and "awake again, woken by its timer" in text, text[-400:]) # Sleep until a stated time 10 minutes away (50 simulated minutes), woken early by the button. at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(time.time() + 600)) s, a = post("sleep", {"wakeAt": at}) msgs.append(a) d = a.get("data", {}) e2e.until(lambda: not answers(LIFE_PORT), 3, "the collar to fall asleep") (life / "button.press").touch() e2e.until(lambda: answers(LIFE_PORT), 8, "the button to wake the collar") act = lget(f"/heart/v1/actions/{d['id']}")["data"] check("sleep until wakeAt: the button wakes it early, and the Action says so", s == 202 and d["nextReachableAt"] >= at and act["status"] == "succeeded" and act["result"]["wokeBy"] == "button", f"{s} {act}") # A graceful restart: down for a few seconds, then back. s, a = post("restart", {}) msgs.append(a) d = a.get("data", {}) e2e.until(lambda: not answers(LIFE_PORT), 3, "the collar to restart") e2e.until(lambda: answers(LIFE_PORT), 12, "the collar to come back") act = lget(f"/heart/v1/actions/{d['id']}")["data"] check("a graceful restart: 202 with nextReachableAt, unreachable for a moment, then succeeded", s == 202 and d.get("nextReachableAt") and act["status"] == "succeeded", f"{s} {act}") msgs += get_actions(key) invalid = [(m.get("type"), x.message[:160]) for m in msgs for x in validator.iter_errors(m)] check("every lifecycle message is valid against the official schemas (card, lifecycle, actions)", not invalid and len(msgs) > 8, str(invalid[:3])) finally: sim.terminate() sim.wait(timeout=10) def get_actions(key): return call("GET", "/heart/v1/actions", key, port=LIFE_PORT)[1]["items"] def main(): work = pathlib.Path(tempfile.mkdtemp(prefix="heart-dog-collar-")) 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") lifecycle_tests(work) with open(log_path, "w") as log: sim = subprocess.Popen([sys.executable, str(HERE / "dogCollarSimulator.py"), "--monitor-url", e2e.SITE, "--scenario", "all", "--speed", "5", "--seed", "7", "--clock", "09:00", "--duration", str(RUN_SEC), "--port", str(API_PORT), "--state-file", str(work / "dog-collar-monitor.json")], stdout=log, stderr=log, cwd=work) sim_start = time.time() 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": "Dog collar 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) # Lost mode: not physical (it changes reporting, nothing the dog feels), so it runs at once with the key. s, a = invoke("collar.lost_mode", {"on": True}, key) d = a.get("data", {}) r = d.get("result", {}) check("the owner turns lost mode on at once with the owner key (not physical): reports every 30 s", s == 200 and d["status"] == "succeeded" and d["physical"] is False and d["initiator"]["kind"] == "human" and r["before"]["reportIntervalSec"] == 600 and r["after"]["reportIntervalSec"] == 30 and prop("lost_mode", key) is True and prop("report_interval_s", key) == 30, f"{s} {a}") check("the lost-mode Action shows the battery it costs", r["after"]["batteryHoursLeft"] < r["before"]["batteryHoursLeft"] / 3, str(r)) s, p = call("POST", "/heart/v1/lifecycle/sleep", key, {"durationSec": 60}) check("safety: the collar refuses to sleep while lost mode is on (409 conflict), and stays awake", s == 409 and p.get("code") == "conflict" and "lost mode" in p.get("detail", "").lower() and get("/heart/v1/lifecycle", key)["data"]["allowed"] == ["restart"], f"{s} {p}") s, a = invoke("collar.lost_mode", {"on": False}, key) check("the owner turns lost mode off again: back to every 10 minutes", s == 200 and a["data"]["status"] == "succeeded" and prop("lost_mode", key) is False and prop("report_interval_s", key) == 600, f"{s} {a}") s1 = invoke("collar.lost_mode", {"on": True}, None)[0] s2 = invoke("collar.lost_mode", {"on": "yes"}, key)[0] check("no change runs without the key (401), and a malformed one is refused (400)", s1 == 401 and s2 == 400, f"{s1} {s2}") # The LED: physical (light at a living animal's neck). It waits for the owner's passkey (here: the button file). led_before = prop("led_mode", key) s, a = invoke("collar.led", {"mode": "blink"}, key) aid = a.get("data", {}).get("id", "") check("an LED change is physical and waits for a person (202 awaiting-approval)", s == 202 and a["data"]["status"] == "awaiting-approval" and a["data"]["physical"] is True, f"{s} {a}") s1, p1 = call("POST", f"/heart/v1/actions/{aid}/decision", key, {"decision": "approve"}) s2, p2 = call("POST", f"/heart/v1/actions/{aid}/decision", key, {"decision": "approve", "presence": {"method": "webauthn", "digest": "0" * 64, "proof": "x"}}) still = get(f"/heart/v1/actions/{aid}", key)["data"]["status"] check("approving the LED change over the network with only the key is refused (403 presence-required)", s1 == 403 and p1.get("code") == "presence-required" and s2 == 403 and still == "awaiting-approval" and prop("led_mode", key) == led_before, f"{s1} {p1.get('code')} {s2} {still}") (work / "button.press").touch() done = e2e.until(lambda: get(f"/heart/v1/actions/{aid}", key)["data"]["status"] != "awaiting-approval", 10, "the button") act = get(f"/heart/v1/actions/{aid}", key)["data"] check("the LED change runs once the owner confirms (touch button.press)", done and act["status"] == "succeeded" and act["decision"]["presenceMethod"] == "device-button" and prop("led_mode", key) == "blink", str(act)) # The buzzer: physical, and the collar's welfare guard refuses loud, long or frequent buzzing. g1, b1 = invoke("collar.buzzer", {"volume": 80, "durationSec": 3}, key) g2 = invoke("collar.buzzer", {"volume": 40, "durationSec": 10}, key)[0] g3 = invoke("collar.buzzer", {"volume": 200, "durationSec": 3}, key)[0] check("the buzzer guard refuses above 50 % or 5 s (409 guard-failed), and the hardware range is enforced (422)", g1 == 409 and b1.get("code") == "guard-failed" and g2 == 409 and g3 == 422, f"{g1} {g2} {g3}") s, a = invoke("collar.buzzer", {"volume": 40, "durationSec": 3}, key) bid = a.get("data", {}).get("id", "") s1, p1 = call("POST", f"/heart/v1/actions/{bid}/decision", key, {"decision": "approve"}) check("a safe buzz is physical too: it waits, and the key alone cannot approve it (403 presence-required)", s == 202 and a["data"]["physical"] is True and s1 == 403 and p1.get("code") == "presence-required", f"{s} {s1}") (work / "button.press").touch() e2e.until(lambda: get(f"/heart/v1/actions/{bid}", key)["data"]["status"] != "awaiting-approval", 10, "the button") act = get(f"/heart/v1/actions/{bid}", key)["data"] again = invoke("collar.buzzer", {"volume": 40, "durationSec": 3}, key)[0] check("the buzz runs after the owner confirms, and another within 60 s is refused (409 guard-failed)", act["status"] == "succeeded" and again == 409, f"{act['status']} {again}") check("other changes are refused (403 capability-not-allowed)", all(call(m_, p_, key)[0] == 403 for m_, p_ in ( ("PUT", "/heart/v1/properties"), ("PATCH", "/heart/v1/properties/lost_mode"), ("POST", "/heart/v1/capabilities/collar.safe_zone/invoke"), ("POST", "/heart/v1/capabilities/collar.gnss/invoke"), ("DELETE", "/heart/v1/rhythm"), ("PUT", "/heart/v1/tending/autonomy")))) e2e.until(lambda: prop("in_safe_zone", key) is False, 90, "the dog to wander out of the safe zone") s, p = call("POST", "/heart/v1/lifecycle/sleep", key, {"durationSec": 60}) check("safety: the collar refuses to sleep while the dog is outside the safe zone (409 conflict)", s == 409 and p.get("code") == "conflict" and "safe zone" in p.get("detail", ""), f"{s} {p}") invalid, leaks, pages, coords = [], [], [], set() end = sim_start + RUN_SEC - 15 # while the simulator still runs while time.time() < end and sim.poll() is None: try: rhythm = get("/heart/v1/rhythm", key)["items"] track = get("/heart/v1/track", key)["items"] msgs = ([get("/.well-known/heart-card.json"), get("/.well-known/heart-card.json", key), get("/heart/v1/pulse"), get("/heart/v1/pulse", key), get("/heart/v1/lifecycle"), get("/heart/v1/lifecycle", key)] + get("/heart/v1/arrhythmias")["items"] + get("/heart/v1/arrhythmias", key)["items"] + get("/heart/v1/properties")["items"] + get("/heart/v1/properties", key)["items"] + rhythm + get("/heart/v1/actions", key)["items"]) pub_card = get("/.well-known/heart-card.json") public = (get("/heart/v1/properties")["items"] + get("/heart/v1/arrhythmias")["items"] + [get("/heart/v1/pulse"), get("/heart/v1/lifecycle")]) 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])) leaks += [e["type"] for e in public + [pub_card] if e["heartprivacy"] != "public"] leaks += [k for e in public if e["type"] == "heart.arrhythmia" for k in e["data"] if k not in ("id", "code", "severity", "status", "detectedAt", "resolvedAt")] leaks += [e["data"]["name"] for e in public if e["type"] == "heart.property" and e["data"]["name"].rsplit(".", 1)[1] not in PUBLIC_PROPS] leaks += ["pulse reconnaissance" for e in public if e["type"] == "heart.pulse" and "reconnaissance" in e["data"]] if "capabilities" in pub_card["data"] or "lifecycle" in pub_card["data"]: leaks.append("card capabilities or lifecycle") leaks += [f"lifecycle {k}" for e in public if e["type"] == "heart.lifecycle" for k in e["data"] if k != "phase"] for f in track: coords |= {f["eastM"], f["northM"]} for ev in rhythm: at = ev["data"].get("attributes", {}) coords |= {at[k] for k in ("eastM", "northM") if k in at} pages.append(e2e.http("GET", f"/device/{dev}", cookie=owner)[2]) time.sleep(10) check("every served message is valid against the official schemas (card, pulse, lifecycle, alerts, properties, " "rhythm, actions)", not invalid, str(invalid[:3])) check("public views carry only public data, and no location of any kind", not leaks, str(sorted(set(leaks)))) check("the track, the safe-zone events and the Actions need the owner key", all(call("GET", p)[0] == 403 for p in ("/heart/v1/track", "/heart/v1/rhythm", "/heart/v1/actions")) and call("GET", "/heart/v1/track", key)[0] == 200) events = [e["data"] for e in get("/heart/v1/rhythm", key)["items"]] alone = [e for e in events if e["name"] == "zone.left" and e["attributes"]["withOwner"] is False] alerts = [e["data"]["code"] for e in get("/heart/v1/arrhythmias", key)["items"]] check("the dog wandering off alone is a private event (zone.left, then zone.returned), never an alert", alone and any(e["name"] == "zone.returned" for e in events) and set(alerts) <= set(CODES) and all(e["heartprivacy"] == "private" for e in get("/heart/v1/rhythm", key)["items"]), f"events {[e['name'] for e in events]} alerts {sorted(set(alerts))}") check("the hot car is two things: a private warning about the dog (heat.risk) and the collar's own fault " "(performance.too-hot)", any(e["name"] == "heat.risk" for e in events) and "performance.too-hot" in alerts, f"events {[e['name'] for e in events]}") shown = set(re.findall(r"(?:system|sensors|power|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)] big = sorted(f"{v:.1f}" for v in coords if abs(v) >= 10) on_page = [c for c in big if any(re.search(r"(? 5, f"found {found} coordinates {on_page[:5]} (of {len(big)})") 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())