"""MAINT-10: the maintenance start hour is LOCAL time.  `python3 test_maint10.py`

Graded here: the slot on ordinary days and on both DST change days, the zone
source and its fallbacks, the 19:00 default, the startup wiring, and that the
controller's commands are unchanged once the start instant is the same.
See DECISIONS/MAINT-10-local-start-hour.md.

Mutating what this file grades? Read MUTATION.md first (stale bytecode).

⚠️ The CI image has no tzdata, so every DST check runs on tariff.py's encoded
EU rule and, where zoneinfo works, on Europe/Brussels from tzdata as well.
"""

import asyncio
import functools
import logging
import os
import re
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path

sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "tools"))

import app.main as M                                        # noqa: E402
from app import maintenance as MT                           # noqa: E402
from app.maintenance import (                               # noqa: E402
    DRAIN, IDLE, MAINT_OPTION_DEFAULTS, MaintConfig, Maintenance,
    in_start_slot, next_slot_start, site_zone,
)
from app.tariff import _EUTimezone                          # noqa: E402

HERE = Path(__file__).resolve().parent
UTC = timezone.utc
fails = []


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


ZONES = [("encoded EU rule", _EUTimezone())]
try:
    from zoneinfo import ZoneInfo
    ZONES.append(("zoneinfo Europe/Brussels", ZoneInfo("Europe/Brussels")))
except Exception as err:                                    # noqa: BLE001
    print(f"    zoneinfo unavailable here ({err}); graded on the EU rule only")

SPRING = datetime(2026, 3, 29, tzinfo=UTC)     # 02:00 CET -> 03:00 CEST
SPRING27 = datetime(2027, 3, 28, tzinfo=UTC)   # the next one, last Sunday of March
AUTUMN = datetime(2026, 10, 25, tzinfo=UTC)    # 03:00 CEST -> 02:00 CET
SUMMER = datetime(2026, 7, 15, tzinfo=UTC)
WINTER = datetime(2026, 1, 15, tzinfo=UTC)


def slot_minutes(day_utc, hour, tz):
    """UTC minutes in the slot, over the LOCAL day of that date."""
    t, out = day_utc - timedelta(hours=3), []
    while t < day_utc + timedelta(hours=27):
        if (t.astimezone(tz).date() == day_utc.date()
                and in_start_slot(t, hour, tz)):
            out.append(t)
        t += timedelta(minutes=1)
    return out


def one_block(mins):
    """`(start, minutes)` when the minutes form one contiguous run, else None."""
    if not mins:
        return None
    for a, b in zip(mins, mins[1:]):
        if b - a != timedelta(minutes=1):
            return None
    return mins[0], len(mins)


# =========================================================================== #
print("1. UTC stays what it was: the slot IS `now.hour == start_hour`")
t, same = datetime(2026, 1, 1, tzinfo=UTC), True
while t < datetime(2027, 1, 1, tzinfo=UTC):
    for h in range(24):
        if in_start_slot(t, h, UTC) != (t.hour == h):
            same = False
    t += timedelta(minutes=37)
check("over a year of instants, every hour 0-23: identical to the old test",
      same)
check("a naive `now` is read as UTC, as before",
      in_start_slot(datetime(2026, 7, 1, 19, 5), 19, UTC)
      and not in_start_slot(datetime(2026, 7, 1, 18, 5), 19, UTC))

# =========================================================================== #
print("2. local time: 19 means 19:00 on the site's clock")
for label, tz in ZONES:
    s = one_block(slot_minutes(SUMMER, 19, tz))
    w = one_block(slot_minutes(WINTER, 19, tz))
    check(f"[{label}] summer: 17:00-18:00 UTC (CEST), one hour",
          s == (SUMMER + timedelta(hours=17), 60))
    check(f"[{label}] winter: 18:00-19:00 UTC (CET), one hour",
          w == (WINTER + timedelta(hours=18), 60))
    check(f"[{label}] the old reading is gone: 19:00 UTC is not the slot "
          f"in summer", not in_start_slot(SUMMER + timedelta(hours=19, minutes=5),
                                          19, tz))

