goodwe-addon/goodwe_controller/test_selfconfig.py

"""Runnable check for the ingress configuration surface.  `python3 test_selfconfig.py`

UX-02. The panel edits, validates and saves the add-on's own options. Four of
its properties are the ones that could hurt somebody, and they are what this
file spends its length on:

  1. AN INVALID BLOB IS NEVER WRITTEN. Not "warned about": not written. An
     invalid blob plus a restart is exit 78, and an add-on that is not running
     is not serving the panel either - the operator's way back would be the
     Supervisor UI, which is the surface this card exists because it is hard to
     read. Part 2 attempts the write and counts what the Supervisor received.
  2. A BLOB THAT MOVED UNDERNEATH IS NEVER CLOBBERED. There are two config
     surfaces and only one blob. Part 3 simulates the other one writing between
     this page's read and its save.
  3. EVERY MESSAGE COMES FROM THE SHIPPED VALIDATOR. Part 6 adds a NEW RULE to
     a COPY of validate.py, serves the page out of that copy in a subprocess,
     and watches the new sentence arrive on the field it names - with every UI
     file byte-identical, which is asserted by hashing them.
  4. NOTHING HERE COMMANDS AN INVERTER. Part 7 lists what the panel exposes and
     compares it with what `/action` already offered.

⚠️ MUTATING THE MODULE THIS FILE GRADES? PURGE `__pycache__` FIRST - or run
`tools/mutation_run.sh`, which does. `python3 -B` alone does NOT save you: it
stops bytecode being WRITTEN, not a stale one being READ. That is TEST-15, and
part 6 below builds a mutated tree on purpose, so it purges its own copy.
"""

import asyncio
import hashlib
import io
import json
import os
import re
import shutil
import subprocess
import sys
import tempfile

from aiohttp import ClientSession, web as aioweb

from app import config_page, configui, selfconfig, web
from app.validate import validate_startup_config

HERE = os.path.dirname(os.path.abspath(__file__))
fails = []


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


def run(coro):
    return asyncio.run(coro)


# --------------------------------------------------------------------------- #
# Fixtures.
#
# ⚠️ `SCHEMA` AND `TRANSLATIONS` ARE FIXTURES OF SOMEBODY ELSE'S RESPONSE SHAPE,
# and they are graded against the real thing somewhere else on purpose.
# `tools/supervisor_selfapi_probe.py` builds the same two structures with
# Supervisor 2026.08.0's OWN `UiOptions` and `_read_app_translations` and runs
# this repository's `field_model()` over them - §C and §E. This file cannot: the
# CI image is Alpine + python3 + aiohttp + paho and has neither the Supervisor
# nor PyYAML. So the shapes below are small, and the claim that they are the
# shapes that arrive is the probe's, not this file's.
# --------------------------------------------------------------------------- #
SCHEMA = [
    {"name": "setpoint_entity", "required": True, "type": "string"},
    {"name": "max_w", "lengthMin": 100.0, "lengthMax": 5000.0,
     "required": True, "type": "integer"},
    {"name": "meter_source", "required": True, "type": "select",
     "options": ["off", "ha_dsmr"]},
    {"name": "log_level", "required": True, "type": "select",
     "options": ["info", "trace"]},
    {"name": "batt_invert", "required": True, "type": "boolean"},
    {"name": "inverters", "type": "schema", "optional": True, "multiple": True,
     "schema": [
         {"name": "name", "required": True, "type": "string"},
         {"name": "setpoint_entity", "required": True, "type": "string"},
         {"name": "max_w", "lengthMin": 100.0, "lengthMax": 5000.0,
          "required": True, "type": "integer"},
         {"name": "bms_soc_floor", "lengthMin": 0.0, "lengthMax": 50.0,
          "required": True, "type": "integer"}]},
]
TRANSLATIONS = {
    "en": {"configuration": {
        "setpoint_entity": {"name": "Hardware · Inverter setpoint entity",
                            "description": "Where the setpoint is written."},
        "max_w": {"name": "Limits · Inverter rating (W)",
                  "description": "The ceiling this site may command."},
        "meter_source": {"name": "Meter · Source",
                         "description": "Where the grid reading comes from."},
        "log_level": {"name": "Advanced · Log level",
                      "description": "How much the add-on writes to its log."},
        "batt_invert": {"name": "Hardware · Battery power sign is reversed",
                        "description": "Only if your sensor is backwards."},
        "inverters": {"name": "Hardware · Inverter list",
                      "description": "Leave empty for one inverter.",
                      "fields": {
                          "name": {"name": "Name"},
                          "setpoint_entity": {"name": "Setpoint entity"},
                          "max_w": {"name": "Rating (W)"},
                          "bms_soc_floor": {"name": "Battery cut-out (%)"}}}}},
    "nl": {"configuration": {
        "max_w": {"name": "Grenzen · Omvormervermogen (W)",
                  "description": "Het plafond voor deze site."}}},
}


def base(**over):
    """An options blob that STARTS. Everything else is this, spoilt.

    The same fixture shape as test_validate.py's `base()`, for the same reason
    it gives: a check that is measuring a broken config has to be able to show
    that the unbroken one is accepted, or it is measuring nothing.
    """
    o = {"setpoint_entity": "number.goodwe_setpoint", "max_w": 2000,
         "soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
         "meter_source": "off", "log_level": "info", "batt_invert": False,
         "inverters": [],
         "maintenance_charge_w": 2000, "maintenance_discharge_w": 2000}
    o.update(over)
    return o


