"""Runnable check for SAFETY-61: the control task's death.  `python3 test_control_task.py`

If `run_control()` raised, nothing noticed: the task ended, `stop` was never
set, the web UI kept serving and the Supervisor saw a healthy container. The
card's fix and its reasons are in DECISIONS/SAFETY-61-control-task-death.md.

Parts 1-3 drive the REAL `app.main.main()` - the process entry point - in a
child process (this file re-invoked with `--driver`), with only the outside
world faked: options, Home Assistant, the store, the web server and the core
API probe. The child's exit status is the real one. Each child runs under a
deadline, because the defect this card closes is a HANG (MUTATION.md habit 5).

Takes ~15 s. Purge `__pycache__` before a mutation run (MUTATION.md).
"""

import asyncio
import json
import os
import signal
import subprocess
import sys
import time

HERE = os.path.dirname(os.path.abspath(__file__))
FAULT = "injected SAFETY-61 fault"
DEADLINE_S = 40

fails = []


def check(name, cond):
    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
    if not cond:
        fails.append(name)


# --------------------------------------------------------------------------- #
# The child: the real main() with the outside world faked.
# --------------------------------------------------------------------------- #
BASE = {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
        "soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
        "meter_source": "off", "max_w": 2000, "gain": 0.6, "slew_w": 0,
        "deadband_w": 15, "target_grid_w": -10, "step_w": 10,
        "heartbeat_s": 10, "stale_input_s": 15, "cycle_floor_s": 1,
        "auto_start": True}


def driver(scenario):
    import aiohttp
    import app.main as M
    import app.store
    import app.web

    def event(*e):
        """Streamed as it happens, so a child killed at the deadline or by
        os._exit() still reports what it did."""
        print("EV " + json.dumps(list(e)), flush=True)
    t0 = time.monotonic()

    class FakeHass:
        def __init__(self):
            self.batt = 0.0
            self.limit_calls = 0

        async def limits(self, entity):
            self.limit_calls += 1
            # "refuse": ESPHome not up at the startup gate, then a rating
            # below max_w once run_control() re-reads it -> its refusal path.
            if scenario == "refuse":
                return (None, None, None) if self.limit_calls == 1 else (-1000.0, 1000.0, 1.0)
            return (-5000.0, 5000.0, 1.0)

        async def number(self, entity, invert=False):
            if entity == "sensor.grid":
                # a steady 600 W import, alternating by 1 W so every pass cycles
                return 600.0 + (int(time.monotonic() - t0) % 2)
            if entity == "sensor.batt":
                return self.batt
            if entity == "sensor.soc":
                return 50.0
            return None

        async def state(self, entity):
            return None

        async def set_number(self, entity, value):
            event("w", float(value))
            self.batt = float(value)
            return True

        async def call(self, *a, **k):
            return True

        async def mqtt_service(self):
            raise RuntimeError("no broker in this test")

    class _Store:
        data = {}

        def get(self, key, default=None):
            return default

        def set(self, *a, **k):
            pass

    class _Runner:
        async def cleanup(self):
            event("cleanup")

    async def _web_start(controller, port=8099, selfcfg=None):
        return _Runner()

    class _NoProbeSession:
        """The core API probe in amain() must not touch a network."""
        async def __aenter__(self):
            return self

        async def __aexit__(self, *a):
            return False

        def get(self, *a, **k):
            raise OSError("no network in this test")

    M.load_options = lambda: dict(BASE)
    M.HomeAssistant = lambda session: FakeHass()
    app.store.Store = _Store
    app.web.start = _web_start
    aiohttp.ClientSession = _NoProbeSession

    real_cycle = M.Controller.cycle
    n = {"cycles": 0}

    async def cycle(self):
        n["cycles"] += 1
        if scenario == "fault" and n["cycles"] == 4:
            event("fault")
            raise RuntimeError(FAULT)
        await real_cycle(self)
        if scenario == "clean" and n["cycles"] == 4:
            os.kill(os.getpid(), signal.SIGTERM)     # the Supervisor's stop

    M.Controller.cycle = cycle
    M.main()


def run_child(scenario):
    env = dict(os.environ, PYTHONDONTWRITEBYTECODE="1")
    try:
        p = subprocess.run([sys.executable, "-B", __file__, "--driver", scenario],
                           cwd=HERE, env=env, capture_output=True, text=True,
                           timeout=DEADLINE_S)
    except subprocess.TimeoutExpired as e:
        rc, out, err = None, e.stdout or "", e.stderr or ""
    else:
        rc, out, err = p.returncode, p.stdout, p.stderr
    out = out.decode() if isinstance(out, bytes) else out
    err = err.decode() if isinstance(err, bytes) else err
    events = [json.loads(ln[3:]) for ln in out.splitlines() if ln.startswith("EV ")]
    return rc, out, err, events


