"""SAFETY-112: waiting for Home Assistant to finish starting.  `python3 -B test_safety112.py`

The Supervisor answers 502 to every add-on read, write and websocket until
Home Assistant is RUNNING (DECISIONS/SAFETY-112-waiting-for-home-assistant.md),
so the add-on cannot control then. This file grades what it says meanwhile:

1. The log: one WARNING when Home Assistant stops answering, one INFO line
   every 5 minutes while it stays down, one INFO line with the wait when it
   answers again. Only the API channel counts as "waiting": a 502 or a
   refused connection turns it on, the next good read off, a timeout never.
2. The status page, N=1 and N=2: the waiting row is first and red, and is the
   banner. Home says "Waiting for Home Assistant" in English and Dutch, not
   "Needs you now", unless a safety stop outranks it. The page script has it.
3. Control is unchanged: the same readings give the same writes with the
   waiting flag on or off, with 502 readings and with live ones.

Mutating a module this file grades? Read MUTATION.md first (purge __pycache__).
"""

import asyncio
import logging
import os
import re
from types import SimpleNamespace

from app import hass as _hass
from app import messages
from app.main import Controller

fails = []


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


class _Rec(logging.Handler):
    def __init__(self):
        super().__init__(logging.DEBUG)
        self.records = []

    def emit(self, record):
        self.records.append(record)


def captured(fn):
    rec, lg = _Rec(), logging.getLogger("goodwe.hass")
    old = lg.level
    lg.addHandler(rec)
    lg.setLevel(logging.DEBUG)
    try:
        fn()
    finally:
        lg.removeHandler(rec)
        lg.setLevel(old)
    return [(r.levelno, r.getMessage()) for r in rec.records
            if r.levelno >= logging.INFO]


class Clock:
    """`hass.time` stand-in: only `monotonic` is read by Outage."""

    def __init__(self):
        self.t = 1000.0

    def monotonic(self):
        return self.t


def with_clock(fn):
    clock, real = Clock(), _hass.time
    _hass.time = SimpleNamespace(monotonic=clock.monotonic)
    try:
        return fn(clock)
    finally:
        _hass.time = real


# --------------------------------------------------------------------------- #
print("1. the log while Home Assistant does not answer")
W, I = logging.WARNING, logging.INFO


def outage(clock):
    o = _hass.Outage()
    o.down("api", "error 502", "read sensor.x: 502")
    for t in (1, 60, 299):
        clock.t = 1000 + t
        o.down("api", "error 502", "read sensor.x: 502")
    clock.t = 1300
    o.down("api", "error 502", "")          # 5 minutes: the first reminder
    clock.t = 1301
    o.down("websocket", "the meter connection closed", "")
    clock.t = 1599
    o.down("api", "error 502", "")          # 299 s after the reminder: none
    clock.t = 1600
    o.down("api", "error 502", "")          # 10 minutes: the second
    clock.t = 1640
    o.up("api")
    o.up("websocket")                       # 640 s in all
    clock.t = 2000
    o.down("api", "no connection", "")      # a new outage starts afresh
    clock.t = 2030
    o.up("api")


lines = with_clock(lambda c: captured(lambda: outage(c)))
check(f"one WARNING per outage, in plain words ({lines})",
      [m for lv, m in lines if lv == W] == [
          "Home Assistant is not answering (error 502), perhaps it is still "
          "starting. Control waits and resumes by itself.",
          "Home Assistant is not answering (no connection), perhaps it is "
          "still starting. Control waits and resumes by itself."])
check("one 'still waiting' line per 5 minutes down, with the time so far",
      [m for lv, m in lines if lv == I and m.startswith("Still")] == [
          "Still waiting for Home Assistant to finish starting, 5 minutes "
          "so far.",
          "Still waiting for Home Assistant to finish starting, 10 minutes "
          "so far."])
check("one INFO line when the last channel is back, with the wait, and a new "
      "outage counts from its own start",
      [m for lv, m in lines if lv == I and m.startswith("Home")] == [
          "Home Assistant answers again after 11 minutes.",
          "Home Assistant answers again after 30 seconds."])


def channels(_clock):
    o = _hass.Outage()
    seen = [o.waiting()]
    o.down("websocket", "the meter connection closed", "")
    seen.append(o.waiting())
    o.down("api", "error 502", "")
    seen.append(o.waiting())
    o.up("websocket")
    seen.append(o.waiting())
    o.up("api")
    seen.append(o.waiting())
    return seen


check("waiting() follows the API channel only, not the meter websocket",
      with_clock(channels) == [False, False, True, True, False])