class FakeSupervisor:
    """The Supervisor's stored options, and a count of what reached them.

    ⚠️ IT COUNTS WRITES, NOT CALLS. "Nothing was written" is the property parts
    2 and 3 are about, and a fake that only recorded that `write_options` was
    reached could not tell a refusal from a write that failed afterwards.
    """

    def __init__(self, options=None):
        self.options = dict(options or base())
        self.writes = []
        self.restarts = 0

    async def info(self):
        return {"options": dict(self.options), "schema": SCHEMA,
                "translations": TRANSLATIONS, "state": "started"}

    async def write_options(self, options):
        self.writes.append(dict(options))
        self.options = dict(options)

    async def restart(self, confirm):
        if confirm != selfconfig.RESTART_CONFIRM:
            raise selfconfig.NotConfirmed("not confirmed")
        self.restarts += 1


class FakeHass:
    """`hass.limits()` and nothing else. It cannot write: that is the point."""

    def __init__(self, ranges=None):
        self.ranges = ranges or {}
        self.reads = []

    async def limits(self, entity):
        self.reads.append(entity)
        return self.ranges.get(entity, (None, None, None))


class FakeController:
    def __init__(self, opts=None, hass=None):
        self.o = dict(opts or base())
        self.hass = hass or FakeHass()
        self.inverters = []
        self.events = []

    def log_event(self, text, hint=None):
        self.events.append(text)

    def status_payload(self):
        return {"banner": "", "level": "ok", "rows": [], "checks": []}

    async def handle_action(self, what):
        raise AssertionError("the configuration surface must never reach "
                             f"handle_action ({what!r})")


# ⚠️ THE ANCHOR LIST IS EVERY OPTION KEY config.yaml DECLARES, read off the
# file with the shipped parser rather than typed here. That is what the running
# panel anchors against too: `GET /addons/self/info` returns the MERGED options,
# so every key is present in the blob the page posts back. A test that anchored
# against a small fixture's keys would report a rule as "unanchored" purely
# because its fixture did not mention the key - which is what the first draft
# of part 4 did, for the price rules.
sys.path.insert(0, os.path.join(HERE, "tools"))
import translations_lint as _tl                                # noqa: E402

ALL_KEYS, _NESTED = _tl.config_surface(
    io.open(os.path.join(HERE, "config.yaml"), encoding="utf-8").read())


# --------------------------------------------------------------------------- #
print("1. the compare-and-swap token")
_a = base()
_b = dict(reversed(list(base().items())))       # same content, other key order
check("two blobs with the same content have the same fingerprint, whatever "
      "order their keys were built in - a token that changed on a dict "
      "reordering would refuse every save for no reason",
      selfconfig.fingerprint(_a) == selfconfig.fingerprint(_b))
check("one changed value changes it",
      selfconfig.fingerprint(base()) != selfconfig.fingerprint(base(max_w=2001)))
check("...and so does a changed ROSTER ORDER, because roster order is meaning",
      selfconfig.fingerprint(base(inverters=[{"name": "a"}, {"name": "b"}]))
      != selfconfig.fingerprint(base(inverters=[{"name": "b"}, {"name": "a"}])))
check("it is a hex digest and not the blob - nothing about the configuration "
      "leaves through it",
      re.fullmatch(r"[0-9a-f]{64}", selfconfig.fingerprint(base())) is not None)


# --------------------------------------------------------------------------- #
print()
print("2. AN INVALID BLOB IS REFUSED, AND NOTHING IS WRITTEN")
sup = FakeSupervisor()
fp = selfconfig.fingerprint(sup.options)
bad = base(setpoint_entity="")          # rule 2: a controller with no actuator
refused = None
try:
    run(selfconfig.save(sup, bad, fp))
except selfconfig.SaveRefused as err:
    refused = err
check("the save raises SaveRefused(invalid)",
      refused is not None and refused.kind == "invalid")
check("⚠️ AND THE SUPERVISOR RECEIVED NOTHING. Not a write that was rolled "
      "back, not a write of the old blob - no write at all",
      sup.writes == [])
check("the stored options are byte-identical to what they were",
      selfconfig.fingerprint(sup.options) == fp)
check("...and the refusal carries the validator's OWN sentence, unedited",
      refused.problems == validate_startup_config(bad, None, None))
check("...which is the sentence that names the key and says what it costs",
      any("(setpoint_entity) is empty" in p for p in refused.problems))

_good = base()
res = run(selfconfig.save(sup, _good, fp))
check("a config that STARTS is saved - or nothing above means anything",
      len(sup.writes) == 1 and sup.writes[0] == _good)
check("...and the caller gets the new token back, so the next save can "
      "compare against what it just wrote",
      res["fingerprint"] == selfconfig.fingerprint(_good))


# --------------------------------------------------------------------------- #
print()
print("2b. ⚠️ THE TWO HALVES OF THE max_w FIX, EACH PINNED SO IT CANNOT BE UNDONE")
# Review found that BOTH halves reverted with the whole 24-file suite still
# green - including the half that turns a crash into a SILENT ACCEPTANCE, which
# is the more dangerous of the two and was a wrong first attempt at this very
# fix. A fix nothing grades is a fix that comes back. MUTATION.md habit 4.
from app.inverter import _flat_max_w, build_roster        # noqa: E402
from app.validate import _number                          # noqa: E402

