goodwe-addon/goodwe_controller/test_rehearse.py

"""Runnable check for the roster rehearsal.  `python3 test_rehearse.py`

ARCH-02. `tools/rehearse_roster.py` is the instrument that decides whether the
flat -> roster switch on 2026-09-12 loses a safety latch, a month of maintenance
history, or a Home Assistant entity that an automation is watching. This file
grades the instrument.

⚠️ AN INSTRUMENT THAT HAS NEVER BEEN SEEN TO GO RED IS NOT EVIDENCE. This
repository's most-repeated defect, `MUTATION.md` habit 4 and now THIRTEEN
recorded instances, is "a check whose DATA cannot reach the failure it names" -
and a rehearsal that reports "migration OK" because it compared a structure to
itself is the purest example of it there is. So this file does two things in
order, and the first one is the one that makes the second mean anything:

  Part 2  TWO CLEAN CONTROLS. Two (state file, roster) pairs on which the
          shipped code is CORRECT and the harness must report NOTHING. A
          harness that is red on every input has measured nothing, and every
          red below it would be free.

  Part 3  ⭐ SIX DELIBERATE MUTATIONS OF THE SHIPPED MIGRATION AND OF THE
          ENTITY-ID COMPOSITION, each applied to a private copy of the tree,
          each required to flip its control from green to red, AND each
          required to produce the SPECIFIC finding it should - not merely a
          non-zero exit. A mutation that goes red for the wrong reason is a
          harness that would go red on the real defect for the wrong reason
          too.

  Part 4  ONE MUTATION THAT DELIBERATELY SURVIVES, with the reason written
          down. A battery in which everything dies is a battery whose author
          stopped as soon as the number looked good.

  Part 5  SAFETY-44's exact path, pinned as a finding, by name, in both
          directions.

⚠️ THE MUTATIONS ARE SOURCE EDITS IN A COPIED TREE, NOT MONKEYPATCHES. A
monkeypatch grades the name a module was asked for; an edit grades the code that
runs. Each copy is made with `mkdtemp()`, so two runs of this file cannot
overwrite each other's work - habit 18 - and `__pycache__` is never copied, so
no mutant is ever graded through a stale `.pyc` - habit 1 / TEST-15.
"""

import json
import os
import re
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path

HERE = Path(__file__).resolve().parent
FIX = HERE / "tools" / "rehearsal"

fails = []


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


# --------------------------------------------------------------------------- #
print("1. the harness's own preconditions")
# --------------------------------------------------------------------------- #
sys.path.insert(0, str(HERE / "tools"))
import rehearse_roster as RR  # noqa: E402

check("the tool refuses rather than passing when it cannot run: its three exit "
      "codes are distinct and 2 is not 0",
      (RR.EXIT_OK, RR.EXIT_FINDING, RR.EXIT_COULD_NOT_RUN) == (0, 1, 2))

# ⚠️ THE ANCHOR. `app/mqtt.py` records `sensor.goodwe_p1_sample_age` as VERIFIED
# against the live system on 2026-08-26, and the ESP32's ingest guard subscribes
# to it. The tool's entity-id model is written independently of `mqtt._slug` on
# purpose (habit 4), so this is the one external fact that says the independent
# model is right. If it fails, every orphan claim the tool makes is worthless
# and the tool exits 2 rather than reporting.
try:
    RR._check_anchor()
    anchored = True
except RR.Refused:
    anchored = False
check("the entity-id model reproduces the ONE id verified against the live "
      f"system ({RR.ANCHOR_ID})", anchored)

check("...and it is NOT built out of `mqtt._slug`, so a mutation in the "
      "shipped slugifier moves one side of the comparison and not both",
      "_slug" not in RR.compose.__code__.co_names)

_opts = RR.load_options(str(FIX / "roster.yaml"))
check("the built-in YAML reader parses the roster fixture the check image "
      "cannot parse with PyYAML (there is no `yaml` module in it)",
      [e["name"] for e in _opts["inverters"]] == ["garage", "shed"]
      and _opts["max_w"] == 5000 and _opts["batt_invert"] is False
      and _opts["meter_source"] == "off")
try:
    RR._mini_yaml("inverters: [a, b]\n")
    _refused = False
except RR.Refused:
    _refused = True
check("...and it REFUSES a construct it does not understand rather than "
      "guessing, because a config silently mis-parsed here rehearses a "
      "configuration nobody is going to deploy", _refused)

