goodwe-addon/goodwe_controller/test_defaults_pin.py

"""UX-04 - the defaults pin UX-01 AC 4 promised and never built.
`python3 -B test_defaults_pin.py`

WHAT UX-01 AC 4 ASKED FOR, IN ITS OWN WORDS: "every `schema:` key absent from
`options:` has exactly one defaults-table entry ... the table changed goes red
... and NO MODULE CARRIES ITS OWN LITERAL FALLBACK FOR A SCHEMA DEFAULT - an AST
check, not a grep."

WHAT SHIPPED. Clause one is VACUOUS: `options:` and `schema:` declare the same
71 keys, so "every schema key absent from options" quantifies over the empty set
and is true of any tree, including one with no defaults at all. Clause two -
"the table changed goes red" - does work, six times over, once per mirror in its
own suite. Clause three, THE ONE WITH THE CONTENT, was never written. The only
literal scan in this repository before this file is
`test_validate.py`'s `re.findall(r'"(maintenance_[a-z_]+)"...')`, which is a
grep, is restricted to eight `maintenance_*` keys and three named files, and
predates UX-01.

⚠️ AND A CHECK THAT CANNOT GO RED IS NOT A CHECK, so this file does not repeat
clause one's mistake. Every part below states the size of the set it quantifies
over and grades that size against a floor derived from the tree, and part 5
plants a literal in a synthetic module and REQUIRES the scanner to catch it, so
the red direction is exercised on every run rather than attested in a commit
message. MUTATION.md habit 4.

--------------------------------------------------------------------------- #
WHAT AN "IN-CODE LITERAL FALLBACK" IS, STRUCTURALLY, so a reader can argue with
the definition rather than guess it. It is a call in `app/` that takes a STRING
LITERAL naming a `schema:` key and, in the next positional slot, a CONSTANT:

    opts.get("max_w", 2000)          _number(opts, "max_w", 2000.0)

Two shapes, one rule, found by walking `ast.Call` nodes - not by matching text,
so `opts.get(\n    "max_w",\n    2000,\n)` and a renamed `opts` are found too and
a `max_w` inside a comment or a docstring is not.

TWO EXCLUSIONS, both structural and both narrow:

  * `getattr(obj, "max_w", 0.0)` reads an ATTRIBUTE off an object, not an option
    off the operator's blob. Its third argument is an absent-attribute guard and
    has no relationship to config.yaml. Excluded by function name.

  * THE EMPTY STRING. `opts.get("static_profile", "")` is Python's "this key is
    not stated", and every one of the six sites in `validate.py` immediately
    normalises it - `str(opts.get(k, "") or "").strip()`. It cannot be a stale
    copy of a NUMBER, which is the whole of the CFG-04 defect class, and
    rewriting those six to index a mirror would replace a sentence that reads
    "absent" with one that reads "the shipped default" at sites where absent is
    what is meant. Excluded by value. `0` is NOT excluded: `slew_w: 0` and
    `integrator_max_w: 0` are real shipped defaults with real meanings.

--------------------------------------------------------------------------- #
WHAT THIS FILE DOES *NOT* DO, said before the parts rather than after them.

It does not remove the 27 (module, key) pairs in part 4's ledger. That is
CFG-03's open ticket - option defaults living in four places - and it is a body
of behaviour change across `main.py`, `p1.py`, `inverter.py` and `validate.py`
that no single card should make in passing. What this file does instead is
FREEZE the ledger: the set is graded EXACTLY, so the twenty-eighth pair is red
on the commit that introduces it and the ledger can only ever shrink. A backlog
that cannot grow is a backlog being paid down; one that is merely written in a
comment is the state UX-01 left.

MUTATION.md is the method for the red-direction evidence in part 5, and
`tools/mutation_run.sh` is the harness.
"""

import ast
import io
import re
import sys
from pathlib import Path

HERE = Path(__file__).resolve().parent
APP = HERE / "app"

FAILED = 0


def check(what, ok):
    global FAILED
    print(("  ok   " if ok else "  FAIL ") + what)
    if not ok:
        FAILED += 1


# --------------------------------------------------------------------------- #
# config.yaml, parsed by hand.
#
# ⚠️ NO PyYAML, AND NOT AS A PREFERENCE. The add-on image installs py3-aiohttp
# and py3-paho-mqtt only; `import yaml` raises in the checks container and a
# check that skips on ImportError is not a check. `test_maintenance.py` and
# `test_price.py` each carry their own two-line reader for the same reason, and
# this is the third - which is itself an instance of the defect this file
# grades, one level up. It is written down rather than tidied away because
# tidying it into a shared helper is a change to two suites that are green.
CFG_TEXT = io.open(HERE / "config.yaml", encoding="utf-8").read()
_SCHEMA_AT = CFG_TEXT.index("\nschema:")
_OPTS_BLOCK, _SCHEMA_BLOCK = CFG_TEXT[:_SCHEMA_AT], CFG_TEXT[_SCHEMA_AT:]