# ⚠️ THIS CHECK EXISTS BECAUSE THE OBVIOUS ONE HAS A HOLE. "problems_and_ranges
# does not raise" is ALSO satisfied by the backstop swallowing the exception and
# returning [] - which is the silent-acceptance bug half two exists to prevent.
# Measured: with BOTH halves reverted, every max_w check below reported ok and
# only the forced-boom pins went red. So half one is pinned HERE, on the
# composition itself, where the backstop cannot answer for it.
_raised = None
try:
    build_roster(base(max_w=""), None)
except Exception as exc:                                   # noqa: BLE001
    _raised = exc
check("⭐ THE COMPOSITION ITSELF TOLERATES A CLEARED max_w - not merely the "
      "backstop above it. `validate.py`'s `_number()` reads '' as the default, "
      "so rule 9 sees 2000 and is content; a bare `float('')` in the flat path "
      "raised instead, inverting inverter.py's own rule that a shape rule 9 "
      "tolerates must be composable",
      _raised is None)

# HALF ONE: composition tolerates exactly what rule 9 tolerates.
for raw in ("", None, 0, False, True, -1, "2000", " 2000 ", "2e3", 1e18):
    o = base(max_w=raw)
    expect = _number(o, "max_w", 2000.0)
    if expect is not None:
        check(f"_flat_max_w agrees with _number() for max_w={raw!r}",
              _flat_max_w(o) == expect)

_raised = None
try:
    _probs, _ = run(selfconfig.problems_and_ranges(base(max_w=""), None))
except Exception as exc:                                   # noqa: BLE001
    _raised, _probs = exc, None
check("⚠️ A CLEARED max_w BOX ANSWERS WITH A VERDICT, NOT AN EXCEPTION. It "
      "used to raise ValueError from inside Controller.__init__, which the "
      "panel served as HTTP 500 on a non-JSON body and the page rendered as "
      "'cannot reach the add-on' - untrue, for the most prominent limit "
      "field's most likely transient state",
      _raised is None)
check("...and it agrees with the validator, which reads '' as the default",
      _probs == list(validate_startup_config(base(max_w=""), None, None)))

_raised = None
try:
    _probs, _ = run(selfconfig.problems_and_ranges(base(max_w="abc"), None))
except Exception as exc:                                   # noqa: BLE001
    _raised, _probs = exc, None
check("an UNPARSEABLE max_w answers with the validator's own sentence",
      _raised is None and any("is not a number" in p for p in (_probs or [])))

# HALF ONE, THE PART THAT MAKES IT SAFE RATHER THAN MERELY NOT-CRASHING:
# clearing the box on the LIVE shape must FAIL CLOSED, not silently halve the
# rail. `unit_caps_indexed` reads the flat rail through the same `_number`, so
# the rail collapses to 2000 in lockstep and rule 10 catches the collapse.
_live = base(max_w=5000, slew_w=400)
_probs, _ = run(selfconfig.problems_and_ranges(dict(_live), None))
check("the live shape (5000 W rail, slew_w 400) starts", _probs == [])
_probs, _ = run(selfconfig.problems_and_ranges(dict(_live, max_w=""), None))
check("⚠️ CLEARING max_w ON THE LIVE SHAPE IS REFUSED - rule 10 catches the "
      "rail collapsing to the 2000 default under an explicit 400 W slew, so "
      "the box cannot be emptied into a quietly halved rail",
      any("for a 2000 W site" in p for p in _probs))

# HALF TWO: the backstop cannot report "no problems" for a config it could not
# compose. Returning [] there enables Save - a crash became a green.
_orig_roster = selfconfig.build_roster


def _boom(*_a, **_kw):
    raise ValueError("forced boom")


selfconfig.build_roster = _boom
try:
    _probs, _ = run(selfconfig.problems_and_ranges(base(), None))
finally:
    selfconfig.build_roster = _orig_roster
check("⚠️ THE BACKSTOP NEVER RETURNS AN EMPTY LIST. A roster that cannot be "
      "composed while the validator names no reason is itself the finding - "
      "returning [] would turn a crash into a silent acceptance and enable "
      "Save, which is worse than the 500 it replaced",
      _probs != [] and len(_probs) == 1)
check("...and it says so in words, naming the exception",
      any("could not be composed" in p and "forced boom" in p
          for p in _probs))
check("...and the patch was undone, so nothing below inherits it",
      selfconfig.build_roster is _orig_roster)


# --------------------------------------------------------------------------- #
print()
print("3. THE OTHER SURFACE WRITES WHILE THIS PAGE IS OPEN")
# ⚠️ SIMULATED AT THE EXACT MOMENT IT MATTERS. The Supervisor's Configuration
# tab is not removable, so this is not a hypothetical race - it is the normal
# way two people, or one person on two devices, use this add-on. The blob is
# changed AFTER the page's read and BEFORE its save.
# ⚠️ EVERY BLOB IN THIS SECTION MUST BE ONE THAT STARTS. The gate validates
# BEFORE it compares, so a fixture that is also invalid would be refused for
# the wrong reason and this section would be measuring rule 6 instead of the
# compare-and-swap. (It was, in the first draft: `max_w: 1800` beneath the
# fixture's `maintenance_charge_w: 2000` is a genuine refusal.)
sup = FakeSupervisor(base(max_w=2000))
page_fp = selfconfig.fingerprint(sup.options)          # what the page loaded
sup.options = base(max_w=2500)                         # the other tab saves
mine = base(max_w=3000, log_level="trace")             # what the operator typed
refused = None
try:
    run(selfconfig.save(sup, mine, page_fp))