# The flag as `HomeAssistant.state()` sets it: a 502 or a refused connection
# turns it on, the next good read turns it off, and a timeout (a slow Home
# Assistant, not a starting one) leaves it alone.
class _Answer:
    def __init__(self, what):
        self.what = what

    async def __aenter__(self):
        if isinstance(self.what, BaseException):
            raise self.what
        return self

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

    @property
    def status(self):
        return self.what

    def raise_for_status(self):
        pass

    async def json(self):
        return {"state": "1"}


class _Session:
    def __init__(self, answers):
        self.answers = list(answers)

    def get(self, *_a, **_k):
        return _Answer(self.answers.pop(0))


def reads(answers):
    async def go():
        real, _hass.OUTAGE = _hass.OUTAGE, _hass.Outage()
        lg = logging.getLogger("goodwe.hass")
        lg.disabled = True
        try:
            h = _hass.HomeAssistant(_Session(answers), token="t")
            seen = []
            for _ in answers:
                await h.state("sensor.x")
                seen.append(_hass.OUTAGE.waiting())
            return seen
        finally:
            _hass.OUTAGE = real
            lg.disabled = False
    return asyncio.run(go())


check("a 502 read turns waiting on and the next good read turns it off",
      reads([502, 200, 503, 404]) == [True, False, True, False])
check("a refused connection is waiting too",
      reads([_hass.aiohttp.ClientConnectionError(), 200]) == [True, False])
check("a read timeout is not waiting",
      reads([TimeoutError(), TimeoutError()]) == [False, False])


# --------------------------------------------------------------------------- #
print("2. the status page, N=1 and N=2")


class _Hass:
    session = None
    token = ""

    def __init__(self):
        self.writes = []

    async def limits(self, _e):
        return (-5000.0, 5000.0, 1.0)

    async def number(self, _e, invert=False):
        return None                     # what a 502 read gives

    async def state(self, _e):
        return None

    async def set_number(self, e, v):
        self.writes.append((e, float(v)))
        return False                    # what a 502 write gives

    async def call(self, *_a):
        return False

    async def states(self):
        return None


class _Store:
    def __init__(self):
        self.data = {}

    def set(self, k, v):
        self.data[k] = v

    def get(self, k, d=None):
        return self.data.get(k, d)

    def set_time(self, k, v):
        self.data[k] = v.isoformat()

    def get_time(self, k):
        return None


class _Pub:
    enabled = False

    def publish(self, *_a, **_k):
        pass

    def close(self):
        pass


ONE = {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
       "soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
       "meter_source": "off", "max_w": 5000, "shadow_mode": "off",
       "maintenance_enabled": False, "stale_input_s": 15}
TWO = dict(ONE, setpoint_entity="", soc_entity="", batt_entity="", inverters=[
    {"name": "inverter-a", "setpoint_entity": "number.a", "soc_entity":
     "sensor.soc_a", "batt_entity": "sensor.batt_a", "max_w": 2500,
     "bms_soc_floor": 10},
    {"name": "inverter-b", "setpoint_entity": "number.b", "soc_entity":
     "sensor.soc_b", "batt_entity": "sensor.batt_b", "max_w": 2500,
     "bms_soc_floor": 10}])


def make(opts, hass=None):
    c = Controller(dict(opts), hass or _Hass(), _Store(), _Pub(),
                   stop=asyncio.Event())
    c.auto = True
    for inv in c.inverters:
        inv.dev_min, inv.dev_max = -5000.0, 5000.0
    return c


def waiting_on():
    _hass.OUTAGE.down_on = {"api"}
    _hass.OUTAGE.since = _hass.OUTAGE._told = _hass.time.monotonic()


def waiting_off():
    _hass.OUTAGE.down_on = set()


BANNER = "Not ready — Waiting for Home Assistant to finish starting"
for label, opts in (("N=1", ONE), ("N=2", TWO)):
    c = make(opts)
    waiting_off()
    plain = c.checks()
    pay_off = c.status_payload()
    waiting_on()
    rows = c.checks()
    pay = c.status_payload()
    pay_nl = c.status_payload(lang="nl")
    waiting_off()
    check(f"{label}: no waiting row while Home Assistant answers",
          not [r for r in plain if r.get("id") == "home_assistant"]
          and pay_off["state"]["kind"] != "waiting")
    check(f"{label}: while it does not, the waiting row is first and red",
          rows[0] == {"ok": False, "warn": False, "id": "home_assistant",
                      "banner": messages.banner_ha_waiting(),
                      "text": messages.row_ha_waiting()}
          and rows[1:] == plain)
    check(f"{label}: ...and it is the banner ({pay['banner']!r})",
          pay["banner"] == BANNER and pay["level"] == "bad")
    check(f"{label}: Home says waiting, not 'Needs you now' "
          f"({pay['state']['pill']!r}, {pay['state']['sub']!r})",
          pay["state"]["kind"] == "waiting"
          and pay["state"]["pill"] == "Waiting for Home Assistant"
          and pay["state"]["sub"] == "Control resumes by itself when Home "
                                     "Assistant answers.")
    check(f"{label}: ...and in Dutch ({pay_nl['state']['pill']!r})",
          pay_nl["state"]["pill"] == "Wacht op Home Assistant"
          and pay_nl["state"]["sub"] == "De regeling gaat vanzelf verder "
                                        "zodra Home Assistant antwoordt.")
    check(f"{label}: Needs attention lists it first, with its own title",
          pay["attention"][0]["id"] == "home_assistant"
          and pay["attention"][0]["title"] == "Home Assistant is not answering"
          and pay_nl["attention"][0]["title"] == "Home Assistant antwoordt "
                                                  "niet")

