#!/usr/bin/env python3 # SPDX-License-Identifier: Apache-2.0 """End to end: the online shop server 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 server 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; * the owner can run a Playbook's action by hand (human, not physical: it runs at once), and guards still apply; * destructive requests (delete orders, refund, deploy, change the payment provider) are refused, even with the key; * all three Playbooks fired as initiator rule, stayed within maxPerDay and cooldownSec, were refused at the daily limit, and the one whose fix did not help was suspended and escalated instead of looping; * Stop works at once, and turning automatic action back on needs a person (the local button), not only the key; * every staged fault shows on the owner's device page with its standard fault code; * no measurement, order, revenue or action detail ever reaches the site. First, on a second simulator of its own (local only), the lifecycle operations (API Standard 13.2 rule 3): * drain stops new orders with a clear refusal and finishes the running ones; refusals follow rule 3; * a graceful restart drains first, fails (never forced) when an order cannot finish by timeoutSec, and can be cancelled while it drains; when it succeeds the API is briefly gone and the shop comes back with state reloaded; * a force restart is destructive: held for a person, nothing runs until the button at the server is pressed. Needs: Node 20+, Python 3 with jsonschema and referencing (pip install "jsonschema[format]>=4.18" referencing). Usage, from the project's top folder: python3 onlineShopSimulator/test_e2e.py """ import json import pathlib import re import subprocess import sys import tempfile import time import urllib.error import urllib.request import uuid 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 = 8804, 43304, 250 LIFE_PORT = 43305 # the second simulator, for the lifecycle operations e2e.SITE_PORT, e2e.SITE = SITE_PORT, f"http://127.0.0.1:{SITE_PORT}" STAGED = ["software.crashed", "software.crash-loop", "storage.nearly-full", "performance.memory-exhausted", "connectivity.dependency-unreachable", "performance.overloaded", "performance.backlog", "performance.slow"] NEVER_ON_SITE = ["revenue", "orders_today", "restartWorker", "clearCache", "rotateLogs", "worker-", "hit rate", "req/s", "Playbook restart", "payment provider is down"] LIMITS = {"restartWorker": (6, 60), "clearCache": (2, 120), "rotateLogs": (4, 300)} # maxPerDay, cooldownSec PLAYBOOKS = {"restart-stuck-worker": "restartWorker", "clear-bloated-cache": "clearCache", "rotate-full-logs": "rotateLogs"} SIM = "org.heartaispec.demo.sim-sec" 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) problems = jsonschema.Draft202012Validator(json.loads((ROOT / "spec/schemas/1.0/problem.schema.json").read_text()), registry=registry) requests = jsonschema.Draft202012Validator(json.loads((ROOT / "spec/schemas/1.0/lifecycle-request.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, idem=True, port=None): headers = {"Authorization": f"Bearer {key}"} if key else {} if body is not None: headers["Content-Type"] = "application/json" if idem is True: headers["Idempotency-Key"] = str(uuid.uuid4()) elif idem: headers["Idempotency-Key"] = idem 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=headers) 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, body = call("GET", path, key) if s != 200: raise OSError(f"GET {path} answered {s}") return body def lifecycle_tests(work): """drain, restart and force restart, on a second simulator of its own: local only, no staged faults.""" here = work / "lifecycle" here.mkdir() log_path = here / "simulator.log" with open(log_path, "w") as log: sim = subprocess.Popen([sys.executable, str(HERE / "onlineShopSimulator.py"), "--no-monitor", "--speed", "5", "--seed", "3", "--duration", "240", "--port", str(LIFE_PORT)], stdout=log, stderr=log, cwd=here) bad = [] def note(b): # every HEART reply: an envelope, a list of them, or a problem for e in (b.get("items", []) if "items" in b else [b]) if isinstance(b, dict) and b else []: v = problems if "specversion" not in e else validator bad.extend((e.get("type"), x.message[:160]) for x in v.iter_errors(e)) return b def lc(method, path, body=None, k=True, idem=True): try: st, b = call(method, path, key if k else None, body, idem, port=LIFE_PORT) except OSError: return 0, {} # not answering: restarting return st, note(b) def action(aid): return lc("GET", f"/heart/v1/actions/{aid}")[1].get("data", {}) def until_status(aid, statuses, sec, watch=None): end, a = time.time() + sec, {} while time.time() < end: if watch: watch() a = action(aid) or a if a.get("status") in statuses: return a time.sleep(0.4) return a def checkout(seconds=None): try: return call("POST", "/shop/checkout", None, {"seconds": seconds} if seconds else {}, False, port=LIFE_PORT) except OSError: return 0, {} def uptime(): return lc("GET", "/heart/v1/pulse")[1].get("data", {}).get("reconnaissance", {}).get("vitals", {}).get("uptimeSec", -1) def problem(st, b, status, code, has=""): return st == status and b.get("code") == code and has in b.get("detail", "") try: key = e2e.until(lambda: re.search(r"owner key for private views: (\S+)\)", log_path.read_text()), 20, "the lifecycle simulator's owner key").group(1) card, public = lc("GET", "/heart/v1/card")[1]["data"], lc("GET", "/.well-known/heart-card.json", k=False)[1]["data"] check("lifecycle: the Card lists module lifecycle, operations drain and restart, wakeSources local (owner only)", "lifecycle" in card["modules"] and card.get("lifecycle") == {"operations": ["drain", "restart"], "wakeSources": ["local"]} and "lifecycle" in public["modules"] and "lifecycle" not in public, str(card.get("lifecycle"))) st, pub = lc("GET", "/heart/v1/lifecycle", k=False) st2, own = lc("GET", "/heart/v1/lifecycle") check("lifecycle: GET /lifecycle shows the phase to anyone, what is allowed only with the key", st == 200 and pub["data"] == {"phase": "running"} and pub["heartprivacy"] == "public" and st2 == 200 and own["data"]["allowed"] == ["drain", "restart"] and own["data"]["wakeSources"] == ["local"], f"{pub} {own}") refusals = { "without the owner key: 403 insufficient-clearance": problem(*lc("POST", "/heart/v1/lifecycle/restart", {}, k=False), 403, "insufficient-clearance"), "unknown operation: 404": problem(*lc("POST", "/heart/v1/lifecycle/explode", {}), 404, "not-found"), "vendor operation not offered: 404": problem(*lc("POST", "/heart/v1/lifecycle/x-self-test", {}), 404, "not-found"), "standard operation not offered (shutdown): 501": problem(*lc("POST", "/heart/v1/lifecycle/shutdown", {}), 501, "not-implemented"), "standard operation not offered (factory-reset): 501": problem(*lc("POST", "/heart/v1/lifecycle/factory-reset", {}), 501, "not-implemented"), "unknown body field: 400": problem(*lc("POST", "/heart/v1/lifecycle/restart", {"now": True}), 400, "invalid-request"), "delaySec above 0 (no schedules module): 400": problem(*lc("POST", "/heart/v1/lifecycle/restart", {"delaySec": 30}), 400, "invalid-request", "schedules"), "operation in the body not the path's: 400": problem(*lc("POST", "/heart/v1/lifecycle/restart", {"operation": "drain"}), 400, "invalid-request"), "mode on a drain: 400": problem(*lc("POST", "/heart/v1/lifecycle/drain", {"mode": "force"}), 400, "invalid-request"), "timeoutSec 0: 400": problem(*lc("POST", "/heart/v1/lifecycle/restart", {"timeoutSec": 0}), 400, "invalid-request"), "no Idempotency-Key: 400": problem(*lc("POST", "/heart/v1/lifecycle/restart", {}, idem=False), 400, "invalid-request"), } made = [e for e in lc("GET", "/heart/v1/actions")[1]["items"] if "operation" in e["data"]] check("lifecycle: refusals follow API Standard 13.2 rule 3, and none of them made an Action", all(refusals.values()) and not made, str([k for k, v in refusals.items() if not v])) # drain: new orders are refused at once, the running ones finish, and after drainSec orders are taken again sent = [{"drainSec": 8, "reason": "the lifecycle test"}, {"timeoutSec": 60}, {"timeoutSec": 3}, {"mode": "force"}, {}] st0, _ = checkout(3) st, b = lc("POST", "/heart/v1/lifecycle/drain", sent[0]) d = b.get("data", {}) st1, refused = checkout() check("lifecycle: drain answers 202 with a running Action: risk write, cancellable", st0 == 201 and st == 202 and d.get("status") == "running" and d.get("risk") == "write" and d.get("operation") == "POST /heart/v1/lifecycle/drain" and d.get("cancellable") is True, f"{st} {b}") check("lifecycle: while it drains, a new order gets a clear refusal (503, try again in a minute)", st1 == 503 and refused.get("accepted") is False and "no new orders" in refused.get("message", ""), f"{st1} {refused}") a = until_status(d.get("id"), ("succeeded", "failed", "cancelled"), 15) check("lifecycle: the drain finishes the orders already running and succeeds within drainSec", a.get("status") == "succeeded" and a.get("result", {}).get("ordersFinished", 0) >= 1 and a.get("progress") == 1.0 and a.get("cancellable") is False, str(a)) reopened = e2e.until(lambda: checkout()[0] == 201, 15, "orders after the drain") check("lifecycle: after drainSec the shop takes new orders again", reopened) # a graceful restart that is cancelled while it drains checkout(300) # an order that will not finish soon: it holds every graceful restart up0 = uptime() st, b = lc("POST", "/heart/v1/lifecycle/restart", sent[1]) d = b.get("data", {}) st1, _ = checkout() st2, c = lc("POST", f"/heart/v1/actions/{d.get('id')}/decision", {"decision": "cancel"}) a = until_status(d.get("id"), ("cancelled",), 5) check("lifecycle: a graceful restart drains first: running, write, cancellable, with nextReachableAt", st == 202 and d.get("status") == "running" and d.get("risk") == "write" and d.get("cancellable") is True and isinstance(d.get("nextReachableAt"), str) and st1 == 503, f"{st} {b} {st1}") check("lifecycle: cancel while it drains: cancelling, then cancelled at the next safe point; nothing restarted", st2 == 200 and c["data"]["status"] in ("cancelling", "cancelled") and a.get("status") == "cancelled" and "progress" in a and a.get("decision", {}).get("decision") == "cancel" and checkout()[0] == 201 and uptime() > up0, f"{st2} {c} {a}") st3, c3 = lc("POST", f"/heart/v1/actions/{d.get('id')}/decision", {"decision": "cancel"}) check("lifecycle: cancelling an Action that has ended answers 409 conflict", problem(st3, c3, 409, "conflict"), f"{st3} {c3}") # a graceful restart whose order cannot finish by timeoutSec: it fails, and is never forced down = [] st, b = lc("POST", "/heart/v1/lifecycle/restart", sent[2]) a = until_status(b.get("data", {}).get("id"), ("failed", "succeeded"), 12, lambda: down.append(lc("GET", "/heart/v1/pulse")[0] != 200)) check("lifecycle: a graceful restart that passes timeoutSec fails and says it was not forced", st == 202 and a.get("status") == "failed" and "not forced" in a.get("error", "") and a.get("risk") == "write" and "mode" not in a.get("arguments", {}) and "nextReachableAt" in a, str(a)) check("lifecycle: ... and the shop never went away, kept its uptime and takes orders again", down and not any(down) and uptime() > up0 and checkout()[0] == 201, f"{down}") # a force restart: destructive, held for a person; deny, and approve only with the button at the server st, b = lc("POST", "/heart/v1/lifecycle/restart", sent[3]) d = b.get("data", {}) st1, b1 = lc("POST", "/heart/v1/lifecycle/restart", {}) st2, b2 = lc("POST", f"/heart/v1/actions/{d.get('id')}/decision", {"decision": "approve"}) st3, b3 = lc("POST", f"/heart/v1/actions/{d.get('id')}/decision", {"decision": "approve", "presence": { "method": "device-button", "digest": d.get("digest", "0" * 64), "proof": "pressed"}}) check("lifecycle: a force restart is destructive and held awaiting-approval, with digest and expiresAt", st == 202 and d.get("status") == "awaiting-approval" and d.get("risk") == "destructive" and re.fullmatch(r"[0-9a-f]{64}", d.get("digest", "")) and d.get("expiresAt") and "nextReachableAt" in d, str(b)) check("lifecycle: one operation at a time (409 conflict while one waits)", problem(st1, b1, 409, "conflict"), f"{st1} {b1}") check("lifecycle: approving with only the key is refused (403 presence-required); a button proof sent through " "the API is refused (403 presence-invalid)", problem(st2, b2, 403, "presence-required") and problem(st3, b3, 403, "presence-invalid"), f"{st2} {b2} {st3} {b3}") st4, b4 = lc("POST", f"/heart/v1/actions/{d.get('id')}/decision", {"decision": "deny", "note": "not now"}) st5, b5 = lc("POST", f"/heart/v1/actions/{d.get('id')}/decision", {"decision": "approve", "presence": { "method": "webauthn", "digest": d.get("digest", "0" * 64), "proof": "x"}}) check("lifecycle: the owner denies it: denied, and a later approve answers 409 conflict", st4 == 200 and b4["data"]["status"] == "denied" and problem(st5, b5, 409, "conflict"), f"{st4} {b4} {st5}") st, b = lc("POST", "/heart/v1/lifecycle/restart", {"mode": "force"}) d = b.get("data", {}) time.sleep(3) held = action(d.get("id")) life = lc("GET", "/heart/v1/lifecycle")[1].get("data", {}) up1 = uptime() check("lifecycle: nothing runs before approval: still awaiting, the slow order still running, no restart", held.get("status") == "awaiting-approval" and life.get("ext", {}).get("org.heartaispec.demo.orders-running", 0) >= 1 and up1 > up0 and checkout()[0] == 201, f"{held.get('status')} {life} {up1}") (here / "button.press").touch() # a person at the server presses the button down = [] a = until_status(d.get("id"), ("succeeded", "failed", "expired"), 25, lambda: down.append(lc("GET", "/heart/v1/pulse")[0] != 200)) check("lifecycle: the button at the server approves it; the shop restarts at once, cutting off the slow order", a.get("status") == "succeeded" and a.get("decision", {}).get("presenceMethod") == "device-button" and a.get("result", {}).get("ordersCutOff", 0) >= 1 and isinstance(a.get("nextReachableAt"), str) and any(down), str(a)) up2 = uptime() check("lifecycle: after the restart it answers again, with state reloaded (uptime from 0, orders taken)", 0 <= up2 < up1 and checkout()[0] == 201, f"{up1} -> {up2}") # a graceful restart that succeeds: drains, the API is briefly gone, and it comes back down = [] st, b = lc("POST", "/heart/v1/lifecycle/restart", sent[4]) a = until_status(b.get("data", {}).get("id"), ("succeeded", "failed"), 30, lambda: down.append(lc("GET", "/heart/v1/pulse")[0] != 200)) check("lifecycle: a graceful restart with no slow order drains, restarts (API briefly gone) and succeeds", st == 202 and a.get("status") == "succeeded" and a.get("result", {}).get("ordersCutOff") == 0 and any(down) and not down[-1] and isinstance(a.get("nextReachableAt"), str) and uptime() < 60, str(a)) mine = [e["data"] for e in lc("GET", "/heart/v1/actions")[1]["items"] if "operation" in e["data"]] check("lifecycle: every lifecycle Action is kept, by a human, with the outcome it had", sorted(a["status"] for a in mine) == sorted(["succeeded", "cancelled", "failed", "denied", "succeeded", "succeeded"]) and all(a["initiator"]["kind"] == "human" for a in mine), str([a["status"] for a in mine])) check("lifecycle: every request sent is valid against lifecycle-request.schema.json", all(requests.is_valid(r) for r in sent)) check("lifecycle: every reply is valid against the official schemas (envelopes and problems)", not bad, str(bad[:3])) except (AssertionError, OSError, KeyError, ValueError) as e: check("lifecycle: the tests ran to the end", False, f"{e.__class__.__name__}: {e} (log: {log_path})") finally: sim.terminate() sim.wait(timeout=10) def main(): work = pathlib.Path(tempfile.mkdtemp(prefix="heart-online-shop-")) lifecycle_tests(work) 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") t0 = time.time() with open(log_path, "w") as log: sim = subprocess.Popen([sys.executable, str(HERE / "onlineShopSimulator.py"), "--monitor-url", e2e.SITE, "--scenario", "all", "--speed", "5", "--seed", "7", "--duration", str(RUN_SEC), "--port", str(API_PORT), "--state-file", str(work / "online-shop-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": "Online shop demo"}, cookie=owner)[0] check("the owner registers the server 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) # Refused at once, with or without the owner key: destructive things are never possible through HEART. refused = [] for name in ("deleteOrders", "refundOrder", "deployVersion", "changePaymentProvider"): for k in (key, None): st, b = call("POST", f"/heart/v1/capabilities/{name}/invoke", k, {"arguments": {}}) refused.append(st == 403 and b.get("code") == "capability-not-allowed") for method in ("DELETE", "PUT", "PATCH"): st, b = call(method, "/heart/v1/orders", key) refused.append(st == 403 and b.get("code") == "capability-not-allowed") check("destructive requests are refused with 403 capability-not-allowed, even with the owner key", all(refused), str(refused)) st, b = call("POST", "/heart/v1/capabilities/restartWorker/invoke", None, {"arguments": {"worker": 0}}) check("running a Playbook's action needs the owner key", st == 403 and b.get("code") == "insufficient-clearance", f"{st} {b}") time.sleep(15) # past the first staged fault, while the cache is warm idem = str(uuid.uuid4()) st, b = call("POST", "/heart/v1/capabilities/restartWorker/invoke", key, {"arguments": {"worker": 0}}, idem) st2, b2 = call("POST", "/heart/v1/capabilities/restartWorker/invoke", key, {"arguments": {"worker": 0}}, idem) d = b.get("data", {}) check("the owner runs restartWorker by hand: human, not physical, it runs at once", st == 200 and d.get("status") == "succeeded" and d.get("initiator", {}).get("kind") == "human" and d.get("physical") is False and validator.is_valid(b), f"{st} {b}") check("a repeated Idempotency-Key returns the same Action and runs nothing twice", st2 == 200 and b2.get("data", {}).get("id") == d.get("id"), f"{st2}") st, b = call("POST", "/heart/v1/capabilities/restartWorker/invoke", key, {"arguments": {"worker": 0}}, idem=False) check("a change without an Idempotency-Key is refused (400)", st == 400 and b.get("code") == "invalid-request", f"{st}") st, b = call("POST", "/heart/v1/capabilities/restartWorker/invoke", key, {"arguments": {"worker": 99}}) check("an argument outside the capability's range is refused (422 out-of-range)", st == 422, f"{st}") st, b = call("POST", "/heart/v1/capabilities/clearCache/invoke", key, {"arguments": {"percent": 50}}) check("the clearCache interlock still applies to the owner: a useful cache is not cleared (409 guard-failed)", st == 409 and b.get("code") == "guard-failed", f"{st} {b}") invalid, leaks, pages = [], [], [] end = t0 + RUN_SEC - 14 while time.time() < end and sim.poll() is None: try: msgs = ([get("/.well-known/heart-card.json"), get("/heart/v1/card", key), get("/heart/v1/pulse"), get("/heart/v1/pulse", key), get("/heart/v1/tending/autonomy", key)] + get("/heart/v1/arrhythmias")["items"] + get("/heart/v1/arrhythmias", key)["items"] + get("/heart/v1/properties")["items"] + get("/heart/v1/properties", key)["items"] + get("/heart/v1/actions", key)["items"] + get("/heart/v1/tending/rules", key)["items"] + get("/heart/v1/rhythm", 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"), get("/.well-known/heart-card.json")] 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")] if "capabilities" in get("/.well-known/heart-card.json")["data"]: leaks.append("capabilities in the public Card") 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)))) need = ["/heart/v1/actions", "/heart/v1/tending/rules", "/heart/v1/tending/autonomy", "/heart/v1/rhythm", "/heart/v1/sales"] check("actions, Playbooks, history and sales need the owner key", all(call("GET", p)[0] == 403 and call("GET", p, key)[0] == 200 for p in need)) sales = get("/heart/v1/sales", key) check("the private sales view holds totals only, no customer", set(sales) == {"ordersToday", "revenueTodayUsd", "note"}, str(sales)) actions = [e["data"] for e in get("/heart/v1/actions", key)["items"]][::-1] # oldest first rules = {e["data"]["id"]: e["data"] for e in get("/heart/v1/tending/rules", key)["items"]} rule_runs = {c: [a for a in actions if a.get("capability") == c and a["initiator"]["kind"] == "rule" and a["status"] in ("succeeded", "failed")] for c in LIMITS} fired = {pb: [a for a in rule_runs[c] if a["initiator"].get("id") == pb and a["status"] == "succeeded"] for pb, c in PLAYBOOKS.items()} check("all three adopted Playbooks fired as initiator rule, with risk write, not physical", all(fired.values()) and all(a["risk"] == "write" and a["physical"] is False for r in fired.values() for a in r), str({k: len(v) for k, v in fired.items()})) check("every Playbook is the owner's own, adopted, and not adopted as physical", set(rules) == set(PLAYBOOKS) and all(r["origin"] == "owner" and r.get("adoptedAt") and r["adoptedPhysical"] is False for r in rules.values()), str(rules)) within = [] for c, (per_day, cool) in LIMITS.items(): times = [a["ext"][SIM] for a in rule_runs[c]] within.append(len(times) <= per_day and all(b - a >= cool for a, b in zip(times, times[1:]))) check("automatic runs stayed within maxPerDay and cooldownSec", all(within), str({c: [a["ext"][SIM] for a in r] for c, r in rule_runs.items()})) denied = [a for a in actions if a["initiator"]["kind"] == "rule" and a["status"] == "denied"] check("at the daily limit the Playbook is refused with a reason, and does not run", any(a["capability"] == "clearCache" and a["error"].startswith("maxPerDay") for a in denied) and len(rule_runs["clearCache"]) == LIMITS["clearCache"][0], str([(a["capability"], a["error"]) for a in denied])) check("a fix that did not help twice in a row suspends its Playbook, which escalates instead of looping", rules["restart-stuck-worker"]["status"] == "suspended" and any(a["capability"] == "restartWorker" and "suspended" in a["error"] for a in denied) and len(rule_runs["restartWorker"]) <= 3, str([a["ext"] for a in rule_runs["restartWorker"]])) alerts = [e["data"] for e in get("/heart/v1/arrhythmias", key)["items"]] uncovered = [a for a in alerts if a["code"] in ("connectivity.dependency-unreachable", "performance.slow", "performance.overloaded", "software.crash-loop")] check("problems no Playbook covers are escalated, not fixed", uncovered and all(a["status"] in ("escalated", "resolved") and a.get("escalation") and not a["actions"] for a in uncovered), str([(a["code"], a["status"]) for a in uncovered])) check("an Action records why: its alert, its Playbook and its result", all(a.get("arrhythmiaId") and a.get("explanation") and "result" in a for r in fired.values() for a in r)) st, b = call("POST", "/heart/v1/tending/stop", key) check("Stop takes effect at once: max 2, mode observe", st == 200 and b["data"]["max"] == 2 and b["data"]["mode"] == "observe", f"{st} {b}") st, b = call("PUT", "/heart/v1/tending/autonomy", key, {"max": 3}) check("a max this shop does not offer is refused (400 invalid-request)", st == 400 and b.get("code") == "invalid-request", f"{st} {b}") st, b = call("PUT", "/heart/v1/tending/autonomy", key, {"max": 4}) check("turning automatic action back on with only the key is refused (403 presence-required)", st == 403 and b.get("code") == "presence-required", f"{st} {b}") (work / "button.press").touch() st, b = call("PUT", "/heart/v1/tending/autonomy", key, {"max": 4}) check("with the local button pressed, a person turns it back on", st == 200 and b["data"]["max"] == b["data"]["current"] == 4 and b["data"]["mode"] == "act", f"{st} {b}") shown = set(re.findall(r"(?:system|software|storage|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, order, revenue or action detail 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())