Files
goodwe-addon/goodwe_controller/test_settings.py
T
adminandClaude Opus 5.5 be57b87c34
checks / unit (pull_request) Successful in 10m22s
infra-pin / fresh (pull_request) Successful in 1s
UX-13 P4: the setup wizard for installers
The Setup tab draws the Installer view's page model one step at a time,
from each field's `wizard` step in app/uilayout.py: meter (with the kettle
test), inverters and batteries, solar (optional), prices, peak limit
(battery modes: "comes in a later update", OPT-11), check and start.
It reuses P2's controls and saves through /config/validate and
/config/save. No route was added.

- First run: /config/model says `setup_needed` when the stored options name
  no power request entity, or on a new install (no `setup_done` and no
  `auto` in the Store, control not started). The page opens #setup only
  when opened with no route. The wizard's save writes `setup_done`.
- Kettle test: GET ha/entities?entity= reads one entity now, in W. The
  change decides: up keeps meter_invert off, down turns it on; a net or
  import/export meter that goes down blocks the step with what to check.
- Last step: validate.setup_checks() beside the refusals: the meter reads,
  each power request entity exists in HA (bad if not, SAFETY-77), battery
  levels and power with direction, solar sign, prices. Saves with
  shadow_mode on by default, then a confirmed restart naming the 0 W.
- P2 test finding (a): a stale save lists only the keys the other surface
  moved (the blob handed out on open, by fingerprint, against stored now).
- test_setup.py; tools/ux13_wizard_blobs.py makes the 88 blobs the wizard
  can save; all accepted by Supervisor 2026.08.0 AppOptions (offline).
- DECISIONS/UX-13-setup-wizard.md; DOCS "Setup, for a new install".

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LyR88rXETvwhTgx6yHE39p
2026-09-28 05:31:18 +02:00

467 lines
20 KiB
Python

