glenn schrooyenandClaude Opus 5.5
b06f1fad5f
checks / unit (pull_request) Successful in 8m35s
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01BwRLw7eKYj2byVdwB6kbjU
263 lines
11 KiB
Python
263 lines
11 KiB
Python
"""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 <slug> --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")
|