goodwe-addon/goodwe_controller/test_maintenance.py

"""Runnable check for the maintenance state machine.  `python3 test_maintenance.py`

Walks a whole cycle in simulated time, which is the test the YAML version never
had - and it is exactly the phase transitions that a real cycle takes ten hours
to exercise once a month.

⚠️ MUTATING THE MODULE THIS FILE GRADES? PURGE `__pycache__` FIRST - or run
`tools/mutation_run.sh`, which does. `python3 -B` alone does NOT save you: it
stops bytecode being WRITTEN, not a stale one being READ, and a size-preserving
edit reverted inside one second is invisible to CPython's staleness test. That
is TEST-15, and it has already turned one live mutation into a false survivor.
Read MUTATION.md before you report a survivor.
"""

import dataclasses
import re
import sys
from datetime import datetime, timedelta, timezone
from pathlib import Path

from app.maintenance import (
    CHARGE, DRAIN, HOLD, IDLE, MAINT_OPTION_DEFAULTS, MaintConfig, Maintenance,
)

fails = []


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


class FakeStore:
    """Same surface as Store, no disk."""

    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 save(self):
        pass


cfg = MaintConfig(enabled=True, interval_days=28, start_hour=10,
                  discharge_w=2500, charge_w=2500,
                  soc_floor=11, soc_target=99, hold_min=120)

t0 = datetime(2026, 8, 23, 10, 0, tzinfo=timezone.utc)

print("scheduling")
m = Maintenance(cfg, FakeStore())
check("never run before is due", m.due(t0))
r = m.tick(t0, soc=80)
check("starts at the start hour when due", r.phase == DRAIN and r.owns_setpoint)
check("drain commands the discharge rate", r.setpoint_w == 2500)

m2 = Maintenance(cfg, FakeStore())
r = m2.tick(t0.replace(hour=11), soc=80)
check("does not start outside the start hour", r.phase == IDLE)

m3 = Maintenance(cfg, FakeStore())
m3.store.set_time("last_completed", t0 - timedelta(days=3))
check("not due three days after a cycle", not m3.due(t0))
check("due 29 days after a cycle",
      Maintenance(cfg, FakeStore()).due(t0) and
      (t0 - (t0 - timedelta(days=29))) >= timedelta(days=28))

print("ownership per phase")
m = Maintenance(cfg, FakeStore())
m.force_start(t0)
r = m.tick(t0, soc=80)
check("drain: maintenance owns the setpoint", r.owns_setpoint and not r.charge_only)
r = m.tick(t0 + timedelta(hours=3), soc=10.5)
check("drain exits at the floor", r.phase == CHARGE)
check("charge: the LOOP owns the setpoint", not r.owns_setpoint and r.charge_only)
r = m.tick(t0 + timedelta(hours=20), soc=99)
check("charge exits at the target", r.phase == HOLD)
check("hold: maintenance owns and commands zero",
      r.owns_setpoint and r.setpoint_w == 0.0)
r = m.tick(t0 + timedelta(hours=21), soc=100)
check("hold keeps holding before hold_min", r.phase == HOLD)
r = m.tick(t0 + timedelta(hours=22, minutes=5), soc=100)
check("hold completes after hold_min", r.phase == IDLE)
check("completion is recorded", m.store.get_time("last_completed") is not None)
check("no longer due right after completing", not m.due(t0 + timedelta(hours=22)))

print("failure handling")
m = Maintenance(cfg, FakeStore())
m.force_start(t0)
r = m.tick(t0 + timedelta(hours=13), soc=60)
check("drain aborts on timeout", r.phase == IDLE)
check("abort commands zero", r.setpoint_w == 0.0)

m = Maintenance(cfg, FakeStore())
m.force_start(t0)
r = m.tick(t0 + timedelta(minutes=5), soc=None)
check("no SoC: holds the phase rather than guessing", r.phase == DRAIN)
check("no SoC: still commands the drain rate", r.setpoint_w == 2500)

m = Maintenance(cfg, FakeStore())
m.store.set("phase", "banana")
r = m.tick(t0, soc=50)
check("unknown phase recovers to idle at 0 W", r.phase == IDLE and r.setpoint_w == 0.0)

print("disabled schedule")
m = Maintenance(MaintConfig(enabled=False), FakeStore())
check("disabled never starts", m.tick(t0, soc=80).phase == IDLE)

print("CFG-03: config.yaml is the source, this module is the only mirror")
# ⚠️ WHY THIS IS PINNED BY READING THE FILE. Before CFG-03 each maintenance
# default was written down three times - config.yaml's `options:` block,
# `opts.get(key, <literal>)` in main.py, and MaintConfig's dataclass defaults -
# and MAINT-04 proved they had already drifted (2000 shipped, 2500 in code). The
# fix is one source and one mirror; the mirror is only worth anything if drift
# from it is a CI failure rather than a comment, and a transcription of
# config.yaml into this file would just be a fourth place to disagree. So the
# shipped values are READ out of the file the Supervisor reads.
#
# Regex rather than PyYAML on purpose, for test_validate.py's reason: the add-on
# image installs py3-aiohttp and py3-paho-mqtt only, and a check that skips on
# ImportError is not a check.
HERE = Path(__file__).resolve().parent
_CFG_TEXT = (HERE / "config.yaml").read_text(encoding="utf-8")
_SCHEMA_AT = _CFG_TEXT.index("\nschema:")