# =========================================================================== #
print("3. the DST change days, every start hour")
for label, tz in ZONES:
    for name, day in (("spring", SPRING), ("spring 2027", SPRING27),
                      ("autumn", AUTUMN)):
        bad = []
        for h in range(24):
            b = one_block(slot_minutes(day, h, tz))
            # the local day runs 23:00 UTC the day before to 22:00/23:00 UTC
            if b is None or b[1] != 60:
                bad.append((h, b))
        check(f"[{label}] {name}: every hour 0-23 gives ONE slot of exactly "
              f"60 minutes that day {bad or ''}", not bad)
    sp2 = one_block(slot_minutes(SPRING, 2, tz))
    check(f"[{label}] spring, hour 2 does not exist: it starts at 03:00 CEST "
          f"(01:00 UTC), once", sp2 == (SPRING + timedelta(hours=1), 60))
    check(f"[{label}] spring 2027 (28 March), hour 2: 01:00 UTC as well",
          one_block(slot_minutes(SPRING27, 2, tz))
          == (SPRING27 + timedelta(hours=1), 60))
    check(f"[{label}] the evening before spring 2027, hour 19: 18:00 UTC "
          f"(CET); that day, 17:00 UTC (CEST)",
          one_block(slot_minutes(SPRING27 - timedelta(days=1), 19, tz))
          == (SPRING27 - timedelta(hours=6), 60)
          and one_block(slot_minutes(SPRING27, 19, tz))
          == (SPRING27 + timedelta(hours=17), 60))
    au2 = one_block(slot_minutes(AUTUMN, 2, tz))
    check(f"[{label}] autumn, hour 2 happens twice: only the first, 02:00 CEST "
          f"(00:00 UTC)", au2 == (AUTUMN, 60))
    check(f"[{label}] autumn: the second 02:xx (CET) is not a slot",
          not in_start_slot(AUTUMN + timedelta(hours=1, minutes=30), 2, tz))
    check(f"[{label}] autumn, hour 19 still 18:00 UTC (CET)",
          one_block(slot_minutes(AUTUMN, 19, tz))
          == (AUTUMN + timedelta(hours=18), 60))
    check(f"[{label}] spring, hour 19 already 17:00 UTC (CEST)",
          one_block(slot_minutes(SPRING, 19, tz))
          == (SPRING + timedelta(hours=17), 60))


# =========================================================================== #
class FakeStore:
    def __init__(self):
        self.data = {"phase": IDLE, "phase_started": None,
                     "last_completed": None, "last_start_attempt": None}

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

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

    def get_time(self, k):
        return self.data.get(k)

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

    def save(self):
        pass


def starts(day, hour, tz, *, abort_after=None, blocked=None):
    """Tick a due pack each minute around a day; the UTC instants in that
    LOCAL day at which it entered DRAIN. `abort_after` aborts each drain that many minutes in (interlock);
    `blocked(t)` defers it."""
    m = Maintenance(MaintConfig(enabled=True, start_hour=hour), FakeStore(),
                    tz=tz)
    out, since = [], None
    t = day - timedelta(hours=2)
    while t < day + timedelta(hours=26):
        why = ""
        if since is not None and abort_after is not None \
                and t - since >= timedelta(minutes=abort_after):
            why = "a test interlock"
        if blocked is not None and blocked(t):
            why = "the other battery is cycling"
        was = m.phase
        m.tick(t, soc=80, blocked_by=why)
        if m.phase == DRAIN and was != DRAIN:
            out.append(t)
            since = t
        if m.phase == IDLE:
            since = None
        t += timedelta(minutes=1)
    return [x for x in out if x.astimezone(tz).date() == day.date()]