_flat = RR.load_options(str(FIX / "flat.yaml"))
check("⚠️ the two configs differ in the ONE way the transition is about: the "
      "BEFORE roster is empty and the AFTER roster is declared - if they did "
      "not, every diff below would be a structure compared to itself",
      _flat.get("inverters") == [] and len(_opts["inverters"]) == 2)
check("...and the AFTER roster's first entry is NOT `inverter-a`, which is "
      "habit 4's seventh instance in one line: with it, `UNIT_A` and "
      "`inverters[0].name` are the same string and a misdirected namespace "
      "cannot be seen",
      _opts["inverters"][0]["name"] != RR.UNIT_A)

_legacy = json.loads((FIX / "state-legacy.json").read_text())
check("⚠️ the synthetic state carries MAINTENANCE HISTORY - state built by the "
      "happy path cannot detect a migration that drops it",
      _legacy.get("last_completed") and _legacy.get("phase"))
check("⚠️ ...and a LEGACY-SHAPED KEY WHOSE LOSS WOULD BE SILENT: a LATCHED "
      "`sign_tripped` with no `schema_version` beside it, i.e. written by a "
      "controller that had never heard of unit namespaces",
      _legacy.get("sign_tripped") is True
      and "schema_version" not in _legacy
      and "maint_schema_version" not in _legacy)


# --------------------------------------------------------------------------- #
print()
print("2. THE CLEAN CONTROLS: two transitions on which the shipped code is "
      "right and the harness must say nothing")
# --------------------------------------------------------------------------- #
# ⚠️ THIS PART IS WHAT MAKES PART 3 EVIDENCE. Six reds prove nothing about an
# instrument that is red on its own fixtures. Each control below is a genuine
# flat -> roster switch that the shipped migration handles correctly, and the
# required answer is EXIT 0.
#
#   CONTROL-A  a PRE-DEV-02 flat store, latch tripped, onto a roster whose
#              first entry IS `inverter-a`. This is the path SAFETY-44's
#              documented reasoning describes, and on it that reasoning holds.
#   CONTROL-B  a store that has already run flat under 0.7.0 (schema stamped,
#              no latch), onto a roster named `garage`/`shed`. This is the
#              MAINT-05 half, and the names are deliberately not `inverter-a`.
#   CONTROL-C  a FRESH INSTALL - an empty state file - onto `garage`/`shed`.
#              ⚠️ THIS ONE FOUND A DEFECT IN THE HARNESS. A fresh store's
#              migration writes `sign_tripped = False` and `sign_reason = ""`
#              under `inverter-a` before any roster exists, and the first draft
#              reported both as STRANDED - red on every fresh install. Losing a
#              DEFAULT loses nothing; that is `MigrationReport.lost`'s own
#              distinction, and an alarm that fires on the harmless case is one
#              an operator learns to dismiss before the real one arrives.
CONTROLS = {
    "CONTROL-A": ("state-legacy.json", "roster-inverter-a.yaml"),
    "CONTROL-B": ("state-flat-070.json", "roster.yaml"),
    "CONTROL-C": ("state-fresh.json", "roster.yaml"),
}


def run(tree: Path, state: str, after: str, *extra) -> "tuple[int, str]":
    """The harness, as the owner runs it, over `tree`.

    ⚠️ ALWAYS WITH `--live-entities`. That file is the only part of the
    entity-id comparison that does not move with the code, so a run without it
    is a run that cannot see a renamed device.
    """
    fixtures = tree / "tools" / "rehearsal"
    proc = subprocess.run(
        [sys.executable, str(tree / "tools" / "rehearse_roster.py"),
         "--state", str(fixtures / state),
         "--before", str(fixtures / "flat.yaml"),
         "--after", str(fixtures / after),
         "--live-entities", str(fixtures / "live-entities-recorded.txt"),
         *extra],
        capture_output=True, text=True,
        env={**os.environ, "PYTHONDONTWRITEBYTECODE": "1"})
    return proc.returncode, proc.stdout + proc.stderr


for label, (state, after) in CONTROLS.items():
    code, out = run(HERE, state, after)
    check(f"{label}: the shipped code, {state} -> {after}, exits 0 with no "
          f"findings", code == RR.EXIT_OK and "NO FINDINGS" in out)

# ⚠️ AND THE INPUT FILE IS NEVER WRITTEN. The owner runs this against a copy of
# a live state file; a tool that edited it would be a tool that had to be
# trusted rather than one that cannot do harm.
_before_bytes = (FIX / "state-legacy.json").read_bytes()
run(HERE, "state-legacy.json", "roster.yaml", "--reverse", "--round-trip")
check("the state file the harness was pointed at is BYTE-IDENTICAL afterwards, "
      "including after the reverse and round-trip legs, which write",
      (FIX / "state-legacy.json").read_bytes() == _before_bytes)