def tail(text, n=15):
    return "\n".join("      | " + ln for ln in text.splitlines()[-n:])


if len(sys.argv) == 3 and sys.argv[1] == "--driver":
    driver(sys.argv[2])
    sys.exit(0)            # unreachable in every scenario: main() exits itself


# --------------------------------------------------------------------------- #
print("1. an exception inside cycle() - the real main(), in a child process")
rc, out, err, ev = run_child("fault")
print(f"    exit status {rc}; events {ev}")
check(f"the process ENDS inside {DEADLINE_S} s rather than sitting there "
      "healthy (the defect is a hang)", rc is not None)
check("exit status is 70 (EX_SOFTWARE), not 0 and not an unhandled crash's 1",
      rc == 70)
fault_at = ev.index(["fault"]) if ["fault"] in ev else None
before = [e[1] for e in ev[:fault_at] if e[0] == "w"] if fault_at is not None else []
after = [e[1] for e in ev[fault_at + 1:] if e[0] == "w"] if fault_at is not None else []
# Precondition (habit 23): the inverter was holding a NON-ZERO command when the
# fault hit, so a 0 W write afterwards is evidence of the wind-down, not rest.
check("precondition: the fault hit while the last command was non-zero "
      f"(last before fault = {before[-1] if before else None})",
      fault_at is not None and bool(before) and before[-1] != 0.0)
check(f"the wind-down ran: 0 W written AFTER the fault (writes after = {after})",
      after == [0.0])
check("the web runner was cleaned up, i.e. the ordinary shutdown path ran",
      ["cleanup"] in ev[(fault_at or 0):])
died_line = next((i for i, ln in enumerate(err.splitlines())
                  if "Control stopped unexpectedly (RuntimeError('" + FAULT + "'))" in ln), None)
wind_line = next((i for i, ln in enumerate(err.splitlines())
                  if "shutting down - commanding 0 W" in ln), None)
check("a CRITICAL line names the exception and the exit status",
      died_line is not None
      and "CRITICAL" in err.splitlines()[died_line]
      and "set to 0 W and the add-on exits with code 70" in err.splitlines()[died_line])
check("the traceback is logged, down to the frame that raised",
      "Traceback (most recent call last)" in err
      and f"RuntimeError: {FAULT}" in err and "in cycle" in err)
check("it is logged AT THE DEATH, before the wind-down - not only at exit",
      died_line is not None and wind_line is not None and died_line < wind_line)
check("exactly one traceback: main() does not print a second one",
      err.count("Traceback (most recent call last)") == 1)
if fails:
    print(tail(err, 30))

# --------------------------------------------------------------------------- #
print("2. the same entry point stopped by SIGTERM - the callback must stay quiet")
rc2, out2, err2, ev2 = run_child("clean")
print(f"    exit status {rc2}; events {ev2}")
check("an ordinary stop still exits 0", rc2 == 0)
check("an ordinary stop logs no unexpected stop", "Control stopped unexpectedly" not in err2)
check("an ordinary stop still winds down to 0 W, from a non-zero command",
      any(e[0] == "w" and e[1] != 0.0 for e in ev2)
      and [e for e in ev2 if e[0] == "w"][-1:] == [["w", 0.0]])

# --------------------------------------------------------------------------- #
print("3. run_control()'s own refusal path returns on purpose - still exit 78")
rc3, out3, err3, ev3 = run_child("refuse")
print(f"    exit status {rc3}")
check("a late-arriving device rating refusal exits 78 (EX_CONFIG), not 70",
      rc3 == 78)
check("and is not reported as a death", "Control stopped unexpectedly" not in err3)
if rc3 != 78:
    print(tail(err3))

# --------------------------------------------------------------------------- #
print("4. hass.py: a body that is not a state object is an ABSENCE")
import aiohttp                                      # noqa: E402
from aiohttp import web as aioweb                   # noqa: E402

import app.hass as H                                # noqa: E402