print("4. the state machine: one start per day, DST days included")
for label, tz in ZONES:
    check(f"[{label}] spring, hour 2: one start, at 01:00 UTC (03:00 CEST)",
          starts(SPRING, 2, tz, abort_after=5)
          == [SPRING + timedelta(hours=1)])
    # An aborted start waits RESTART_COOLDOWN_MIN (90). The repeated hour is
    # 120 minutes long, so without the first-pass rule it would start again.
    check(f"[{label}] autumn, hour 2, aborted after 5 min: NO second start "
          f"in the repeated hour",
          starts(AUTUMN, 2, tz, abort_after=5) == [AUTUMN])
    check(f"[{label}] autumn, hour 2, deferred for the whole first pass: it "
          f"does not start in the second pass either",
          starts(AUTUMN, 2, tz, blocked=lambda t: t < AUTUMN + timedelta(hours=1))
          == [])
    check(f"[{label}] summer, hour 19: one start at 17:00 UTC",
          starts(SUMMER, 19, tz, abort_after=5)
          == [SUMMER + timedelta(hours=17)])

# =========================================================================== #
print("5. where the start is shown (homeview) is where it happens")
for label, tz in ZONES:
    ok = True
    for day in (SPRING, AUTUMN, SUMMER):
        for h in (0, 2, 3, 10, 19, 23):
            first = next(iter(slot_minutes(day, h, tz)), None)
            if first is None:
                continue
            for probe in (first - timedelta(hours=5), first,
                          first + timedelta(minutes=59)):
                if next_slot_start(probe, h, tz) != first:
                    ok = False
    check(f"[{label}] next_slot_start() lands on the slot's first minute, "
          f"from before it and from inside it", ok)

# =========================================================================== #
print("6. which zone: Home Assistant, else the container's TZ, else UTC")
_FAKE = {"Europe/Brussels": ZONES[-1][1], "America/New_York": object(),
         "UTC": UTC}
_real_load = MT._load_zone
MT._load_zone = lambda n: ((_FAKE[n], "") if n in _FAKE
                           else (None, f"{n!r} is not a known zone"
                                 if n else "none given"))
try:
    tz, name, src, _ = site_zone("Europe/Brussels", "", "America/New_York")
    check("Home Assistant's zone wins over the container's TZ",
          name == "Europe/Brussels" and src == "ha"
          and tz is _FAKE["Europe/Brussels"])
    tz, name, src, why = site_zone(None, "ClientConnectorError",
                                   "America/New_York")
    check("Home Assistant unavailable: the container's TZ, and why",
          name == "America/New_York" and src == "tz"
          and why == "ClientConnectorError")
    tz, name, src, why = site_zone("Mars/Base", "", "Europe/Brussels")
    check("an unknown Home Assistant zone falls back too",
          name == "Europe/Brussels" and src == "tz" and "Mars/Base" in why)
    tz, name, src, why = site_zone(None, "no answer", None)
    check("neither: UTC", tz is UTC and src == "utc" and "no answer" in why)
finally:
    MT._load_zone = _real_load
if len(ZONES) > 1:
    check("the real loader resolves a real name and refuses a made-up one",
          MT._load_zone("Europe/Brussels")[0] is not None
          and MT._load_zone("Mars/Base")[0] is None)
check("the real loader refuses an empty name without raising",
      MT._load_zone("")[0] is None and MT._load_zone(None)[0] is None)

# =========================================================================== #
print("7. the default is 19, an explicit value is honoured")
cfg_text = (HERE / "config.yaml").read_text()
opts_block = cfg_text.split("\nschema:")[0]
check("config.yaml options: ships 19",
      re.search(r"^  maintenance_start_hour: 19$", opts_block, re.M) is not None)
check("config.yaml schema: still int(0,23)",
      re.search(r"^  maintenance_start_hour: int\(0,23\)$", cfg_text, re.M)
      is not None)
check("the in-code mirror says 19",
      MAINT_OPTION_DEFAULTS["maintenance_start_hour"] == 19
      and MaintConfig().start_hour == 19
      and MaintConfig.from_options({}).start_hour == 19)
check("an explicit 10 stays 10 (a stored option is never rewritten)",
      MaintConfig.from_options({"maintenance_start_hour": 10}).start_hour == 10)