# --------------------------------------------------------------------------- #
print()
print("3. ⭐ THE MUTATIONS: each one breaks the shipped migration or the "
      "entity-id composition, and each must turn its control RED")
# --------------------------------------------------------------------------- #
def mutant_tree(edits) -> str:
    """A private copy of the add-on with `edits` applied.

    ⚠️ EVERY EDIT IS ASSERTED TO HAVE MATCHED. Four defects in this session came
    from `str.replace` calls whose pattern did not match: the call silently
    returns the original string, the mutation never happens, and the harness
    reports a SURVIVOR on code it never changed. That is a fabricated result.
    """
    tmp = tempfile.mkdtemp(prefix="rehearse-mut-")
    dest = Path(tmp) / "tree"
    ignore = shutil.ignore_patterns("__pycache__", "*.pyc")
    (dest / "tools").mkdir(parents=True)
    shutil.copytree(HERE / "app", dest / "app", ignore=ignore)
    shutil.copytree(FIX, dest / "tools" / "rehearsal", ignore=ignore)
    shutil.copyfile(HERE / "tools" / "rehearse_roster.py",
                    dest / "tools" / "rehearse_roster.py")
    for rel, old, new in edits:
        path = dest / rel
        text = path.read_text(encoding="utf-8")
        hits = text.count(old)
        if hits != 1:
            raise AssertionError(
                f"mutation target appears {hits} times in {rel}, expected 1: "
                f"{old!r} - the mutation did NOT happen and any result from it "
                f"would be fabricated")
        path.write_text(text.replace(old, new), encoding="utf-8")
    return str(dest)


MUTATIONS = [
    # (id, control, what it breaks, edits, the finding it must produce)
    ("M1", "CONTROL-B",
     "the maintenance migration is handed the hardcoded `UNIT_A` instead of "
     "the roster's own first name - MAINT-05's exact shipped regression, which "
     "left all 19 suites green",
     [("app/main.py",
       "store, self.inverters[0].name, [inv.name for inv in self.inverters])",
       "store, UNIT_A, [inv.name for inv in self.inverters])")],
     "MAINTENANCE HISTORY LOST"),

    ("M2", "CONTROL-B",
     "the maintenance namespace is DROPPED: the migration moves no keys at all, "
     "so a month of history stays under a flat spelling the new packs do not "
     "read",
     [("app/store.py",
       "_move_flat_keys(store, unit, MAINT_UNIT_KEYS, rep)",
       "_move_flat_keys(store, unit, (), rep)")],
     "MAINTENANCE HISTORY LOST"),

    ("M3", "CONTROL-A",
     "the SAFETY-02 pair loses HALF of itself: `sign_reason` drops out of "
     "`LEGACY_UNIT_KEYS`, so the latch migrates without the sentence that says "
     "why it tripped",
     [("app/store.py",
       'LEGACY_UNIT_KEYS = ("sign_tripped", "sign_reason")',
       'LEGACY_UNIT_KEYS = ("sign_tripped",)')],
     "LATCH KEY NOT UNDER A ROSTER NAME"),

    ("M4", "CONTROL-A",
     "the READER's key composition changes while the migration's does not - "
     "`Inverter.key()` upper-cases the unit - so the latch is written "
     "perfectly, read back perfectly by the migration, and looked for under a "
     "name that is not there",
     [("app/inverter.py",
       "return unit_key(self.name, name)",
       "return unit_key(self.name.upper(), name)")],
     "LATCH CAME UP DISARMED"),

    ("M5", "CONTROL-A",
     "the entity-id composition changes on the DEVICE half: the MQTT device is "
     "renamed, which is the 'tidy it up' change `app/mqtt.py` spends thirty "
     "lines in capitals refusing. ⚠️ THE BEFORE/AFTER DIFF IS BLIND TO THIS - "
     "both sides are composed by the mutant, so every id moves equally and "
     "nothing looks orphaned. Only the recorded list catches it",
     [("app/mqtt.py", '"name": "GoodWe",', '"name": "GoodWe Controller",')],
     "COMPOSITION DRIFT"),

    ("M6", "CONTROL-A",
     "the entity-id composition changes on the ENTITY half: one site sensor is "
     "renamed, which orphans exactly one id and leaves the other seven alone",
     [("app/mqtt.py",
       '("grid", "goodwe_grid_power", "Grid power", "W", "power", "measurement", None),',
       '("grid", "goodwe_grid_power", "Grid power W", "W", "power", "measurement", None),')],
     "COMPOSITION DRIFT: this build announces sensor.goodwe_grid_power_w"),
]

