#!/usr/bin/env python3
# SPDX-License-Identifier: Apache-2.0
"""End to end: the thermostat simulator reports to a HEART Monitor site running on this machine, and its gate holds.

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, turns approvals on, and checks for about five 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;
  * guards refuse what they must: out-of-range setpoints (422), read-only limits (409), a repeat too soon (409);
  * a setpoint change waits for the thermostat's button: an approval with only the key is refused 403
    presence-required, the change stays waiting, and runs after `touch button.press`;
  * the mock AI's suggestion never runs without a yes: refused with the key, then it expires and nothing changes;
    its next suggestion runs only after the owner approves it on the monitor site (signed, remote-site);
  * Stop cancels a waiting change at once, and resuming needs the button;
  * every staged fault shows on the owner's page with its standard fault code;
  * no temperature, occupancy or schedule reaches the site, and the page is private until shared.

Needs: Node 20+, Python 3 with jsonschema and referencing (pip install "jsonschema[format]>=4.18" referencing).
Usage, from the project's top folder:  python3 thermostatSimulator/test_e2e.py
"""
import json
import os
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 = 8802, 43302, 300
e2e.SITE_PORT, e2e.SITE = SITE_PORT, f"http://127.0.0.1:{SITE_PORT}"
SYS = "org.heartaispec.demo.thermostat"
STAGED = ["performance.overshoot", "connectivity.link-lost", "power.battery-low", "sensors.stuck", "sensors.out-of-range",
          "system.main-function-lost", "actuators.short-cycling"]
NEVER_ON_SITE = ["Cel", "room_temperature", "flow_temperature", "occupied", "people_home", "window", "23:00",
                 "night_hours", "empty", "MOCK AI"]

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, served, pages = [], [], []


def check(name, ok, detail=""):
    results.append(bool(ok))
    print(f"{'PASS' if ok else 'FAIL'}  {name}{'' if ok else ': ' + str(detail)}", flush=True)