_eu = ZONES[0][1]
_m10 = Maintenance(MaintConfig.from_options(
    {"maintenance_enabled": True, "maintenance_start_hour": 10}),
    FakeStore(), tz=_eu)
check("...and now means 10:00 local: 08:00 UTC in summer, not 10:00 UTC",
      _m10.wants_start(SUMMER + timedelta(hours=8, minutes=1))
      and not _m10.wants_start(SUMMER + timedelta(hours=10, minutes=1)))
check("a Maintenance built with no zone reads UTC, as before",
      Maintenance(MaintConfig(), FakeStore()).tz is UTC)


# =========================================================================== #
print("8. startup: Home Assistant's zone is asked for and reaches the machines")


class _Resp:
    def __init__(self, body, status=200):
        self.body, self.status = body, status

    async def __aenter__(self):
        return self

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

    async def json(self, content_type=None):
        if isinstance(self.body, Exception):
            raise self.body
        return self.body

    async def text(self):
        return "{}"


class _Session:
    def __init__(self, zone_body):
        self.zone_body, self.asked = zone_body, []

    async def __aenter__(self):
        return self

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

    def get(self, url, **kw):
        self.asked.append(url)
        self.timeouts = getattr(self, "timeouts", []) + [kw.get("timeout")]
        if url.endswith("/core/api/config"):
            return _Resp(self.zone_body)
        return _Resp({"data": {}})


s = _Session({"time_zone": "Europe/Brussels"})
check("ha_time_zone() reads /core/api/config's time_zone",
      asyncio.run(M.ha_time_zone(s, "tok")) == ("Europe/Brussels", "")
      and s.asked == [M.CORE_API + "/config"])
# Review: HA can take minutes to start (2026-10-05); the ask must not wait on it.
check("...with a time budget of at most 3 s, so a slow HA cannot hold up start",
      s.timeouts == [M.DOCS_LINK_TIMEOUT_S] and 0 < M.DOCS_LINK_TIMEOUT_S <= 3)
check("...and a failure is None plus the reason, never a raise",
      asyncio.run(M.ha_time_zone(_Session(OSError("down")), "tok"))
      == (None, "OSError: down")
      and asyncio.run(M.ha_time_zone(_Session({}), "tok"))[0] is None)


class _Built(Exception):
    pass


def amain_zone(zone_body, env_tz):
    """Run the real amain() up to the Controller; the tz its machines get."""
    got, logs = {}, []

    class _Capture(logging.Handler):
        def emit(self, rec):
            logs.append((rec.levelname, rec.getMessage()))

    class _Hass:
        async def limits(self, e):
            return (-5000.0, 5000.0, 1.0)

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

    real = {k: getattr(M, k) for k in ("load_options", "HomeAssistant",
                                       "validate_startup_config",
                                       "mode_announcement", "Controller")}
    real_session, real_store = M.aiohttp.ClientSession, None
    import app.store
    real_store = app.store.Store

    def _controller(opts, hass, store, pub, *, stop, site_tz=None, **kw):
        got["tz"] = site_tz
        raise _Built()

    M.load_options = lambda: {"setpoint_entity": "number.sp"}
    M.HomeAssistant = lambda session, **kw: _Hass()
    M.validate_startup_config = lambda *a, **k: []
    M.mode_announcement = lambda opts, store: ((), ())
    M.Controller = _controller
    M.aiohttp.ClientSession = lambda *a, **k: _Session(zone_body)
    app.store.Store = lambda *a, **k: object()
    old_tz = os.environ.pop("TZ", None)
    if env_tz:
        os.environ["TZ"] = env_tz
    h = _Capture()
    logging.getLogger("goodwe").addHandler(h)
    try:
        asyncio.run(M.amain())
    except _Built:
        pass
    finally:
        logging.getLogger("goodwe").removeHandler(h)
        for k, v in real.items():
            setattr(M, k, v)
        M.aiohttp.ClientSession = real_session
        app.store.Store = real_store
        os.environ.pop("TZ", None)
        if old_tz is not None:
            os.environ["TZ"] = old_tz
    return got.get("tz"), [m for lvl, m in logs if "Maintenance start" in m], logs


