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