except selfconfig.SaveRefused as err:
    refused = err
check("the save raises SaveRefused(stale)",
      refused is not None and refused.kind == "stale")
check("⚠️ AND AGAIN NOTHING WAS WRITTEN - the other surface's value stands, "
      "and the operator's edits were not silently thrown at it",
      sup.writes == [] and sup.options["max_w"] == 2500)
check("the refusal NAMES the keys that differ, so the page can print them "
      "rather than saying 'something changed'",
      refused.changed == ["log_level", "max_w"])
check("...and carries what is stored, so the operator decides in front of the "
      "difference rather than through it",
      refused.stored["max_w"] == 2500)
_after = run(selfconfig.save(sup, mine, selfconfig.fingerprint(sup.options)))
check("saving again against the CURRENT token succeeds - the refusal is a "
      "gate, not a lock",
      len(sup.writes) == 1 and sup.options["max_w"] == 3000
      and _after["fingerprint"] == selfconfig.fingerprint(mine))

# ⚠️ AND THERE IS NO WAY ROUND IT. A caller with no token cannot save.
sup = FakeSupervisor()
try:
    run(selfconfig.save(sup, base(max_w=2400), ""))
    check("an empty token is refused", False)
except selfconfig.SaveRefused as err:
    check("an empty token is refused as stale rather than treated as 'do not "
          "check' - there is no force door in this module",
          err.kind == "stale" and sup.writes == [])


# --------------------------------------------------------------------------- #
print()
print("4. WHERE A REFUSAL LANDS ON THE PAGE")
# ⚠️ MEASURED, NOT CLAIMED. The anchoring is name occurrence: a sentence is
# printed under every option key it spells. Rather than assert a coverage
# figure, this runs the SHIPPED validator on a spread of broken configs and
# prints what anchored and what did not.
_cases = [
    ("no actuator", base(setpoint_entity="")),
    ("max_w over the device", base(max_w=4000)),
    ("charge rate above the ceiling", base(maintenance_charge_w=3000)),
    ("sign threshold at the ceiling", base(sign_disagree_w=2000)),
    ("two writers on one actuator",
     base(inverters=[{"name": "a", "setpoint_entity": "number.goodwe_setpoint",
                      "max_w": 2000, "bms_soc_floor": 10}])),
    ("a price source that cannot mean what it says",
     base(price_provider="manual_csv")),
]
_ranges = {"number.goodwe_setpoint": (-3000.0, 3000.0, 1.0)}
_total = _anchored = 0
_orphans = []
for title, opts in _cases:
    probs = run(selfconfig.problems_for(opts, FakeHass(_ranges)))
    by_key, un = configui.problems_by_field(probs, ALL_KEYS)
    _total += len(probs)
    _anchored += len(probs) - len(un)
    _orphans += [(title, p[:70]) for p in un]
    print(f"    {title:34s} {len(probs)} problem(s), "
          f"{len(probs) - len(un)} beside a field")
check(f"every one of the {_total} refusals these six broken configs produce "
      f"lands beside at least one of the fields it is about. Unanchored: "
      f"{_orphans}", _total > 0 and not _orphans)

_probs = run(selfconfig.problems_for(base(max_w=4000), FakeHass(_ranges)))
_by, _un = configui.problems_by_field(_probs, ALL_KEYS)
check("the device-range rule is REACHED from the panel - the setpoint entity's "
      "own range is read, so rule 1 fires here exactly as it does at startup",
      any("Lower the rating to" in p for p in _probs))
check("...and it lands on `max_w`, which is the field the operator has to "
      "change", "max_w" in _by)
# ⚠️ THE COLLISION, CHECKED HERE AND NOT ONLY IN THE PROBE. `sign_disagree_w`
# is a suffix of both `unit_sign_disagree_w` and `site_sign_disagree_w` - three
# real options, two of them SAFETY-30's. A plain substring test would print
# every per-unit refusal next to the SITE field as well, which is a message
# about unit-0 beside a box that has nothing to do with it. CI runs this file
# and not the probe, so the assertion has to live here too.
_collide = [k for k in ALL_KEYS
            if any(o != k and o.endswith("_" + k) for o in ALL_KEYS)]
check(f"config.yaml really does declare keys that are suffixes of other keys, "
      f"so the boundary matching is load-bearing: {_collide}", bool(_collide))
check("...and a sentence naming `unit_sign_disagree_w` is NOT also filed under "
      "`sign_disagree_w`",
      sorted(configui.problems_by_field(
          ["unit_sign_disagree_w is 3000 W but unit-0's max_w is 2000 W"],
          ALL_KEYS)[0]) == ["max_w", "unit_sign_disagree_w"])
check("a cross-field refusal is filed under EVERY key it names, because a "
      "cross-field rule is about all of its fields",
      len(configui.problems_by_field(
          ["max_w is 8000 W but number.x accepts at most 5000 W"],
          ["max_w", "setpoint_entity"])[0]) == 1)
check("the panel shows no refusal at all for a config that starts",
      run(selfconfig.problems_for(base(), FakeHass(_ranges))) == [])