# A safety stop outranks the wait: it still needs a person.
c = make(TWO)
c.inverters[1].sign_wd.tripped = True
waiting_on()
pay = c.status_payload()
waiting_off()
check(f"a safety stop keeps its banner and 'Needs you now' over the wait "
      f"({pay['banner']!r}, {pay['state']['kind']!r})",
      pay["banner"] == messages.banner_direction(True)
      and pay["state"]["kind"] == "alarm")
# A stalled control loop does not resume by itself, so it stays the headline.
c = make(ONE)
c.loop_pass_mono = 0.0
waiting_on()
rows, pay = c.checks(), c.status_payload()
waiting_off()
check(f"a stalled control loop stays first and keeps 'Needs you now' "
      f"({pay['banner']!r})",
      [r["id"] for r in rows[:2]] == ["controller", "home_assistant"]
      and pay["state"]["kind"] == "alarm")
c = make(dict(ONE, shadow_mode="on"))
waiting_on()
pay = c.status_payload()
waiting_off()
check("shadow mode keeps its own state over the wait",
      pay["state"]["kind"] == "shadow")


# The page script, read as text (the image has no JS engine): an amber dot,
# and the batteries tab carries the banner like the other held states.
JS = open(os.path.join(os.path.dirname(os.path.abspath(__file__)), "app",
                       "static", "app.js"), encoding="utf-8").read()
check("the page script gives 'waiting' an amber dot",
      re.search(r'alarm: " bad", waiting: " warn" \}\[st\.kind\]', JS)
      is not None)
check("...and shows it as the batteries-tab banner",
      '["shadow", "stopped", "alarm", "waiting"].includes(S.state.kind)'
      in JS)


# --------------------------------------------------------------------------- #
print("3. control is unchanged by the waiting flag")


def run(opts, flag):
    async def go():
        h = _Hass()
        c = make(opts, h)
        (waiting_on if flag else waiting_off)()
        targets = []
        for _ in range(4):
            await c.read_inputs()
            await c.cycle()
            await c.send_heartbeats(asyncio.get_running_loop().time(), 10.0)
            targets.append(c.target)
        waiting_off()
        return targets, h.writes
    return asyncio.run(go())


for label, opts in (("N=1", ONE), ("N=2", TWO)):
    off, on = run(opts, False), run(opts, True)
    check(f"{label}: same targets and writes with the flag on or off "
          f"({on[0]}, {len(on[1])} writes)", off == on)
    check(f"{label}: ...and every write while Home Assistant does not answer "
          f"asks 0 W", on[1] and all(v == 0.0 for _e, v in on[1]))


# With 502 readings every write is 0 W whatever the flag says, so the check
# above cannot see a flag that forces 0 W. Live readings can: the flag alone
# must not change a non-zero write.
class _LiveHass(_Hass):
    async def number(self, e, invert=False):
        v = {"sensor.grid": 900.0}.get(e, 50.0 if "soc" in e else 0.0)
        return -v if invert else v

    async def set_number(self, e, v):
        self.writes.append((e, float(v)))
        return True


def run_live(opts, flag):
    async def go():
        h = _LiveHass()
        c = make(opts, h)
        (waiting_on if flag else waiting_off)()
        for _ in range(4):
            await c.read_inputs()
            await c.cycle()
            await c.send_heartbeats(asyncio.get_running_loop().time(), 10.0)
        waiting_off()
        return h.writes
    return asyncio.run(go())


for label, opts in (("N=1", ONE), ("N=2", TWO)):
    off, on = run_live(opts, False), run_live(opts, True)
    check(f"{label}: with live readings the flag changes no write "
          f"({on[-2:]})", off == on and any(v != 0.0 for _e, v in on))


print()
if fails:
    print(f"{len(fails)} FAILED")
    for f in fails:
        print("  -", f)
    raise SystemExit(1)
print("all passed")