for mid, control, what, edits, must_say in MUTATIONS:
    state, after = CONTROLS[control]
    tree = mutant_tree(edits)
    code, out = run(Path(tree), state, after)
    red = code == RR.EXIT_FINDING
    named = must_say in out
    print()
    print(f"  --- {mid} ({control}) {what}")
    print(f"      exit {code} (0=clean 1=finding 2=could-not-run)")
    # ⚠️ THE FINDINGS, DE-DUPLICATED AND IN ORDER. The tool prints each finding
    # once per transition leg and once more in its summary, and its own header
    # is full of ⚠️ paragraphs - printing every line that starts with one would
    # bury the evidence this part exists to show under the tool's docstring.
    seen = []
    for line in out.splitlines():
        text = line.strip()
        if text.startswith("⚠️ ") and re.match(r"⚠️ [A-Z][A-Z -]+[A-Z()]:", text):
            if text not in seen:
                seen.append(text)
    for text in seen:
        print(f"      {text[:160]}")
    check(f"{mid} turns {control} RED", red)
    check(f"{mid} says WHY, and it is the right why: {must_say!r}", named)
    shutil.rmtree(Path(tree).parent, ignore_errors=True)


# --------------------------------------------------------------------------- #
print()
print("4. THE MUTATION THAT SURVIVES, ON PURPOSE, WITH THE REASON WRITTEN DOWN")
# --------------------------------------------------------------------------- #
# ⚠️ A BATTERY IN WHICH EVERYTHING DIES IS A BATTERY SOMEBODY STOPPED WRITING
# AS SOON AS THE NUMBER LOOKED GOOD. `store.unit_key()` is the ONE spelling of
# the namespaced key, read by `Inverter.key()`, by `UnitHistory.key()` and by
# both migrations - that is store.py's stated design, "one spelling, in one
# place". Changing it therefore moves the writer AND every reader together, and
# the store comes out internally consistent under a different name.
#
# The harness is BLIND to it and cannot be otherwise: nothing inside one version
# of this add-on can see that the key shape stopped matching what an OLDER
# version reads. That is a ROLLBACK contract, and grading it needs two versions
# of the code, which is a different instrument (`tools/predeploy_check.py`'s
# territory). Stated here rather than left as an unexplained green.
_tree = mutant_tree([("app/store.py",
                      'return f"unit/{unit}/{key}"',
                      'return f"ns.{unit}.{key}"')])
_code, _out = run(Path(_tree), *CONTROLS["CONTROL-A"])
print(f"      renaming the ONE key spelling -> exit {_code}")
check("renaming `unit_key()` survives, because writer and readers move "
      "together - a fact about the design, recorded, not a gap being hidden",
      _code == RR.EXIT_OK)
check("...and the store really did change shape underneath it, so this is a "
      "survivor of a mutation that HAPPENED rather than of one that did not",
      "ns.inverter-a.sign_tripped" in _out and "unit/inverter-a/" not in _out)
shutil.rmtree(Path(_tree).parent, ignore_errors=True)


# --------------------------------------------------------------------------- #
print()
print("5. SAFETY-44: a pre-DEV-02 flat store meeting a DECLARED ROSTER, which "
      "is the path nobody had established either way")
# --------------------------------------------------------------------------- #
# The question on the card: `migrate_unit_keys(store, UNIT_A)` takes a hardcoded
# constant and has no `misdirected` guard, while `migrate_maint_keys(store,
# unit, roster)` takes the roster and does. Is that asymmetry safe?
#
# ⚠️ MEASURED, NOT ARGUED. The two checks below are the answer, and they are
# written as REQUIREMENTS so that a fix to `migrate_unit_keys` turns them red
# and this file has to be updated deliberately rather than drifting.
code44, out44 = run(HERE, "state-legacy.json", "roster.yaml")
print(f"      flat store + roster ['garage','shed'] -> exit {code44}")
check("⚠️ SAFETY-44 IS A DEFECT AGAIN: a store that ran pre-DEV-02 flat and "
      "declares a multi-inverter roster comes up with its SAFETY-02 latch "
      "DISARMED - the file says `sign_tripped: true` and not one inverter is "
      "latched",
      code44 == RR.EXIT_FINDING and "LATCH CAME UP DISARMED" in out44)
check("...and the harness names the two latch keys SAFETY-44 asks about, by "
      "the namespace they were stranded in",
      "unit/inverter-a/sign_tripped" in out44
      and "unit/inverter-a/sign_reason" in out44)