# ⚠️ AND WHEN A DEVICE DOES NOT ANSWER, THE PAGE SAYS WHICH ONE. The two
# device-range rules are then skipped - here and at startup, identically - so a
# green page would be green for a reason the operator cannot see.
_p, _unread = run(selfconfig.problems_and_ranges(base(max_w=4000), FakeHass()))
check("with the setpoint device silent, rule 1 does not fire and the entity "
      "that did not answer is named, so the page can say what was skipped",
      _p == [] and _unread == ["number.goodwe_setpoint"])
_p2, _unread2 = run(selfconfig.problems_and_ranges(base(max_w=4000),
                                                   FakeHass(_ranges)))
check("...and with the same device answering, the same config IS refused - "
      "which is what makes the line above a skip rather than a pass",
      any("Lower the rating to" in p for p in _p2) and _unread2 == [])
check("the note is about the config UNDER TEST, not about the running roster: "
      "a newly typed entity that does not answer is the one named",
      run(selfconfig.problems_and_ranges(
          base(setpoint_entity="number.just_typed"), FakeHass(_ranges)))[1]
      == ["number.just_typed"])


# --------------------------------------------------------------------------- #
print()
print("5. THE UI OWNS NO COPY OF ANY RULE, LABEL OR KEY")
_page = io.open(os.path.join(HERE, "app", "config_page.py"),
                encoding="utf-8").read()
_ui_files = {name: io.open(os.path.join(HERE, "app", name),
                           encoding="utf-8").read()
             for name in ("config_page.py", "configui.py", "web.py")}
# The option keys config.yaml declares, read off the file rather than typed
# here - a hand-written list would go stale in the direction that hides a
# copy.
# `name` and `max_w` also occur as ordinary English/JS words or as model field
# names, so the test is on the SCRIPT half of the page and on keys that could
# only be there deliberately.
_script = _page.split("<script>", 1)[1]
_leaked = sorted(k for k in ALL_KEYS
                 if len(k) > 6 and re.search(r"[^A-Za-z0-9_]" + k + r"[^A-Za-z0-9_]",
                                             _script))
check(f"the page's script names NOT ONE option key - so it cannot be rendering "
      f"a field list of its own. Leaked: {_leaked}", not _leaked)
_phrases = ["Lower the rating to", "cannot control the inverter",
            "point at the same inverter entity", "(setpoint_entity) is empty",
            "accepts at most"]
_copied = sorted(p for p in _phrases
                 for f, text in _ui_files.items() if p in text)
check(f"no phrase out of a validate.py refusal appears anywhere in the three "
      f"UI files - the sentences are fetched, never written here. Copied: "
      f"{_copied}", not _copied)
# ⚠️ THIS ONE HAS ALREADY EARNED ITS KEEP. The suffix map shipped with
# `("_hour", "h (0-23)")` in it - a RANGE, next to a field, in a second place
# from config.yaml's `int(0,23)`. It is exactly the drift this card forbids,
# in the one corner of the panel that is allowed to print anything of its own.
check("no unit string contains a digit, so the unit map cannot be stating a "
      "bound, a default or a relation - config.yaml is the only place a range "
      "is written and the schema is where the form gets it from",
      all(not any(ch.isdigit() for ch in unit)
          for _suffix, unit in configui._UNITS))
check("...and no entry of it is a whole option key - it is a suffix map, so it "
      "cannot become a per-key opinion by accident",
      not [suf for suf, _u in configui._UNITS if suf.lstrip("_") in ALL_KEYS])

# ⚠️ RELATIVE URLS, WHICH IS WHAT INGRESS REQUIRES. An absolute path works
# perfectly on a laptop and 404s in the field, so it cannot be left to review.
_abs = re.findall(r"""(?:fetch\(|href=)['"](/[^'"]*)['"]""",
                  _page + _ui_files["web.py"])
check(f"no absolute URL in either page - every fetch and every link is "
      f"relative, because ingress serves this under a generated path prefix. "
      f"Absolute: {_abs}", not _abs)
check("the page declares a viewport, which is the whole of 'readable on the "
      "phone you are holding at the inverter'",
      'name="viewport"' in _page and "max-width:760px" in _page)


# --------------------------------------------------------------------------- #
print()
print("6. ⚠️ A RULE ADDED TO validate.py APPEARS IN THE UI WITH NO UI CHANGE")
# THE ACCEPTANCE CRITERION, RUN RATHER THAN ARGUED. A copy of this package is
# made, ONE new rule is inserted into validate.py's `validate_startup_config()`,
# and the real HTTP route is served out of that copy in a subprocess. If the
# page carried any list of known rules, the new sentence would not arrive.
SENTINEL = ("SENTINEL-UX02: log_level is set to 'trace', which this rule "
            "refuses so that a check can watch it arrive.")
_INSERT = ('    if str(opts.get("log_level", "")) == "trace":\n'
           f'        problems.append({SENTINEL!r})\n')

_probe = r'''
import asyncio, json, os, sys
sys.path.insert(0, os.environ["TREE"])
from aiohttp import ClientSession, web as aioweb
from app import web

class H:
    session = None
    token = ""
    async def limits(self, e): return (None, None, None)
class C:
    o = json.loads(os.environ["OPTS"]); hass = H(); inverters = []
    def log_event(self, t, hint=None): pass
    def status_payload(self): return {}
    async def handle_action(self, w): raise AssertionError("no")

async def main():
    app = web.build_app(C())
    runner = aioweb.AppRunner(app); await runner.setup()
    site = aioweb.TCPSite(runner, "127.0.0.1", 0); await site.start()
    port = site._server.sockets[0].getsockname()[1]
    async with ClientSession() as s:
        async with s.post(f"http://127.0.0.1:{port}/config/validate",
                          json={"options": json.loads(os.environ["OPTS"])}) as r:
            print(json.dumps(await r.json()))
    await runner.cleanup()

asyncio.run(main())
'''