"""UX-13 P2: the Installer settings view. `python3 test_settings.py`
The view is drawn from app/uilayout.py through app/uipage.py and saves the
page model through app/uimodel.py and the shipped validator. Part 1 checks
that every group and every mapped field reaches the page with its words,
range and choices. Part 2 checks the page script names no option key. Parts
3-5 drive the real routes over HTTP against a fake Supervisor: the save round
trip, where a refusal lands, and a stale save. Part 6 holds the view's words
to STYLE. The headless-browser click-through is in the PR, not in CI (the CI
image has no browser).
Mutating a module this file grades? Read MUTATION.md first (purge __pycache__).
"""
import asyncio
import copy
import glob
import io
import json
import logging
import os
import re
import sys
HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, os.path.join(HERE, "tools"))
logging.disable(logging.WARNING) # the validator's own log lines
from aiohttp import ClientSession, web as aioweb # noqa: E402
import translations_lint as TL # noqa: E402
import ux13_demo as D # noqa: E402
from app import selfconfig, uilayout as L, uimodel as U, uipage, web # noqa: E402
from app.validate import validate_startup_config # noqa: E402
from named_absence import field # noqa: E402
fails = []
def check(name, cond):
print(f" {'ok ' if cond else 'FAIL'} {name}")
if not cond:
fails.append(name)
SCHEMA = D.schema_ui(D.CONFIG)
TRANSLATIONS = {"en": TL.parse_translation_file(
os.path.join(HERE, "translations", "en.yaml"))}
DEFAULTS = D.defaults(D.CONFIG)
SITE = dict(DEFAULTS, setpoint_entity="number.goodwe_setpoint",
soc_entity="sensor.soc", batt_entity="sensor.batt")
TWO = dict(SITE, setpoint_entity="", inverters=[
{"name": "inverter-a", "setpoint_entity": "number.a", "max_w": 2500,
"bms_soc_floor": 10},
{"name": "inverter-b", "setpoint_entity": "number.b", "max_w": 2500,
"bms_soc_floor": 10}])
UI_TEXT = uipage.ui_text()
W = UI_TEXT["settings"]
def browser(value):
"""What JSON.parse then JSON.stringify does to a value: 3.0 becomes 3."""
if isinstance(value, float) and value.is_integer():
return int(value)
if isinstance(value, list):
return [browser(v) for v in value]
if isinstance(value, dict):
return {k: browser(v) for k, v in value.items()}
return value
# --------------------------------------------------------------------------- #
print("1. every group and every mapped field reaches the page")
# --------------------------------------------------------------------------- #
LAYOUT = uipage.layout(SCHEMA, TRANSLATIONS)
_flat = []
for _g in LAYOUT["groups"]:
for _p in _g["parts"]:
for _f in _p["fields"]:
_flat.append((_g["id"], _f))
_flat += [(_g["id"], _s) for _s in _f.get("fields", [])]
_keys = [f["key"] for _g, f in _flat]
FIELD = {f["key"]: f for _g, f in _flat}
check("the six groups, in the design's order",
[g["id"] for g in LAYOUT["groups"]]
== ["meter", "inverters", "tariff", "safety", "maintenance", "tuning"]
and [g["name"] for g in LAYOUT["groups"]]
== ["Meter", "Inverters and batteries", "Tariff and prices",
"Safety limits", "Maintenance", "Control tuning"])
check(f"every field of the table is on the page once ({len(_keys)} fields)",
sorted(_keys) == sorted(f.key for _g, _p, f in L.fields())
and len(_keys) == len(set(_keys)))
_counts = {g["id"]: sum(1 for gid, _f in _flat if gid == g["id"])
for g in LAYOUT["groups"]}
print(f" fields per group: {_counts}")
check("every field has a label, and every option field's label is its "
"translation without the group prefix",
all(f["label"] and " · " not in f["label"] for _g, f in _flat)
and field(FIELD, "meter_source")["label"]
== "Where the meter reading comes from")
_ranged = [(f["key"], f.get("min"), f.get("max")) for _g, f in _flat
if f["control"] in (L.SLIDER, L.LOCKED)]
check(f"every slider and locked value has its range from the schema "
f"({len(_ranged)})", all(lo is not None and hi is not None
for _k, lo, hi in _ranged)
and ("peak_cap_w", 500.0, 15000.0) in _ranged)
check("the peak limit is a kW slider over a value stored in W, and its label "
"does not say W (the unit is shown beside the control)",
field(FIELD, "peak_cap_w")["scale"] == 1000
and field(FIELD, "peak_cap_w")["label"] == "Peak limit"
and not [f["key"] for _g, f in _flat
if f["unit"] and re.search(r"\([^()\s]{1,6}\)", f["label"])])
_choice = [f for _g, f in _flat
if f["control"] in (L.SEGMENTED, L.SELECT, L.CHOICE)]
check("every choice control has choices with words",
all(f["choices"] and all(c[1] for c in f["choices"]) for f in _choice))
_ms = field(FIELD, "meter_source")
check("the meter source is radio cards with a line of help on each",
_ms["control"] == L.CHOICE and set(_ms["hints"]) == {c[0] for c in
_ms["choices"]})
_reg = field(FIELD, "ui.region")
check("the region shows its preset year beside the choice (P0 review)",
["flanders", "Flanders, Fluvius Antwerpen, 2026 figures", False]
in _reg["choices"] and _reg["presets"]["flanders"]["year"] == 2026)
check("...and says plainly that Wallonia and Brussels have no preset yet "
"(TARIFF-22)", _reg["later"] == [["wallonia", "Wallonia, no preset yet"],
["brussels", "Brussels, no preset yet"]]
and "Wallonia or Brussels" in _reg["help"])
_locked = sorted(f["key"] for _g, f in _flat if f["control"] == L.LOCKED)
check(f"measured and derived values are locked behind 'Change' "
f"({len(_locked)}): every control tuning value, every safety stop and "
"the peak reserve",
{f.key for p in L.GROUPS[-1].parts for f in p.fields} <= set(_locked)
and "sign_disagree_w" in _locked and "peak_soc_floor" in _locked
and LAYOUT["groups"][-1]["support_only"])
check("each group says when its settings apply",
all(g["applies"] == [UI_TEXT["applies"]["restart"]["name"]]
for g in LAYOUT["groups"]))
_inv = field(FIELD, "inverters")
check("the inverter cards carry their count, and adding one makes a list",
_inv["count"] == U.COUNT and _inv["grow"] == U.GROWN
and [s["key"] for s in _inv["fields"]]
== [f.key for f in L.ROSTER_FIELDS])
check("a page-only field names the option keys it writes, for the grey key "
"line", _reg["writes"] == ["price_taxes_profile"]
and field(FIELD, "ui.solar")["writes"]
== ["pv_entities", "pv_entity", "pv_invert_entities"]
and "static_profile" in field(FIELD, "ui.contract")["writes"])
check("without a Supervisor schema the page still draws, with no ranges",
len(uipage.layout([], {})["groups"]) == 6)
# --------------------------------------------------------------------------- #
print("\n2. the page script names no option key")
# --------------------------------------------------------------------------- #
JS = io.open(os.path.join(HERE, "app", "static", "settings.js"),
encoding="utf-8").read()
ORDER, _nested = TL.config_surface(D.CONFIG)
_leak = sorted(k for k in ORDER if re.search(
r"[^A-Za-z0-9_]" + re.escape(k) + r"[^A-Za-z0-9_]", JS))
check(f"settings.js names no option key ({_leak})", not _leak)
check("data is set as text, never as HTML",
"innerHTML" not in JS and "insertAdjacentHTML" not in JS)
_fetch = sorted(set(re.findall(r'fetch\("([^"?]+)', JS)))
check(f"every request is relative, to the add-on's own routes ({_fetch})",
_fetch == ["config/model", "config/restart", "config/save",
"config/validate", "ha/entities"])
def _resolve(path, node=W):
for k in path.split("."):
if not isinstance(node, dict) or k not in node:
return None
node = node[k]
return node
_used = {re.sub(r"\.(replace|join|length)$", "", k) for k in
re.findall(r'\bW\.([a-z_]+(?:\.[a-z_]+)*)', JS)}
_used |= set(re.findall(r'data-s(?:-ph|-label)?="([\w.]+)"', io.open(
os.path.join(HERE, "app", "static", "index.html"), encoding="utf-8").read()))
_miss = sorted(k for k in _used if _resolve(k) is None)
check(f"every word the view asks for is in app/ui_text ({len(_used)} keys, "
f"missing: {_miss})", len(_used) > 25 and not _miss)
check("search matches a field's label, its key and its help",
"(f.label + \" \" + keys + \" \" + f.key + \" \" + (f.help || \"\"))"
in JS)
check("a locked value shows no input until 'Change' is pressed",
re.search(r"function locked\(.*?\n }\n", JS, re.S)
and "UNLOCKED.has(id)) { open(); return box; }" in JS)
check("nothing is saved while the add-on's own checks refuse the model",
"if (MSG.some((m) => m.tier === \"bad\")) { toast(" in JS)
check("Discard reloads the stored settings (after a stale save the page's "
"copy is no longer what is stored)",
re.search(r'"#sb-discard"\)\.addEventListener\("click", async \(\) => '
r'\{\s*await load\(\);', JS) is not None)
check("a load locks every locked value again, and a choice the page can only "
"keep stays offered while it is the stored value",
"UNLOCKED.clear();" in JS.split("async function load")[1].split("render();")[0]
and "same(c[0], was)" in JS)
check("removing cards back to one restores the stored shape (a flat inverter "
"is not turned into a list of one)",
"if (V[f.key].length === 1) for (const k in f.grow) V[k] = clone(ORIG[k]);"
in JS)
check("the old settings page stays reachable (until P5)",
'href="config" id="set-classic"' in web.page())
# --------------------------------------------------------------------------- #
# The real routes, over HTTP, against a fake Supervisor
# --------------------------------------------------------------------------- #
class Supervisor:
def __init__(self, options):
self.options = copy.deepcopy(options)
self.writes = []
async def info(self):
return {"options": copy.deepcopy(self.options), "schema": SCHEMA,
"translations": TRANSLATIONS}
async def write_options(self, options):
self.writes.append(copy.deepcopy(options))
self.options = copy.deepcopy(options)
async def restart(self, confirm):
raise AssertionError("no restart here")
class Hass:
session = None
token = ""
async def limits(self, _e):
return (-5000.0, 5000.0, 1.0)
async def states(self):
return []
class Controller:
def __init__(self, opts):
self.o, self.hass, self.inverters = dict(opts), Hass(), []
def log_event(self, text, hint=None):
pass
def status_payload(self):
return {"banner": "", "level": "ok", "rows": [], "checks": []}
async def handle_action(self, what):
raise AssertionError(f"the settings view reached {what!r}")
async def session(stored, steps):
"""Run `steps(s, root)` against the app over a fake Supervisor."""
sup = Supervisor(stored)
runner = aioweb.AppRunner(web.build_app(Controller(stored), sup))
await runner.setup()
site = aioweb.TCPSite(runner, "127.0.0.1", 0)
await site.start()
root = f"http://127.0.0.1:{site._server.sockets[0].getsockname()[1]}/"
try:
async with ClientSession() as s:
out = await steps(s, root, sup)
finally:
await runner.cleanup()
return out, sup
async def _get(s, url):
async with s.get(url) as r:
return await r.json()
async def _post(s, url, body):
async with s.post(url, json=body) as r:
return r.status, await r.json()
# --------------------------------------------------------------------------- #
print("\n3. the save round trip")
# --------------------------------------------------------------------------- #
check("the fixture site starts (so a refusal below is the edit's)",
not validate_startup_config(TWO) and not validate_startup_config(SITE))
async def _round(s, root, sup):
out = {}
m = await _get(s, root + "config/model?lang=en")
out["model"] = m
model = browser(m["ui"])
out["same"] = await _post(s, root + "config/save",
{"model": model, "fingerprint": m["fingerprint"]})
out["after_same"] = copy.deepcopy(sup.options)
m = await _get(s, root + "config/model?lang=en")
model = browser(m["ui"])
model["peak_cap_w"] = 4200
model["inverters"][1]["max_w"] = 2200
out["edit"] = await _post(s, root + "config/save",
{"model": model, "fingerprint": m["fingerprint"]})
return out
_stored = dict(TWO, gain=0.5, target_grid_w=-10.0, peak_cap_w=3500)
_r, _sup = asyncio.run(session(_stored, _round))
check("config/model carries the layout and the page model",
_r["model"]["layout"] == json.loads(json.dumps(uipage.layout(
SCHEMA, TRANSLATIONS))) and _r["model"]["ui"]["inverter_count"] == 2)
check("a save that changed nothing writes an identical blob: same values, "
"types and key order, although the browser turned -10.0 into -10",
_r["same"][0] == 200 and json.dumps(_r["after_same"])
== json.dumps(_stored) and isinstance(_r["after_same"]["target_grid_w"],
float))
_written = _sup.writes[-1]
print(f" the edit: {_r['edit']}, changed "
f"{sorted(k for k in _stored if _written[k] != _stored[k])}")
check("an edit writes exactly the fields edited, the peak limit in W",
_r["edit"][0] == 200
and sorted(k for k in _stored if _written[k] != _stored[k])
== ["inverters", "peak_cap_w"] and _written["peak_cap_w"] == 4200
and _written["inverters"][1]["max_w"] == 2200
and _written["inverters"][0] == _stored["inverters"][0])
_blobs = [("config.yaml", DEFAULTS), ("site", SITE), ("two", TWO)] + [
(os.path.basename(f), json.load(open(f))) for f in sorted(glob.glob(
os.path.join(HERE, "scenarios", "ux13", "test23", "*.json")))]
_bad = [n for n, o in _blobs if json.dumps(uipage.to_options(browser(
json.loads(json.dumps(U.options_to_ui(o)))), o)) != json.dumps(o)]
check(f"every known blob comes back byte for byte through the browser "
f"({len(_blobs)} blobs, not: {_bad})", not _bad and len(_blobs) >= 8)
_grown = U.options_to_ui(SITE)
_grown["inverters"].append(dict(_grown["inverters"][0], name="b",
setpoint_entity="number.b"))
_grown["inverters"][0]["name"] = "a"
_grown.update(U.GROWN)
_g = U.ui_to_options(_grown)
check("'Add an inverter' on a flat site writes a list and clears the flat "
"power request entity, which the validator refuses beside a list",
[e["name"] for e in _g["inverters"]] == ["a", "b"]
and _g["setpoint_entity"] == "" and not validate_startup_config(_g))
# --------------------------------------------------------------------------- #
print("\n4. a refusal lands beside the field it is about")
# --------------------------------------------------------------------------- #
async def _refuse(s, root, sup):
out = {}
m = await _get(s, root + "config/model?lang=en")
base = m["ui"]
async def validate(**edits):
model = copy.deepcopy(base)
for k, v in edits.items():
if k == "card":
model["inverters"][v[0]][v[1]] = v[2]
else:
model[k.replace("__", ".")] = v
return (await _post(s, root + "config/validate", {"model": model}))[1]
out["card"] = await validate(card=(1, "setpoint_entity", ""))
out["same"] = await validate(card=(1, "setpoint_entity", "number.a"))
out["floor"] = await validate(peak_soc_floor=5)
out["derive"] = await validate(ui__contract="dynamic", price_provider="")
model = copy.deepcopy(base)
model.update({"ui.contract": "dynamic", "price_provider": ""})
out["derive_save"] = await _post(s, root + "config/save", {
"model": model, "fingerprint": m["fingerprint"]})
out["derive_writes"] = len(sup.writes)
model = copy.deepcopy(base)
model["inverters"][1]["setpoint_entity"] = ""
out["save"] = await _post(s, root + "config/save",
{"model": model, "fingerprint": m["fingerprint"]})
return out
_r, _sup = asyncio.run(session(TWO, _refuse))
def _at(payload, tier="bad"):
return {(m["field"], m["card"]) for m in payload["messages"]
if m["tier"] == tier}
print(f" blank power request entity on card 2: {_r['card']['messages']}")
check("a blank power request entity on the second card is refused beside "
"that card's field, and not beside the list as well",
("inverters.setpoint_entity", 1) in _at(_r["card"])
and ("inverters", None) not in _at(_r["card"]))
print(f" two cards on one entity: {_r['same']['messages']}")
check("two cards on one entity are refused beside that field on both cards",
{("inverters.setpoint_entity", 0), ("inverters.setpoint_entity", 1)}
== _at(_r["same"]))
check("a peak reserve under the cut-out level lands on the peak reserve and "
"on each card's cut-out level",
{("peak_soc_floor", None), ("inverters.bms_soc_floor", 0),
("inverters.bms_soc_floor", 1)} <= _at(_r["floor"]))
check("a model no blob can say is refused beside its field, in plain words",
_r["derive"]["messages"] == [{
"field": "price_provider", "card": None, "tier": "bad",
"text": W["derive"]["no_source"]["description"]}])
check("a save no blob can say answers 422 with its sentence and writes nothing",
_r["derive_save"][0] == 422 and _r["derive_writes"] == 0
and _r["derive_save"][1]["messages"] == _r["derive"]["messages"])
check("a refused save answers 422 with the same placement and writes nothing",
_r["save"][0] == 422 and ("inverters.setpoint_entity", 1)
in _at(_r["save"][1]) and _sup.writes == [])
_fixed = U.options_to_ui(dict(TWO, static_profile="not json"))
check("a sentence about a key the page derives goes to the field that writes "
"it, or to the visible choice that sets it",
uipage.targets("static_profile", "", _fixed) == [("ui.contract", None)]
and uipage.targets("max_w", "", U.options_to_ui(SITE))
== [("inverters.max_w", 0)]
and uipage.targets("pv_entity", "", _fixed) == [("ui.solar", None)])
check("a sentence naming no setting on the page is kept, with no field",
uipage.anchors(["Something odd."], [], _fixed, ["gain"])
== [{"field": None, "card": None, "tier": "bad",
"text": "Something odd."}])
# --------------------------------------------------------------------------- #
print("\n5. a stale save names the fields that changed elsewhere")
# --------------------------------------------------------------------------- #
async def _stale(s, root, sup):
m = await _get(s, root + "config/model?lang=en")
sup.options["peak_cap_w"] = 5000 # the Configuration tab saves
model = copy.deepcopy(m["ui"])
model["gain"] = 0.4
return await _post(s, root + "config/save",
{"model": model, "fingerprint": m["fingerprint"]})
_r, _sup = asyncio.run(session(TWO, _stale))
check("409 with the page fields to look at, and nothing written. Only the "
"field the other surface changed, not the page's own edit (P4, P2 "
"test finding a)",
_r[0] == 409 and _r[1]["kind"] == "stale"
and _r[1]["changed_fields"] == ["peak_cap_w"] and _sup.writes == []
and _r[1]["fingerprint"] == selfconfig.fingerprint(_sup.options))
# --------------------------------------------------------------------------- #
print("\n6. the view's words keep to STYLE")
# --------------------------------------------------------------------------- #
def _texts(node, path=""):
for k, v in node.items():
if isinstance(v, dict):
yield from _texts(v, f"{path}{k}.")
elif isinstance(v, str):
yield f"{path}{k}", v
_long = [(p, t) for p, t in _texts(W)
if len(t.split()) > (25 if p.endswith("description") else 6)
or (p.endswith("description") and len(re.findall(r"[.!?](?:\s|$)", t)) > 1)
or ";" in t]
check(f"every settings string is one sentence of at most 25 words, or a "
f"name of at most 6 ({_long})", not _long)
_labels = [c[1] for _g, f in _flat for c in f.get("choices", [])]
check("every choice, including the region with its year, is at most 6 words",
all(len(x.split()) <= 6 for x in _labels))
print()
if fails:
print(f"{len(fails)} FAILED")
sys.exit(1)
print("all passed")