def _scalar(raw):
    """One `options:` value, as the yaml module this image lacks would read it."""
    raw = raw.split("#", 1)[0].strip().strip('"').strip("'")
    if raw in ("true", "false"):
        return raw == "true"
    if raw == "[]":
        return []
    for cast in (int, float):
        try:
            return cast(raw)
        except ValueError:
            pass
    return raw


def _block_keys(block):
    return [m.group(1) for m in re.finditer(r"^  ([a-z0-9_]+):", block, re.M)]


SHIPPED = {m.group(1): _scalar(m.group(2))
           for m in re.finditer(r"^  ([a-z0-9_]+): *(.*)$", _OPTS_BLOCK, re.M)}
SCHEMA = set(_block_keys(_SCHEMA_BLOCK))

print("UX-04 - the defaults pin\n")
print("1. the surface this file quantifies over")
print(f"    config.yaml: {len(SHIPPED)} keys under `options:`, {len(SCHEMA)} "
      f"under `schema:`")
# The floor, for part 1's own reason: a reader that returned {} would make every
# check below vacuously true. 60 is the count of REQUIRED schema keys and is the
# number `tools/supervisor_shape_probe.py` section F now grades; any floor well
# inside the real size does the job, and this one cannot be reached by accident.
check(f"the hand reader found a real options block ({len(SHIPPED)} keys, floor 60)",
      len(SHIPPED) >= 60)
check(f"...and a real schema block ({len(SCHEMA)} keys, floor 60)",
      len(SCHEMA) >= 60)
check("every shipped default names a schema key, so a mirror pinned to one is "
      "pinned to the other",
      sorted(set(SHIPPED) - SCHEMA) == [])


# --------------------------------------------------------------------------- #
print()
print("2. the mirrors, DISCOVERED from the tree rather than listed here")
# ⚠️ WHY DISCOVERED. Six suites already pin their own mirror to config.yaml -
# test_maintenance, test_price, test_watchdog (twice), test_peakguard,
# test_delivery. Every one of those pins names its table by hand, so a SEVENTH
# table added tomorrow is pinned by nobody and nothing says so. This walks
# `app/` for module-level assignments whose name ends in OPTION_DEFAULTS and
# grades whatever it finds, so the seventh is covered on the commit that
# introduces it. That is the property the six hand-written pins cannot have.
MODULES = sorted(p.name for p in APP.glob("*.py"))
TREES = {name: ast.parse(io.open(APP / name, encoding="utf-8").read(), name)
         for name in MODULES}

MIRROR = {}          # schema key -> (module, table name, value)
TABLES = []
for name in MODULES:
    for node in TREES[name].body:
        if not (isinstance(node, ast.Assign) and len(node.targets) == 1
                and isinstance(node.targets[0], ast.Name)
                and node.targets[0].id.endswith("OPTION_DEFAULTS")):
            continue
        table = node.targets[0].id
        TABLES.append((name, table))
        for k, v in zip(node.value.keys, node.value.values):
            # A literal key with a literal value is the only shape a mirror
            # has; anything else is not a table this file can grade, and
            # saying so out loud beats skipping it silently.
            MIRROR[k.value] = (name, table, ast.literal_eval(v))

print(f"    {len(TABLES)} table(s): " + ", ".join(f"{m}:{t}" for m, t in TABLES))
print(f"    {len(MIRROR)} mirrored key(s) of {len(SCHEMA)} in `schema:`")
check(f"the walk found the mirrors that exist ({len(TABLES)} tables, floor 6)",
      len(TABLES) >= 6)
_unschemad = sorted(k for k in MIRROR if k not in SCHEMA)
check(f"every mirrored key is a real schema key {_unschemad or ''}",
      _unschemad == [])
_unshipped = sorted(k for k in MIRROR if k not in SHIPPED)
check(f"...and is shipped in the options block {_unshipped or ''}",
      _unshipped == [])
_drift = sorted(k for k, (m, t, v) in MIRROR.items()
                if k in SHIPPED and v != SHIPPED[k])
check(f"...and reproduces config.yaml exactly, so ANY mirror - including one no "
      f"suite names by hand - goes red on drift {_drift or ''}",
      _drift == [])