_tmp = tempfile.mkdtemp(prefix="ux02-mutant-")
try:
    _tree = os.path.join(_tmp, "goodwe_controller")
    shutil.copytree(os.path.join(HERE, "app"), os.path.join(_tree, "app"))
    # ⚠️ PURGE THE COPY'S BYTECODE. MUTATION.md habit: `python3 -B` stops
    # bytecode being WRITTEN, never a stale one being READ, and a copytree
    # brings any `__pycache__` along with it.
    _purged = 0
    for root, dirs, files in os.walk(_tree):
        for d in list(dirs):
            if d == "__pycache__":
                shutil.rmtree(os.path.join(root, d))
                dirs.remove(d)
                _purged += 1
    print(f"    purged {_purged} __pycache__ director(y|ies) from the copy; "
          f"{sum(1 for r, _d, fs in os.walk(_tree) for f in fs if f.endswith('.pyc'))}"
          " .pyc left")

    _vpath = os.path.join(_tree, "app", "validate.py")
    _src = io.open(_vpath, encoding="utf-8").read()
    _at = _src.index("def validate_startup_config")
    _anchor = "    problems: list[str] = []\n"
    _cut = _src.index(_anchor, _at) + len(_anchor)
    io.open(_vpath, "w", encoding="utf-8").write(
        _src[:_cut] + _INSERT + _src[_cut:])
    check("the mutated copy differs from the shipped tree in validate.py, and "
          "in validate.py ONLY - so anything the page gains is the rule's",
          [name for name in sorted(os.listdir(os.path.join(HERE, "app")))
           if name.endswith(".py")
           and hashlib.sha256(io.open(os.path.join(HERE, "app", name), "rb")
                              .read()).hexdigest()
           != hashlib.sha256(io.open(os.path.join(_tree, "app", name), "rb")
                             .read()).hexdigest()] == ["validate.py"])

    _env = dict(os.environ, TREE=_tree, OPTS=json.dumps(base(log_level="trace")),
                PYTHONDONTWRITEBYTECODE="1")
    _out = subprocess.run([sys.executable, "-B", "-c", _probe], env=_env,
                          capture_output=True, text=True, timeout=120)
    if _out.returncode != 0:
        print(_out.stdout[-2000:])
        print(_out.stderr[-2000:])
    check("the page's own /config/validate route answers out of the mutated "
          "tree", _out.returncode == 0)
    _payload = json.loads(_out.stdout.strip().splitlines()[-1]) if _out.stdout else {}
    check("⚠️ THE NEW RULE'S SENTENCE IS ON THE PAGE, and no file under app/ "
          "other than validate.py was touched to put it there",
          SENTINEL in (_payload.get("problems") or []))
    check("...and it is filed under `log_level`, the field it names, by the "
          "same anchoring that files every other rule - so a new rule arrives "
          "in the right place rather than in a catch-all",
          SENTINEL in (_payload.get("by_key") or {}).get("log_level", []))
    check("...while the SHIPPED tree does not refuse that config at all, which "
          "is what makes the line above evidence about the rule rather than "
          "about the fixture",
          run(selfconfig.problems_for(base(log_level="trace"), None)) == [])
finally:
    shutil.rmtree(_tmp, ignore_errors=True)


# --------------------------------------------------------------------------- #
print()
print("7. THE PANEL'S REACH INTO THE INVERTER IS UNCHANGED")
_app = web.build_app(FakeController(), FakeSupervisor())
_routes = sorted({(r.method, r.get_info().get("path"))
                  for r in _app.router.routes() if r.method != "HEAD"})
print("    " + "  ".join(f"{m} {p}" for m, p in _routes))
check("the configuration surface adds five routes and UX-13 one GET, and "
      "none touches the three that were there",
      _routes == [("GET", "/"), ("GET", "/config"), ("GET", "/config/model"),
                  ("GET", "/ha/entities"), ("GET", "/status"),
                  ("POST", "/action"), ("POST", "/config/restart"),
                  ("POST", "/config/save"), ("POST", "/config/validate")])
_ents_src = io.open(os.path.join(HERE, "app", "haentities.py"),
                    encoding="utf-8").read()
check("UX-13's entity list only reads: haentities.py calls hass.states() and "
      "no service, setter or action",
      "hass.states()" in _ents_src
      and not re.search(r"\.(call|set_number|handle_action|post)\(", _ents_src))
_selfconfig_src = io.open(os.path.join(HERE, "app", "selfconfig.py"),
                          encoding="utf-8").read()
for forbidden in ("set_number", "write_setpoint", "handle_action",
                  "command_site_zero", "sign_wd", "reset("):
    check(f"neither selfconfig.py nor the page can reach `{forbidden}`",
          forbidden not in _selfconfig_src
          and forbidden not in _ui_files["config_page.py"])
# ⚠️ ASSERTED ON THE CONSTANTS, NOT ON A GREP OF THE FILE. The module's own
# docstring quotes Supervisor's bypass regex, which names `security`, `update`
# and `options/validate`; a grep would find those and report a reach the code
# does not have. The three constants are what `_call()` can be given.
check("the ONLY Supervisor endpoints this module defines are self/info, "
      "self/options and self/restart",
      sorted(v for k, v in vars(selfconfig).items()
             if k.startswith("SELF_"))
      == ["http://supervisor/addons/self/info",
          "http://supervisor/addons/self/options",
          "http://supervisor/addons/self/restart"])