async def hass_cases():
    async def handler(req):
        k = req.match_info["k"]
        if k == "sensor.good":
            return aioweb.json_response({"entity_id": k, "state": "42"})
        if k == "sensor.badjson":
            return aioweb.Response(text="{not json", content_type="application/json")
        if k == "sensor.list":
            return aioweb.Response(text="[1, 2]", content_type="application/json")
        if k == "sensor.html":
            return aioweb.Response(text="<html>", content_type="text/html")
        return aioweb.Response(status=404)

    async def svc(req):
        return aioweb.Response(body=b"\xff\xfe\xfa", status=400,
                               headers={"Content-Type": "text/plain; charset=utf-8"})

    app_ = aioweb.Application()
    app_.router.add_get("/states/{k}", handler)
    app_.router.add_post("/services/{d}/{s}", svc)
    runner = aioweb.AppRunner(app_)
    await runner.setup()
    site = aioweb.TCPSite(runner, "127.0.0.1", 0)
    await site.start()
    port = site._server.sockets[0].getsockname()[1]
    H.CORE_API = f"http://127.0.0.1:{port}"
    got = {}
    try:
        async with aiohttp.ClientSession() as session:
            h = H.HomeAssistant(session, token="t")
            for e in ("sensor.good", "sensor.badjson", "sensor.list", "sensor.html"):
                try:
                    got[e] = ("ok", await h.number(e))
                except Exception as err:                    # noqa: BLE001
                    got[e] = ("raised", type(err).__name__)
            try:
                got["call"] = ("ok", await h.set_number("number.sp", 0))
            except Exception as err:                        # noqa: BLE001
                got["call"] = ("raised", type(err).__name__)
    finally:
        await runner.cleanup()
    return got

got = asyncio.run(hass_cases())
print(f"    {got}")
check("baseline: a well-formed state reads as its value (42.0)",
      got["sensor.good"] == ("ok", 42.0))
check("a malformed JSON body labelled application/json reads as None, not a raise",
      got["sensor.badjson"] == ("ok", None))
check("a JSON body that is not an object reads as None, not a raise",
      got["sensor.list"] == ("ok", None))
check("a text/html 200 reads as None (already true before SAFETY-61)",
      got["sensor.html"] == ("ok", None))
check("a rejected write with an undecodable body is a failed write (False)",
      got["call"] == ("ok", False))

# --------------------------------------------------------------------------- #
print("5. the control-loop status row")
from app.main import Controller                     # noqa: E402


class _Pub:
    enabled = False

    def publish(self, *a, **k):
        pass

    def close(self):
        pass


class _Store:
    data = {}

    def set(self, *a, **k):
        pass


class _RowHass:
    async def limits(self, entity):
        return (-5000.0, 5000.0, 1.0)

    async def number(self, entity, invert=False):
        return {"sensor.grid": 600.0, "sensor.soc": 50.0}.get(entity, 0.0)

    async def state(self, entity):
        return None

    async def set_number(self, entity, value):
        return True

    async def call(self, *a, **k):
        return True


def rows_for(pass_age):
    c = Controller(dict(BASE), _RowHass(), _Store(), _Pub(), stop=asyncio.Event())
    c.loop_pass_mono = None if pass_age is None else time.monotonic() - pass_age
    return c.checks()


r_new, r_29, r_31 = rows_for(None), rows_for(29), rows_for(31)
print(f"    {r_new[0]} | {r_29[0]} | {r_31[0]}")
# UX-08: the wording is app/messages.py's; these pin the meaning - which row,
# its colour, and the two numbers (age, veto) an operator acts on.
check("before the first pass the row is ok and says 'starting'",
      r_new[0]["id"] == "controller" and r_new[0]["ok"] and not r_new[0]["warn"]
      and "starting" in r_new[0]["text"])
check("29 s after a pass it is ok",
      r_29[0]["id"] == "controller" and r_29[0]["ok"] and not r_29[0]["warn"]
      and "29 seconds ago" in r_29[0]["text"])
check("31 s after a pass it is RED and first, naming the age and the veto",
      r_31[0]["id"] == "controller" and not r_31[0]["ok"]
      and not r_31[0]["warn"]
      and "not run for 31 seconds" in r_31[0]["text"]
      and "0 W after 30 seconds" in r_31[0]["text"])


async def one_pass():
    c = Controller(dict(BASE), _RowHass(), _Store(), _Pub(), stop=asyncio.Event())
    c.auto = True
    before = getattr(c, "loop_pass_mono", "absent")
    t = asyncio.create_task(c.run_control())
    await asyncio.sleep(2.5)
    c.stopping = True
    await asyncio.wait_for(t, timeout=10)
    return before, getattr(c, "loop_pass_mono", None)

_before, _after = asyncio.run(one_pass())
print(f"    stamp before run: {_before}; age after 2.5 s: "
      f"{None if _after is None else round(time.monotonic() - _after, 2)} s")
check("the real run_control() stamps each pass",
      _before is None and _after is not None and time.monotonic() - _after < 3)

print()
if fails:
    print(f"FAILED: {len(fails)}")
    for f in fails:
        print(f"  - {f}")
    sys.exit(1)
print("all SAFETY-61 checks passed")