"""SAFETY-60/62: the operator tools refuse input they cannot read. `python3 test_tool_inputs.py` Each tool is driven as the operator runs it, first on good input (the accepted baseline, MUTATION.md habit 7) and then on the shapes that used to pass. """ import contextlib import copy import importlib.util import io import json import os import shutil import subprocess import sys import tempfile from pathlib import Path HERE = Path(__file__).resolve().parent sys.path.insert(0, str(HERE)) TOOLS = HERE / "tools" FIX = TOOLS / "rehearsal" CAPTURE = HERE / "scenarios" / "safety57-2026-09-21" / \ "bus-a-2026-09-21-1030-1110.log.gz" TMP = tempfile.mkdtemp(prefix="test_tool_inputs.") ENV = {**os.environ, "PYTHONDONTWRITEBYTECODE": "1"} fails = [] def check(name, ok): print(f" {'ok ' if ok else 'FAIL'} {name}") if not ok: fails.append(name) def dump(name, obj=None, text=None): p = Path(TMP) / name p.write_text(text if text is not None else json.dumps(obj, indent=1), encoding="utf-8") return str(p) def tool(script, *args): r = subprocess.run([sys.executable, str(TOOLS / script), *args], capture_output=True, text=True, env=ENV) return r.returncode, r.stdout + r.stderr def ha_apps_info(options): """What `ha apps info --raw-json` prints: the options sit under data.options.""" text = {"en": ("Setpoint entity", "The ESPHome number carrying the setpoint."), "nl": ("Setpoint-entiteit", "Het ESPHome-getal met het setpoint.")} return {"result": "ok", "data": { "name": "GoodWe RS485 Controller", "slug": "local_goodwe_controller", "version": "0.12.0", "version_latest": "0.12.0", "state": "started", "boot": "auto", "ingress": True, "hassio_api": True, "options": copy.deepcopy(options), "schema": [{"name": k, "required": True, "optional": False, "type": "string"} for k in options], "translations": {lang: {"configuration": { "setpoint_entity": {"name": name, "description": desc}, "inverters": {"name": "Inverters", "description": "Roster.", "fields": {"setpoint_entity": { "name": name, "description": f"{desc} (per unit)"}}}}} for lang, (name, desc) in text.items()}}} # --------------------------------------------------------------------------- # print("1. predeploy_check.py --options (SAFETY-60)") # --------------------------------------------------------------------------- # # A deployed version that differs from the shipped one, so only the options half decides. BARE = {"setpoint_entity": "number.goodwe_setpoint_w", "inverters": [], "max_w": 5000, "slew_w": 0, "gain": 0.6, "target_grid_w": -10, "deadband_w": 15, "meter_source": "off", "log_level": "info"} NOREPO = tempfile.mkdtemp(prefix="norepo.", dir=TMP) def gate(opts_path): return tool("predeploy_check.py", "--deployed-version", "0.0.0-test", "--repo", NOREPO, "--options", opts_path) code, out = gate(dump("bare.json", BARE)) print(f" bare options -> exit {code}") check("baseline: the bare options object is CLEARED (exit 0)", code == 0 and "CLEARED for deploy" in out) code, out = gate(dump("bare-bad.json", dict(BARE, slew_w=5000))) check("baseline: the bare object with slew_w 5000 is REFUSED by the validator", code == 1 and "(slew_w) is 5000 W" in out) for label, doc in (("wrapped", ha_apps_info(BARE)), ("wrapped, slew_w 5000", ha_apps_info(dict(BARE, slew_w=5000)))): code, out = gate(dump("wrapped.json", doc)) print(f" {label} -> exit {code}") check(f"the `ha apps info` capture ({label}) is REFUSED, exit 1", code == 1 and "CLEARED" not in out) check("...naming the shape and the fix, not a validator rule", "0 of 2 top-level key(s) recognised" in out and ".data.options" in out and "VALIDATOR" not in out) from app.validate import actuator_writers, validate_startup_config # noqa: E402 _w = ha_apps_info(dict(BARE, slew_w=5000)) print(f" (for reference: the validator alone, handed that capture, finds " f"{len(validate_startup_config(_w))} problem(s) and " f"{len(actuator_writers(_w))} 'actuators' - the gate has to refuse it)") code, out = gate(dump("list.json", [BARE])) check("a list-wrapped options object is REFUSED, naming the type", code == 1 and "is a JSON list, not an object" in out) code, out = gate(dump("mostly-foreign.json", {"setpoint_entity": "number.x", "a": 1, "b": 2})) check("an object with fewer than half its keys recognised is REFUSED", code == 1 and "1 of 3 top-level key(s) recognised" in out) # --------------------------------------------------------------------------- # print("\n2. rehearse_roster.py --state/--before/--after (SAFETY-62)") # --------------------------------------------------------------------------- # sys.path.insert(0, str(TOOLS)) import rehearse_roster as RR # noqa: E402 STATE = FIX / "state-legacy.json" LIVE = ["--live-entities", str(FIX / "live-entities-recorded.txt"), "--reference", str(FIX / "announced-reference.txt")] def rehearse(state=str(STATE), before=str(FIX / "flat.yaml"), after=str(FIX / "roster.yaml"), extra=LIVE): return tool("rehearse_roster.py", "--state", state, "--before", before, "--after", after, *extra) code, out = rehearse() print(f" control -> exit {code}") check("baseline: the control run reports its 2 findings, exit 1", code == RR.EXIT_FINDING and "2 FINDING(S)" in out) code, out = rehearse(state=str(FIX / "state-fresh.json")) check("baseline: a fresh (comment-only) state file still runs clean, exit 0", code == RR.EXIT_OK and "NO FINDINGS" in out) raw = STATE.read_text(encoding="utf-8") state = json.loads(raw) cases = { "wrapped": (dump("s-wrapped.json", {"result": "ok", "data": state}), "0 of 2 top-level key(s) recognised"), "truncated by 20 bytes": (dump("s-trunc.json", text=raw[:-20]), "not a readable JSON state file"), "list-wrapped": (dump("s-list.json", [state]), "is a JSON list, not an object"), } for label, (path, words) in cases.items(): code, out = rehearse(state=path) print(f" state {label} -> exit {code}") check(f"state {label}: REFUSED with exit 2 and the reason", code == RR.EXIT_COULD_NOT_RUN and words in out and "NO FINDINGS" not in out) try: _auto = RR.load_state(dump("s-auto.json", {"auto": True})) except RR.Refused as err: _auto = str(err) check("a state holding only the Automatic toggle (`auto`, app/main.py) is " "recognised as a store", _auto == {"auto": True}) flat = RR.load_options(str(FIX / "flat.yaml")) roster = RR.load_options(str(FIX / "roster.yaml")) for side, kw in (("--before", {"before": dump("b.json", ha_apps_info(flat))}), ("--after", {"after": dump("a.json", ha_apps_info(roster))})): code, out = rehearse(**kw) print(f" {side} wrapped -> exit {code}") check(f"{side} as an `ha apps info` capture: REFUSED with exit 2", code == RR.EXIT_COULD_NOT_RUN and ".data.options" in out and "NO FINDINGS" not in out) code, out = rehearse(after=str(FIX / "roster-inverter-a.yaml"), extra=["--no-live-entities", "--reference", str(FIX / "announced-reference.txt")]) check("a clean --no-live-entities run is exit 3, not 0", code == RR.EXIT_DEGRADED and "NO FINDINGS" in out) # --------------------------------------------------------------------------- # print("\n3. alloc_sim.py - the split-invariance verdict reaches the exit code") # --------------------------------------------------------------------------- # spec = importlib.util.spec_from_file_location("_alloc_sim", TOOLS / "alloc_sim.py") AS = importlib.util.module_from_spec(spec) spec.loader.exec_module(AS) class _Run: """One rig run, reduced to what `splits()` reads.""" def __init__(self, grid): self.samples = [{"shares": [1000.0, 1000.0]}] self._grid = grid def grids(self): return self._grid def targets(self): return [2000.0] * len(self._grid) class _Sitesim: def load_addon(self, *_): return None def alloc_main(same): """`main(--splits)` with the rig stubbed; the real `splits()` grades the traces.""" seen = [] def fake(_sitesim, _control, mode, *a, **kw): seen.append(mode) return _Run([0.0, 1.0] if same or len(seen) == 1 else [0.0, 2.0]) AS.find_sim = lambda: Path("/nonexistent-rig") AS.load = lambda _sim: (None, _Sitesim(), None, None) AS.run_site_feedback = fake os.environ["GOODWE_ADDON_REPO"] = str(HERE.parent) sys.argv = ["alloc_sim.py", "--splits"] buf = io.StringIO() with contextlib.redirect_stdout(buf): rc = AS.main() return rc, buf.getvalue() rc, out = alloc_main(same=True) check("baseline: identical traces HOLD, exit 0", rc == 0 and "split-invariance HOLDS" in out) rc, out = alloc_main(same=False) print(f" differing traces -> exit {rc}") check("differing traces FAIL, exit 1", rc == 1 and "split-invariance FAILS" in out) # --------------------------------------------------------------------------- # print("\n4. decode_bus_capture.py - a file that is not a capture") # --------------------------------------------------------------------------- # code, out = tool("decode_bus_capture.py", str(CAPTURE)) check("baseline: the committed capture decodes, exit 0", code == 0 and "writes 1200" in out and "REFUSED" not in out) junk = dump("junk.log", text="hello\nthis is not a capture\n") code, out = tool("decode_bus_capture.py", junk) print(f" junk -> exit {code}") check("a junk file is REFUSED, exit 1", code == 1 and "REFUSED" in out and "0 of 2 line(s) decoded" in out) code, out = tool("decode_bus_capture.py", dump("empty.log", text="")) check("an empty file is REFUSED, exit 1", code == 1 and "0 of 0 line(s) decoded" in out) code, out = tool("decode_bus_capture.py", str(CAPTURE), junk) check("several paths, as the docstring's glob passes them: each is decoded and " "one bad file makes the run exit 1", code == 1 and f"== {CAPTURE}" in out and f"== {junk}" in out) code, out = tool("decode_bus_capture.py", str(CAPTURE), str(CAPTURE)) check("...and two good files exit 0", code == 0 and out.count("REFUSED") == 0) shutil.rmtree(TMP, ignore_errors=True) print() if fails: print(f"FAILED: {len(fails)}") for f in fails: print(f" - {f}") sys.exit(1) print("all tool-input checks passed")