# --------------------------------------------------------------------------- #
print()
print("8. end to end, over real HTTP, against a fake Supervisor")


async def _serve(app):
    runner = aioweb.AppRunner(app)
    await runner.setup()
    site = aioweb.TCPSite(runner, "127.0.0.1", 0)
    await site.start()
    return runner, site._server.sockets[0].getsockname()[1]


async def _e2e():
    out = {}
    sup = FakeSupervisor(base(max_w=2000))
    ctrl = FakeController(base(max_w=1500),          # running != stored
                          FakeHass({"number.goodwe_setpoint":
                                    (-3000.0, 3000.0, 1.0)}))
    runner, port = await _serve(web.build_app(ctrl, sup))
    root = f"http://127.0.0.1:{port}"
    try:
        async with ClientSession() as s:
            async with s.get(f"{root}/config") as r:
                out["page"] = (r.status, await r.text())
            async with s.get(f"{root}/config/model?lang=nl") as r:
                out["model"] = await r.json()
            async with s.post(f"{root}/config/validate",
                              json={"options": base(setpoint_entity="")}) as r:
                out["invalid"] = await r.json()
            async with s.post(f"{root}/config/validate",
                              json={"options": base(integrator_max_w=3000)}
                              ) as r:
                out["warned"] = await r.json()
            # ⚠️ THE COUNTS ARE SNAPSHOTTED AT EACH STAGE. Asserting
            # `sup.writes == []` after the whole sequence would be asserting
            # about the LAST request, not this one - the first draft did, and
            # went red only because a later, legitimate save had happened.
            async with s.post(f"{root}/config/save",
                              json={"options": base(setpoint_entity=""),
                                    "fingerprint": out["model"]["fingerprint"]}
                              ) as r:
                out["save_invalid"] = (r.status, await r.json(),
                                       len(sup.writes))
            async with s.post(f"{root}/config/save",
                              json={"options": base(max_w=2400),
                                    "fingerprint": "not-the-token"}) as r:
                out["save_stale"] = (r.status, await r.json(), len(sup.writes))
            async with s.post(f"{root}/config/save",
                              json={"options": base(max_w=2400),
                                    "fingerprint": out["model"]["fingerprint"]}
                              ) as r:
                out["save_ok"] = (r.status, await r.json())
            async with s.post(f"{root}/config/restart", json={}) as r:
                out["restart_bare"] = (r.status, await r.json(), sup.restarts)
            async with s.post(f"{root}/config/restart",
                              json={"confirm": "restart"}) as r:
                out["restart_ok"] = (r.status, await r.json())
    finally:
        await runner.cleanup()
    out["sup"] = sup
    out["ctrl"] = ctrl
    return out


E = run(_e2e())
check("GET config serves the page", E["page"][0] == 200
      and "GoodWe - Configuration" in E["page"][1])
_m = E["model"]
check("the model carries the groups, the stored blob and its token",
      # `Grenzen`, not `Limits`: this request asked for nl, and nl.yaml
      # carries `max_w`. The three groups around it fall back to English per
      # key, which is the frontend's own rule and is asserted two checks below.
      [g["name"] for g in _m["groups"]] == ["Hardware", "Grenzen", "Meter",
                                            "Advanced"]
      and _m["options"]["max_w"] == 2000
      and _m["fingerprint"] == selfconfig.fingerprint(_m["options"]))
check("...and it says which stored settings are NOT what this add-on is "
      "running - which is how an edit made in the Configuration tab, or a save "
      "not yet restarted into, becomes visible here",
      _m["running_differs"] == ["max_w"] and _m["running"]["max_w"] == 1500)
check("...and it honours the browser's language per key, falling back to "
      "English for the keys nl.yaml does not carry - the frontend's own rule",
      _m["lang"] == "nl"
      and [f["label"] for g in _m["groups"] for f in g["fields"]
           if f["key"] == "max_w"] == ["Omvormervermogen (W)"]
      and [f["label_lang"] for g in _m["groups"] for f in g["fields"]
           if f["key"] == "meter_source"] == ["en"])
check("POST config/validate returns the validator's sentences, anchored",
      E["invalid"]["problems"]
      and "setpoint_entity" in E["invalid"]["by_key"])
check("UX-13: POST config/validate also returns the validator's warnings, "
      "beside no refusal",
      E["warned"]["problems"] == [] and [n["keys"] for n in
                                         E["warned"]["notes"]]
      == [["integrator_max_w"]] and E["warned"]["notes"][0]["tier"] == "warn")
check("a save of an invalid blob answers 422 and the Supervisor got nothing",
      E["save_invalid"][0] == 422
      and E["save_invalid"][1]["kind"] == "invalid"
      and E["save_invalid"][2] == 0)
check("a save with a stale token answers 409, writes nothing, and hands back "
      "the CURRENT token so the operator can save again in front of the "
      "difference",
      E["save_stale"][0] == 409
      and E["save_stale"][1]["kind"] == "stale"
      and E["save_stale"][1]["fingerprint"]
      == selfconfig.fingerprint(E["save_stale"][1]["stored"])
      and E["save_stale"][2] == 0)