def shipped_default(key):
    """The value under `options:` - the FIRST match, before `schema:`."""
    m = re.search(rf"^  {key}: *(.+?) *$", _CFG_TEXT[:_SCHEMA_AT], re.M)
    return None if m is None else m.group(1)


def schema_range(key):
    """`int(lo,hi)` under `schema:` -> (lo, hi); None if the key is absent."""
    m = re.search(rf"^  {key}: *\w+\((-?[\d.]+),(-?[\d.]+)\) *$",
                  _CFG_TEXT[_SCHEMA_AT:], re.M)
    return None if m is None else (float(m.group(1)), float(m.group(2)))


def as_number(text):
    if text in ("true", "false"):
        return 1.0 if text == "true" else 0.0
    return float(text)


_ship = {k: shipped_default(k) for k in MAINT_OPTION_DEFAULTS}
print(f"    config.yaml ships {_ship}")
check("every mirrored key is actually shipped in config.yaml's options: block",
      all(v is not None for v in _ship.values()))
check("...and MAINT_OPTION_DEFAULTS reproduces each one exactly, so the mirror "
      "is a copy of the source and not an opinion about it",
      all(str(MAINT_OPTION_DEFAULTS[k]).lower() == _ship[k]
          for k in MAINT_OPTION_DEFAULTS))
# ⚠️ THE OTHER DIRECTION MATTERS TOO. Equality alone would still pass if the
# mirror quietly stopped covering a key someone added to config.yaml, and the
# key nobody mirrors is exactly the key that gets a literal typed for it.
_shipped_keys = set(re.findall(r"^  (maintenance_[a-z_]+):", _CFG_TEXT[:_SCHEMA_AT],
                               re.M))
check(f"...and it covers EVERY maintenance_* option the file ships, so a ninth "
      f"key cannot arrive unmirrored {sorted(_shipped_keys)}",
      _shipped_keys == set(MAINT_OPTION_DEFAULTS))

# ⚠️ WHERE THE RANGE LIVES, ASSERTED RATHER THAN COMMENTED. The PM proofread on
# this card named a FOURTH declaration site of a different kind:
# `maintenance_charge_w: int(500,5000)` in the schema block. It is a BOUND, not
# a default, and the decision this card records is that bounds live in the
# schema block ONLY - the Supervisor enforces them and reads only that file, so
# a Python copy would be a place to disagree and nothing else. What that
# decision needs from a check is (a) every mirrored key really does have a
# schema entry, or the Supervisor drops the option and the add-on silently runs
# on the mirror instead of the operator's value, and (b) the shipped default
# lies INSIDE its own range, because a default the schema rejects is a config
# that cannot be re-entered once it is edited.
_ranged = {k: schema_range(k) for k in MAINT_OPTION_DEFAULTS}
check("every mirrored key is in the schema: block too - `bool` has no range, so "
      "it is the one exception and it is named",
      all(v is not None for k, v in _ranged.items()
          if k != "maintenance_enabled")
      and _ranged["maintenance_enabled"] is None)
_outside = {k: (as_number(_ship[k]), rng) for k, rng in _ranged.items()
            if rng is not None and not rng[0] <= as_number(_ship[k]) <= rng[1]}
check(f"...and every shipped default sits inside its own schema range, so the "
      f"Supervisor accepts the file it ships with {_outside or 'all inside'}",
      not _outside)

# ⚠️ AND THE CONSUMERS MOVE WITH IT - checked as a RELATION, not against
# numbers. AC4 asks for the replacement to be broken and every consumer observed
# moving together. A number typed here would pass whether or not the derivation
# survived; `dataclasses.fields()` reads what the class actually defaulted to,
# so this can only pass while the class is still indexing the mirror.
_fields = {f.name: f.default for f in dataclasses.fields(MaintConfig)}
check(f"MaintConfig's field names are the option names minus the prefix, so a "
      f"new field cannot miss the mirror by being spelt differently {sorted(_fields)}",
      {f"maintenance_{n}" for n in _fields} == set(MAINT_OPTION_DEFAULTS))
check("...and every dataclass default IS the mirror's value - no literal "
      "survives in the class that used to hold the third opinion",
      all(float(d) == float(MAINT_OPTION_DEFAULTS[f"maintenance_{n}"])
          for n, d in _fields.items()))
# from_options() is the only reader, so it is the only thing that can put a
# different number in front of main.py.
_from_empty = MaintConfig.from_options({})
check("...and from_options({}) - the sim/inject.py and test-suite path, the one "
      "the Supervisor never takes and the only one the divergence was reachable "
      "from - equals MaintConfig() exactly",
      _from_empty == MaintConfig())
check("...while a supplied value still wins, or the mirror would be a ceiling "
      "rather than a default",
      MaintConfig.from_options({"maintenance_charge_w": 1234}).charge_w == 1234.0)
# ⚠️ `key in opts`, not truthiness: `false` and `0` are values an operator can
# set, and a truthy read would silently replace both with the default - this
# card's own bug in a different costume.
check("...and an explicit falsey value is honoured rather than defaulted away",
      MaintConfig.from_options({"maintenance_start_hour": 0}).start_hour == 0
      and MaintConfig.from_options({"maintenance_enabled": False}).enabled is False)

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