check("...and the mechanism is visible in the report: the keys are stranded "
      "under `inverter-a`, which is not on the roster, so no machine reads them",
      "STRANDED STORE KEY" in out44 and "['garage', 'shed']" in out44)
check("⚠️ AND IT IS NOT THE SAME-STEP PATH ALONE. Running the flat config "
      "first and declaring the roster afterwards - the staged upgrade - "
      "reaches the same disarmed latch, because the first start stamps "
      "`schema_version` and the second start's migration then returns early",
      out44.count("LATCH CAME UP DISARMED") >= 2)

# ⚠️ AND THE CONTRAST, WHICH IS WHAT STOPS THE CHECK ABOVE BEING VACUOUS. The
# SAME state file onto a roster whose first entry IS `inverter-a` comes up
# LATCHED. The two runs differ in one thing - the first roster name - so the
# finding is attributable to it and to nothing else.
code_ok, out_ok = run(HERE, "state-legacy.json", "roster-inverter-a.yaml")
check("the SAME state file onto a roster whose first entry IS `inverter-a` "
      "comes up latched and clean - so the finding above is caused by the "
      "roster NAME and not by the fixture being tripped",
      code_ok == RR.EXIT_OK and "LATCH CAME UP DISARMED" not in out_ok)


# --------------------------------------------------------------------------- #
print()
print("6. THE REVERSE DIRECTION: the retreat, and the return from it")
# --------------------------------------------------------------------------- #
code_r, out_r = run(HERE, "state-flat-070.json", "roster.yaml",
                    "--reverse", "--round-trip")
# ⚠️ AND THE SOFTENING IS NOT A HOLE. The SAME key stranded with a NON-default
# value is still a finding - that is the whole of part 5 - so the two cases are
# separated by the VALUE and not by the key name.
_fresh_code, _fresh_out = run(HERE, "state-fresh.json", "roster.yaml")
check("a fresh install DOES strand `unit/inverter-a/sign_tripped` - the key is "
      "written and nothing on the roster reads it - and the report SAYS SO",
      _fresh_code == RR.EXIT_OK
      and "unit/inverter-a/sign_tripped = False" in _fresh_out)
check("...and does NOT call it a finding, because the value is the default; "
      "the same key at True is the finding part 5 turns on, so the two cases "
      "are separated by the VALUE and not by the key name",
      "STRANDED STORE KEY: unit/inverter-a/sign_tripped" not in _fresh_out
      and "STRANDED STORE KEY: unit/inverter-a/sign_tripped = True"
      in "\n".join(l.strip() for l in
                    run(HERE, "state-legacy.json", "roster.yaml")[1].splitlines()))

_orphaned = set(re.findall(r"ORPHANED ENTITY: (sensor\.\S+)", out_r))
print(f"      the retreat orphans: {sorted(_orphaned)}")
check("the RETREAT is rehearsed at all, and it reports the entities it "
      "orphans - the eight per-unit rows, which is the cost of backing out",
      "THE RETREAT" in out_r and _orphaned == {
          f"sensor.goodwe_{u}_{k}"
          for u in ("garage", "shed")
          for k in ("setpoint", "battery_power", "battery_soc", "sign_latch")})
check("...and NOT one site entity: the eight ids the install has today survive "
      "the retreat, which is what makes the eight above attributable to the "
      "per-unit rows and not to a blanket rename",
      not any(o in _orphaned for o in
              ("sensor.goodwe_grid_power", "sensor.goodwe_setpoint",
               "sensor.goodwe_battery_power", "sensor.goodwe_p1_sample_age")))
check("...and the first pack's maintenance history SURVIVES the retreat, "
      "because `UnitHistory.mirror_flat` keeps the flat keys live - so the "
      "retreat itself does not lose state",
      "MAINTENANCE HISTORY LOST" not in out_r.split("THE ROUND TRIP")[0]
      .split("THE RETREAT")[-1])
check("⚠️ ...but the ROUND TRIP does not: a maintenance cycle completed while "
      "the install is rolled back is DISCARDED when it re-advances, because "
      "`maint_schema_version` is already stamped and the migration skips",
      "RETREAT-WINDOW WORK DISCARDED" in out_r)
check("...and that is a leg whose data can reach the failure it names: the "
      "retreat window actually contains a completion, written through the "
      "shipped `UnitHistory`, at a timestamp the starting fixture does not have",
      RR.RETREAT_WINDOW_COMPLETION in out_r
      and RR.RETREAT_WINDOW_COMPLETION
      not in (FIX / "state-flat-070.json").read_text())


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