# --------------------------------------------------------------------------- #
print()
print("3. the AST scan: no module carries its own literal fallback for a "
      "MIRRORED schema default")


def literal_fallbacks(tree, name):
    """Every `f(..., "<schema key>", <constant>, ...)` in one module.

    Yields (module, lineno, callee, key, value). See the header for the two
    exclusions and the argument for each.
    """
    for node in ast.walk(tree):
        if not isinstance(node, ast.Call):
            continue
        fn = node.func
        callee = fn.attr if isinstance(fn, ast.Attribute) else getattr(fn, "id", "")
        if callee == "getattr":
            continue
        for i, arg in enumerate(node.args[:-1]):
            if not (isinstance(arg, ast.Constant) and isinstance(arg.value, str)
                    and arg.value in SCHEMA):
                continue
            default = node.args[i + 1]
            if not isinstance(default, ast.Constant):
                continue
            if default.value is None or default.value == "":
                continue
            yield (name, node.lineno, callee, arg.value, default.value)


ALL_SITES = [s for name in MODULES for s in literal_fallbacks(TREES[name], name)]
MIRRORED_SITES = [s for s in ALL_SITES if s[3] in MIRROR]
UNMIRRORED_SITES = [s for s in ALL_SITES if s[3] not in MIRROR]

print(f"    scanned {len(MODULES)} modules, {len(ALL_SITES)} literal fallback "
      f"site(s): {len(MIRRORED_SITES)} over mirrored keys, "
      f"{len(UNMIRRORED_SITES)} over unmirrored ones")
# ⚠️ THE SCANNER'S OWN FLOOR, and it is the line that stops this whole file
# rotting into clause one. A scanner that silently stopped matching - a renamed
# helper, an ast API change, a `SCHEMA` that came back empty - would report ZERO
# sites and every check below would pass. So the total is graded against a floor
# the tree already exceeds by a wide margin, and part 5 additionally feeds the
# scanner a source it MUST flag.
check(f"the scanner is not blind: it found {len(ALL_SITES)} sites (floor 20)",
      len(ALL_SITES) >= 20)
for site in sorted(MIRRORED_SITES):
    print(f"      {site[0]}:{site[1]}  {site[2]}({site[3]!r}, {site[4]!r})")
check(f"no literal fallback survives for any of the {len(MIRROR)} keys a mirror "
      f"already answers for - those sites must index the mirror",
      MIRRORED_SITES == [])


# --------------------------------------------------------------------------- #
print()
print("4. the CFG-03 ledger: the unmirrored keys that still carry a literal, "
      "frozen at exactly today's set")
# ⚠️ WHY A FROZEN SET AND NOT A COUNT. A count goes green again the moment
# somebody deletes one site and adds another, which is the same-size drift this
# repository keeps finding. The pair (module, key) is stable under every edit
# that is not the one being guarded against: reformatting, moving the call
# within the file, renaming the local `opts`. It changes only when a module
# starts or stops carrying a literal for a key - which is exactly the event.
#
# LINE NUMBERS ARE DELIBERATELY NOT IN THE KEY. A ledger that goes red when an
# unrelated edit shifts a line is a ledger people learn to re-bless without
# reading, and a ledger re-blessed without reading grades nothing.
LEDGER = {
    ("inverter.py", "max_w"),
    ("main.py", "auto_start"),
    ("main.py", "cycle_floor_s"),
    ("main.py", "deadband_w"),
    ("main.py", "estop_fitted"),
    ("main.py", "gain"),
    ("main.py", "heartbeat_s"),
    ("main.py", "log_level"),
    ("main.py", "meter_freeze_s"),
    ("main.py", "meter_max_age_s"),
    ("main.py", "meter_phases"),
    ("main.py", "peak_cap_w"),
    ("main.py", "saturation_cycles"),
    ("main.py", "saturation_w"),
    ("main.py", "sign_disagree_s"),
    ("main.py", "sign_disagree_w"),
    ("main.py", "slew_w"),
    ("main.py", "stale_input_s"),
    ("main.py", "step_w"),
    ("main.py", "target_grid_w"),
    ("p1.py", "meter_poll_s"),
    ("p1.py", "meter_source"),
    ("validate.py", "max_w"),
    ("validate.py", "meter_max_age_s"),
    ("validate.py", "meter_phases"),
    ("validate.py", "sign_disagree_w"),
    ("validate.py", "slew_w"),
}
_found = {(s[0], s[3]) for s in UNMIRRORED_SITES}
_new = sorted(_found - LEDGER)
_gone = sorted(LEDGER - _found)
print(f"    {len(_found)} (module, key) pair(s) carry a literal for a key no "
      f"mirror answers for")