def call(method, path, body=None, key=None):
    """Returns (status, JSON). Every 2xx message is kept, to be checked against the schemas."""
    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:{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:
            out = json.loads(r.read())
            served.extend(out["items"] if isinstance(out, dict) and "items" in out else [out])
            return r.status, out
    except urllib.error.HTTPError as e:
        return e.code, json.loads(e.read() or b"{}")


def get(path, key=None):
    return call("GET", path, key=key)[1]


def main():
    work = pathlib.Path(tempfile.mkdtemp(prefix="heart-thermostat-"))
    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 / "thermostatSimulator.py"), "--monitor-url", e2e.SITE,
                                    "--scenario", "all", "--speed", "5", "--seed", "7", "--duration", str(RUN_SEC),
                                    "--port", str(API_PORT), "--state-file", str(work / "thermostat-monitor.json"),
                                    "--approval-timeout", "60", "--monitor-approvals", "--monitor-approvals-physical"],
                                   stdout=log, stderr=log, cwd=work)
        started = 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": "Thermostat demo"}, cookie=owner)[0]
        check("the owner registers the device with its claim code", s in (200, 303), f"status {s}")
        s = e2e.http("POST", f"/device/{dev}/approvals", {"mode": "on"}, cookie=owner)[0]
        check("the owner lets the device ask for approvals on the site", s == 200, f"status {s}")
        mode = oct((work / "thermostat-monitor.json").stat().st_mode & 0o777)
        check("the identity file is readable only by its owner", mode == "0o600", mode)
        key = e2e.until(lambda: re.search(r"owner key for private views: (\S+)\)", text()), 10, "owner key").group(1)
        leaks, last_poll = [], [0.0]

        def poll():
            """Every 10 s: read every view (kept for the schema check), look for leaks, and keep the owner's page."""
            if time.time() - last_poll[0] < 10 or sim.poll() is not None:
                return
            last_poll[0] = time.time()
            try:
                for path in ("/.well-known/heart-card.json", "/heart/v1/pulse", "/heart/v1/arrhythmias", "/heart/v1/properties",
                             "/heart/v1/tending/autonomy"):
                    get(path)
                    get(path, key)
                get("/heart/v1/actions", key)
                get("/heart/v1/rhythm", key)
                public = get("/heart/v1/properties")["items"] + get("/heart/v1/arrhythmias")["items"] + [get("/heart/v1/pulse")]
            except OSError:
                return
            leaks.extend(e["type"] for e in public if e["heartprivacy"] != "public")
            leaks.extend(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])

        def wait(cond, secs):
            end = time.time() + secs
            while time.time() < end:
                v = cond()
                if v:
                    return v
                poll()
                time.sleep(0.5)
            return None

        def action(aid):
            return get(f"/heart/v1/actions/{aid}", key)["data"]

        def prop(name):
            return get(f"/heart/v1/properties/{SYS}.{name}", key)["data"]["value"]

        # ---- guards and refusals
        sp = f"/heart/v1/properties/{SYS}.setpoint"
        check("a setpoint above 28 °C or below 10 °C is refused 422 out-of-range",
              all(call("PUT", sp, {"value": v}, key)[0] == 422 for v in (30, 8, 28.5)))
        check("an unknown mode is refused 422", call("PUT", f"/heart/v1/properties/{SYS}.mode", {"value": "cool"}, key)[0] == 422)
        check("the firmware's limits are read-only (409 conflict)",
              all(call("PUT", f"/heart/v1/properties/{SYS}.{n}", {"value": 3}, key)[0] == 409
                  for n in ("frost_protection", "max_setpoint", "boiler_firing")))
        check("a change without the owner key is refused 401", call("PUT", sp, {"value": 21})[0] == 401)
        check("private views need the owner key", all(call("GET", p)[0] == 401 for p in ("/heart/v1/actions", "/heart/v1/rhythm"))
              and call("GET", f"/heart/v1/properties/{SYS}.occupied")[0] == 404)

        # ---- a person's own physical change waits for the button
        s, a = call("PUT", sp, {"value": 21}, key)
        a = a.get("data", {})
        check("a setpoint change answers 202 with an Action awaiting approval, physical",
              s == 202 and a.get("status") == "awaiting-approval" and a.get("physical") is True
              and a.get("initiator", {}).get("kind") == "human" and "expiresAt" in a, f"{s} {a}")
        aid = a.get("id", "x")
        s, p = call("POST", f"/heart/v1/actions/{aid}/decision", {"decision": "approve"}, key)
        check("approving it over the network with only the key is refused 403 presence-required",
              s == 403 and p.get("code") == "presence-required", f"{s} {p}")
        wait(lambda: False, 6)
        check("without the button it stays waiting, and the setpoint is unchanged",
              action(aid)["status"] == "awaiting-approval" and prop("setpoint") == 20)
        (work / "button.press").write_text("")
        done = wait(lambda: action(aid)["status"] == "succeeded", 15)
        a = action(aid)
        check("after a press of the thermostat's button it runs, recorded as device-button",
              done and a.get("decision", {}).get("presenceMethod") == "device-button" and prop("setpoint") == 21, a)
        s, p = call("PUT", sp, {"value": 22}, key)
        check("a second change too soon is refused 409 guard-failed", s == 409 and p.get("code") == "guard-failed", f"{s} {p}")
        s, a = call("PUT", f"/heart/v1/properties/{SYS}.mode", {"value": "eco"}, key)
        eco = a.get("data", {}).get("id", "x")
        s2, a2 = call("POST", f"/heart/v1/actions/{eco}/decision", {"decision": "deny"}, key)
        check("a waiting mode change can be denied without the button, and nothing changes",
              s == 202 and s2 == 200 and a2["data"]["status"] == "denied" and prop("mode") == "heat", f"{s} {s2}")

        # ---- the mock AI suggests; it never runs without a yes
        def ai_waiting(other=None):
            for x in get("/heart/v1/actions", key)["items"]:
                d = x["data"]
                if d["initiator"]["kind"] == "ai" and d["status"] == "awaiting-approval" and d["id"] != other:
                    return d
            return None
        ai1 = wait(ai_waiting, 90)
        check("the mock AI suggests a lower night temperature as an ai Action, physical and destructive",
              ai1 and ai1["risk"] == "destructive" and ai1["physical"] and ai1["initiator"].get("onBehalfOf") == "owner"
              and "MOCK AI" in ai1.get("explanation", ""), ai1)
        ai1 = ai1 or {"id": "x"}
        s, p = call("POST", f"/heart/v1/actions/{ai1['id']}/decision", {"decision": "approve"}, key)
        check("approving the AI's suggestion with only the key is refused 403 presence-required", s == 403, s)
        gone = wait(lambda: action(ai1["id"])["status"] == "expired", 90)
        check("unanswered, the AI's suggestion expires and the night temperature is unchanged",
              gone and prop("night_setpoint") is None, action(ai1["id"]) if ai1["id"] != "x" else "no suggestion")

        # ---- its next suggestion is approved by the owner on the monitor site (remote-site, physical opt-in)
        ai2 = wait(lambda: ai_waiting(ai1["id"]), 90) or {"id": "x"}
        shown = wait(lambda: f'value="{ai2["id"]}"><button>Approve' in e2e.http("GET", f"/device/{dev}", cookie=owner)[2], 60)
        page = e2e.http("GET", f"/device/{dev}", cookie=owner)[2]
        check("the suggestion appears on the owner's page, as a title and a change only",
              shown and "Night temperature" in page and "18 °C" in page, ai2)
        e2e.http("POST", f"/device/{dev}/approve", {"action": ai2["id"]}, cookie=owner)
        mails = [m_ for m_ in json.loads(e2e.http("GET", "/__test/outbox")[2]) if "approve a change" in m_["body"]]
        mcode = re.search(r"^ {4}(\d{6})$", mails[-1]["body"], re.M).group(1) if mails else "000000"
        s, _, body = e2e.http("POST", f"/device/{dev}/approve/verify", {"action": ai2["id"], "code": mcode}, cookie=owner)
        check("the owner approves it on the site with an emailed code", "Approved:" in body, body[:300])
        ran = wait(lambda: action(ai2["id"])["status"] == "succeeded", 60)
        a = action(ai2["id"]) if ai2["id"] != "x" else {}
        check("the device checks the site's signature and runs it, recorded as remote-site",
              ran and a.get("decision", {}).get("presenceMethod") == "remote-site" and prop("night_setpoint") == 18, a)

        # ---- Stop
        s, a = call("PUT", sp, {"value": 20.5}, key)
        waiting = a.get("data", {}).get("id", "x")
        s2, c = call("POST", "/heart/v1/tending/stop", None, key)
        check("Stop works at once and cancels a waiting change",
              s == 202 and s2 == 200 and c["data"]["max"] == 2 and c["data"]["mode"] == "observe" and action(waiting)["status"] == "cancelled",
              f"{s} {s2} {action(waiting)['status'] if waiting != 'x' else ''}")
        s, a = call("PUT", "/heart/v1/tending/autonomy", {"max": 5}, key)
        check("a max above what it promises is refused: it never acts on its own (400)",
              s == 400 and a.get("code") == "invalid-request", f"{s} {a}")
        s, a = call("PUT", "/heart/v1/tending/autonomy", {"max": 3}, key)
        resume = a.get("data", {}).get("id", "x")
        check("resuming after Stop waits for a person (destructive)",
              s == 202 and a["data"]["risk"] == "destructive" and get("/heart/v1/tending/autonomy", key)["data"]["mode"] == "observe")
        (work / "button.press").write_text("")
        check("after the button, the HEART suggests again",
              wait(lambda: get("/heart/v1/tending/autonomy")["data"]["mode"] == "suggest", 15) and action(resume)["status"] == "succeeded")

        # ---- watch until the end
        wait(lambda: False, RUN_SEC - 10 - (time.time() - started))
        invalid = []
        for e in served:
            errs = [x.message[:160] for x in validator.iter_errors(e)]
            if errs:
                invalid.append((e.get("type"), errs[0]))
        check(f"every served message is valid against the official schemas ({len(served)} checked)", served and not invalid,
              str(invalid[:3]))
        check("public views carry only public data", not leaks, str(sorted(set(leaks))))
        found = set(re.findall(r"(?:system|sensors|power|connectivity|performance|actuators)\.[a-z-]+", " ".join(pages)))
        missing = [c for c in STAGED if c not in found]
        check("every staged fault shows on the owner's page with its fault code", not missing, f"missing {missing}")
        leaked = [w for w in NEVER_ON_SITE if any(w in p for p in pages)]
        check("no temperature, occupancy or schedule reaches the site", not leaked, f"found {leaked}")
        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__":
    os.environ.setdefault("PYTHONDONTWRITEBYTECODE", "1")
    sys.exit(main())