glenn schrooyenandClaude Opus 5.5
2c516a032c
Validator refusals name the setting by its label and key, say what is wrong and what to change, and keep the rejected value and limit. Option names and descriptions are short in en and nl; citations moved to DOCS "For installers" and the test_translations citation check now pins the instruction on the tab and the record in DOCS. No rule, key, schema, default or exit code changed. Allowlist 167 -> 43 (log lines only). Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01LyR88rXETvwhTgx6yHE39p
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")
|