check("a valid save with the right token is written, once",
      E["save_ok"][0] == 200 and len(E["sup"].writes) == 1
      and E["sup"].writes[0]["max_w"] == 2400)
check("...and it does NOT restart anything - applying is a separate, "
      "confirmed act, because it costs a control interruption",
      E["restart_bare"][0] == 400 and E["restart_bare"][2] == 0)
check("...which the confirmed request performs",
      E["restart_ok"][0] == 200)
check("the action log records both, so §15 can say who changed what",
      any("configuration saved" in e for e in E["ctrl"].events)
      and any("restart requested" in e for e in E["ctrl"].events))


# --------------------------------------------------------------------------- #
print()
print("9. the page says what a save costs BEFORE it is pressed")
for phrase in ("0 W on every inverter", "Saving does not restart anything",
               "Restart to apply"):
    check(f"the page states {phrase!r} in its own markup, not in a message "
          "that only appears afterwards", phrase in _page)
check("the restart confirmation names the cost again at the moment of pressing",
      "Yes, restart now" in _page
      and _page.count("0 W on every inverter") >= 2)
_flat = " ".join(_page.split())
check("and it says which surface is authoritative when the add-on is down",
      "If the add-on is not running, the Configuration tab is the only one of "
      "the two that still exists" in _flat)


# --------------------------------------------------------------------------- #
print()
print("10. the Supervisor conversation itself, over real HTTP")
# ⚠️ EVERYTHING ABOVE INJECTED A FAKE `selfcfg`, so none of it exercised the
# one piece that talks to somebody else's server: the headers, the request
# body's shape, the `{"result": "ok", "data": {...}}` envelope and what a
# refusal turns into. A gate whose transport is never run is a gate that has
# only been read.


async def _transport():
    seen = []

    async def h_info(request):
        seen.append(("info", request.headers.get("Authorization"), None))
        return aioweb.json_response({"result": "ok",
                                     "data": {"options": {"max_w": 2000}}})

    async def h_options(request):
        body = await request.json()
        seen.append(("options", request.headers.get("Authorization"), body))
        return aioweb.json_response({"result": "ok", "data": {}})

    async def h_restart(request):
        seen.append(("restart", request.headers.get("Authorization"), None))
        return aioweb.json_response({"result": "ok", "data": {}})

    async def h_denied(_request):
        return aioweb.json_response(
            {"result": "error", "message": "Missing API permission"},
            status=403)

    app = aioweb.Application()
    app.add_routes([aioweb.get("/addons/self/info", h_info),
                    aioweb.post("/addons/self/options", h_options),
                    aioweb.post("/addons/self/restart", h_restart),
                    aioweb.get("/denied", h_denied)])
    runner, port = await _serve(app)
    root = f"http://127.0.0.1:{port}"
    saved = (selfconfig.SELF_INFO, selfconfig.SELF_OPTIONS,
             selfconfig.SELF_RESTART)
    selfconfig.SELF_INFO = f"{root}/addons/self/info"
    selfconfig.SELF_OPTIONS = f"{root}/addons/self/options"
    selfconfig.SELF_RESTART = f"{root}/addons/self/restart"
    out = {}
    try:
        async with ClientSession() as session:
            cfg = selfconfig.SelfConfig(session, "TOKEN-123")
            out["info"] = await cfg.info()
            await cfg.write_options({"max_w": 2400})
            await cfg.restart(selfconfig.RESTART_CONFIRM)
            try:
                await cfg.restart("")
                out["unconfirmed"] = None
            except selfconfig.NotConfirmed as err:
                out["unconfirmed"] = str(err)
            try:
                await cfg._call("GET", f"{root}/denied")
                out["denied"] = None
            except selfconfig.SupervisorUnavailable as err:
                out["denied"] = str(err)
            try:
                await cfg._call("GET", f"{root}/nothing-here")
                out["missing"] = None
            except selfconfig.SupervisorUnavailable as err:
                out["missing"] = str(err)
    finally:
        (selfconfig.SELF_INFO, selfconfig.SELF_OPTIONS,
         selfconfig.SELF_RESTART) = saved
        await runner.cleanup()
    out["seen"] = seen
    return out


T = run(_transport())
check("`info()` unwraps the Supervisor's `{result, data}` envelope",
      T["info"] == {"options": {"max_w": 2000}})
check("every request carries the add-on's own SUPERVISOR_TOKEN as a bearer",
      [auth for _what, auth, _b in T["seen"]] == ["Bearer TOKEN-123"] * 3)
check("the options write goes to /addons/self/options as {'options': blob} - "
      "the shape `SCHEMA_OPTIONS` accepts, and the shape that REPLACES the "
      "stored user options",
      [(w, b) for w, _a, b in T["seen"] if w == "options"]
      == [("options", {"options": {"max_w": 2400}})])
check("the restart is a POST to /addons/self/restart and reaches it only when "
      "confirmed",
      [w for w, _a, _b in T["seen"]] == ["info", "options", "restart"]
      and T["unconfirmed"])
check("a Supervisor refusal is carried through with ITS message and its status "
      "rather than rewritten - the page shows what the Supervisor said",
      "403" in T["denied"] and "Missing API permission" in T["denied"])
check("...and so is a 404, so a path this add-on is not permitted is "
      "distinguishable from one that does not exist",
      "404" in T["missing"])


# --------------------------------------------------------------------------- #
print()
if fails:
    print(f"{len(fails)} FAILED: " + "; ".join(fails))
    sys.exit(1)
print("all checks passed")