MT._load_zone = lambda n: ((_FAKE[n], "") if n in _FAKE
                           else (None, "none given"))
try:
    tz, lines, _ = amain_zone({"time_zone": "Europe/Brussels"}, "America/New_York")
    check("amain(): Home Assistant's zone reaches the Controller",
          tz is _FAKE["Europe/Brussels"])
    check("...logged once, naming the zone and the source",
          lines == ["Maintenance start hour is local time in Europe/Brussels, "
                    "from Home Assistant."])
    tz, lines, _ = amain_zone(OSError("down"), "America/New_York")
    check("amain(), Home Assistant unavailable: the container's TZ, said "
          "plainly", tz is _FAKE["America/New_York"]
          and lines == ["Maintenance start hour is local time in "
                        "America/New_York, from the add-on's time zone. Home "
                        "Assistant gave none: OSError: down."])
    tz, lines, logs = amain_zone(OSError("down"), None)
    check("amain(), neither: UTC, with a WARNING line",
          tz is UTC and any(lvl == "WARNING" and "uses UTC" in m
                            for lvl, m in logs))
    # Review: HA still starting (502) and the Supervisor's TZ is UTC.
    tz, lines, logs = amain_zone(OSError("502"), "UTC")
    check("amain(), Home Assistant unavailable and TZ is UTC: a WARNING "
          "that says to restart", tz is UTC
          and [(lvl, m) for lvl, m in logs if "Maintenance start" in m]
          == [("WARNING", "Maintenance start hour uses UTC, because Home "
               "Assistant did not answer. Restart the add-on when it is up.")])
    tz, lines, logs = amain_zone(OSError("502"), "America/New_York")
    check("...but a real TZ is used without a warning",
          not any(lvl == "WARNING" for lvl, m in logs
                  if "Maintenance start" in m))
finally:
    MT._load_zone = _real_load


# =========================================================================== #
print("9. nothing else in control moves: the same start instant, the same "
      "commands")
import opt11_identity as rig                                # noqa: E402
import safety78_identity as s78                             # noqa: E402

# rig's clock starts 2026-09-20 11:00 UTC = 13:00 CEST. The same drain,
# asked for as UTC hour 11 (the old reading) and as local hour 13 (the new).


def rows_with(start_hour, tz, n=1, both=False):
    """The rig's rows; `both=True` also returns the Controller."""
    real_drain, real_ctl = s78.DRAIN, rig.Controller
    s78.DRAIN = {"maintenance_enabled": True,
                 "maintenance_start_hour": start_hour}
    rig.Controller = functools.partial(real_ctl, site_tz=tz)
    try:
        # s78.run() wraps rig.Controller once more for its own tariff db.
        rows, c = s78.run(n, dur=600.0)
        return (rows, c) if both else rows
    finally:
        s78.DRAIN, rig.Controller = real_drain, real_ctl


_utc11 = rows_with(11, None)
_loc13 = rows_with(13, _eu)
check("precondition: the drain really starts in the UTC-11 run",
      any("drain" in r[-1][4] for r in _utc11))
check(f"start hour 11 UTC and 13 Europe/Brussels: every command, claim, "
      f"reason and log line identical over {len(_utc11)} cycles",
      _utc11 == _loc13)
_utc12 = rows_with(12, None)
_loc14 = rows_with(14, _eu)
check("precondition: no drain in the not-yet-due-hour run",
      not any("drain" in r[-1][4] for r in _utc12))
check("outside the start hour (12 UTC / 14 local): identical as well",
      _utc12 == _loc14)
check("...and the zone matters: hour 11 read in Brussels does NOT start",
      not any("drain" in r[-1][4] for r in rows_with(11, _eu)))

# =========================================================================== #
print("10. two inverters: every pack reads the start hour in local time")
_r2u, _c2u = rows_with(11, None, n=2, both=True)
_r2l, _c2l = rows_with(13, _eu, n=2, both=True)
check("N=2: both packs get the site's zone, not just the first",
      len(_c2l.maints) == 2 and all(mm.tz is _eu for mm in _c2l.maints))