check(f"no NEW module/key pair has started carrying a literal fallback {_new or ''}",
      _new == [])
check(f"...and none has been paid off without the ledger being shortened - "
      f"shrink LEDGER when it has {_gone or ''}",
      _gone == [])

# --- and every one of them is graded AGAINST config.yaml, which is the half
#     that catches CFG-04's actual failure: two copies of one number, one moves.
DIVERGENT = {
    # main.py:290 `target_grid_w=float(opts.get("target_grid_w", 0))`.
    # config.yaml ships -10 W. MEASURED here for the first time, 2026-09-07,
    # and it is a real divergence, not a sentinel: a dict handed to
    # `Controller` WITHOUT the key holds net exchange at 0 W instead of the
    # shipped -10 W bias into import. On a Supervisor install the key is always
    # present - the deep-merge of `options:` guarantees it - so this cannot
    # reach an operator; the paths that CAN hand over a partial dict are
    # `sim/inject.py` and this repository's own fixtures. It is recorded rather
    # than corrected because changing it is a behaviour change to the control
    # loop's target, which belongs to a card that grades the loop.
    ("main.py", "target_grid_w"): (0, -10),
}
_diverge = {}
for mod, line, callee, key, value in UNMIRRORED_SITES:
    want = SHIPPED.get(key)
    same = (isinstance(value, bool) == isinstance(want, bool)) and value == want
    if not same:
        _diverge[(mod, key)] = (value, want)
for pair, (got, want) in sorted(_diverge.items()):
    print(f"      {pair[0]}  {pair[1]}: code {got!r} vs config.yaml {want!r}")
check(f"every remaining literal still AGREES with config.yaml, except the "
      f"{len(DIVERGENT)} divergence(s) recorded above it",
      _diverge == DIVERGENT)


# --------------------------------------------------------------------------- #
print()
print("5. the red direction, RUN on every invocation")
# ⚠️ A CHECK THAT HAS NEVER BEEN SHOWN FAILING IS NOT A CHECK. Parts 3 and 4
# both pass on a clean tree, which is exactly what UX-01 AC 4's clause one did
# while grading nothing. So the scanner is fed a source that is a deliberate
# violation of each rule and is REQUIRED to flag it - in-process, against the
# same `literal_fallbacks()` the parts above call, so a scanner that stops
# working stops working here too.
_PLANTED = '''
def build(opts):
    a = opts.get("maintenance_charge_w", 2000)     # mirrored key -> part 3
    b = _number(opts, "cycle_floor_s", 15)         # unmirrored, main.py's key
    c = opts.get("static_profile", "")             # the "" sentinel: NOT a hit
    d = getattr(unit, "max_w", 0.0)                # attribute guard: NOT a hit
    e = opts.get("peak_soc_floor", 20)             # mirrored key -> part 3
    return a, b, c, d, e
'''
_hits = list(literal_fallbacks(ast.parse(_PLANTED, "<planted>"), "<planted>"))
_keys = sorted(h[3] for h in _hits)
for h in _hits:
    print(f"      caught  {h[2]}({h[3]!r}, {h[4]!r})  at line {h[1]}")
check("a planted literal for a MIRRORED key is caught - this is part 3's red "
      "direction, exercised",
      "maintenance_charge_w" in _keys and "peak_soc_floor" in _keys)
check("a planted literal for an unmirrored key is caught too - part 4's",
      "cycle_floor_s" in _keys)
check('...and the two documented exclusions still hold: the "" sentinel and '
      "getattr are not reported",
      "static_profile" not in _keys and _keys.count("max_w") == 0)
check(f"exactly the three intended sites, no more {_keys}",
      _keys == ["cycle_floor_s", "maintenance_charge_w", "peak_soc_floor"])

# ...and the mirror comparison's red direction, likewise executed rather than
# claimed: part 2 goes red when a table and config.yaml disagree by one.
_drifted = dict(MIRROR)
_pick = sorted(MIRROR)[0]
_mod, _tab, _val = _drifted[_pick]
_drifted[_pick] = (_mod, _tab, "###drifted###")
check(f"part 2's comparison rejects a mirror that has drifted (simulated on "
      f"{_pick!r})",
      any(v != SHIPPED.get(k) for k, (_, _, v) in _drifted.items()))


print()
if FAILED:
    print(f"{FAILED} FAILED")
    sys.exit(1)
print("all checks passed")