goodwe-addon/goodwe_controller/test_tool_inputs.py

"""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)

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")