check("N=2 precondition: a drain starts in the UTC-11 run",
      any(DRAIN in r[-1][4] for r in _r2u))
check(f"N=2: start hour 11 UTC and 13 Europe/Brussels identical over "
      f"{len(_r2u)} cycles", _r2u == _r2l)
check("N=2: hour 11 read in Brussels does NOT start either pack",
      not any(DRAIN in r[-1][4] for r in rows_with(11, _eu, n=2)))


def pair_starts(day0, days, hour, tz, cycle_min=70):
    """Two due packs, ticked each minute through `arbitrate()` as main.py
    does; a cycle takes about `cycle_min` minutes. `(pack, "DD HH:MM" UTC)`s."""
    cfg = MaintConfig(enabled=True, start_hour=hour, hold_min=10)
    ms = [Maintenance(cfg, FakeStore(), name=u, tz=tz)
          for u in ("A", "B")]
    ms[0].store.set_time("last_completed", day0 - timedelta(days=60))
    ms[1].store.set_time("last_completed", day0 - timedelta(days=40))
    half = (cycle_min - 10) // 2
    soc, out = [80.0, 80.0], []
    t = day0 - timedelta(hours=3)
    while t < day0 + timedelta(days=days):
        blocks = MT.arbitrate(ms, t)
        for i, mm in enumerate(ms):
            was = mm.phase
            mm.tick(t, soc=soc[i], blocked_by=blocks[i])
            if mm.phase == DRAIN:
                soc[i] -= (80.0 - cfg.soc_floor) / half
            elif mm.phase == MT.CHARGE:
                soc[i] += (cfg.soc_target - cfg.soc_floor) / half
            if mm.phase == DRAIN and was != DRAIN:
                out.append((mm.name, t))
        t += timedelta(minutes=1)
    return [(n, x.strftime("%d %H:%M")) for n, x in out]


def _at(*pairs):
    return [(n, x.strftime("%d %H:%M")) for n, x in pairs]


for label, tz in ZONES:
    # 24 Oct (CEST) and 25 Oct (CET): 19:00 local is 17:00 then 18:00 UTC.
    got = pair_starts(AUTUMN - timedelta(days=1), 2, 19, tz)
    check(f"[{label}] N=2 across the autumn change, hour 19: A at 19:00 "
          f"local on the 24th, B at 19:00 local on the 25th {got}",
          got == _at(("A", AUTUMN - timedelta(hours=7)),
                     ("B", AUTUMN + timedelta(hours=18))))
    got = pair_starts(SPRING27 - timedelta(days=1), 2, 19, tz)
    check(f"[{label}] N=2 across the spring 2027 change, hour 19: 18:00 UTC "
          f"then 17:00 UTC {got}",
          got == _at(("A", SPRING27 - timedelta(hours=6)),
                     ("B", SPRING27 + timedelta(hours=17))))
    # Hour 2 on 25 Oct: A takes the first pass (00:00 UTC) and is done in its
    # second pass. B must not start there - it waits for the next day.
    got = pair_starts(AUTUMN, 2, 2, tz)
    check(f"[{label}] N=2, autumn, hour 2: B does not start in the repeated "
          f"hour after A is done; it starts at 02:00 CET on the 26th {got}",
          got == _at(("A", AUTUMN),
                     ("B", AUTUMN + timedelta(days=1, hours=1))))
    got = pair_starts(SPRING27, 2, 2, tz)
    check(f"[{label}] N=2, spring 2027, hour 2: A at 03:00 CEST (01:00 UTC), "
          f"B the next night at 02:00 CEST (00:00 UTC) {got}",
          got == _at(("A", SPRING27 + timedelta(hours=1)),
                     ("B", SPRING27 + timedelta(days=1))))

print()
if fails:
    print(f"{len(fails)} FAILED: {', '.join(fails)}")
    sys.exit(1)
print("all checks passed")