"""UX-13 P4: the setup wizard. `python3 test_setup.py` The wizard is the Installer view's page model drawn one step at a time from each field's `wizard` step (app/uilayout.py), saved through the same routes. Part 1 checks the steps and the tables the page reads. Part 2 the kettle test's sign logic. Part 3 which refusals block which step. Part 4 the first-run rule on new and existing sites. Part 5 the last step's live checks. Part 6 the save through the P2 path, and every blob the wizard can produce (tools/ux13_wizard_blobs.py; the Supervisor's own validator runs over the same set in the PR, not in CI, which has no Supervisor image). Part 7 the stale save names only what changed elsewhere (P2 test, finding a). Part 8 the words. The headless-browser click-through is in the PR. Mutating a module this file grades? Read MUTATION.md first (purge __pycache__). """ import asyncio import copy 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 import ux13_wizard_blobs as B # noqa: E402 from app import uilayout as L, uimodel as U, uipage, web # noqa: E402 from app.validate import setup_checks, validate_startup_config # 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) NEW = dict(DEFAULTS, setpoint_entity="number.inv1_power") UI_TEXT = uipage.ui_text() WS = UI_TEXT["settings"]["setup"] JS = io.open(os.path.join(HERE, "app", "static", "settings.js"), encoding="utf-8").read() # --------------------------------------------------------------------------- # print("1. six steps, drawn from the layout table") # --------------------------------------------------------------------------- # LAYOUT = uipage.layout(SCHEMA, TRANSLATIONS) TOP = [f for g in LAYOUT["groups"] for p in g["parts"] for f in p["fields"]] BY_STEP = {n: [f["key"] for f in TOP if f["wizard"] == n] for n in range(1, L.WIZARD_STEPS + 1)} print(f" {BY_STEP}") check("every field reaches the page with its wizard step", all("wizard" in f for f in TOP) and all(s["wizard"] == 2 for f in TOP for s in f.get("fields", []))) check("the mockup's six steps: meter, inverters, solar, prices, peak limit, " "check and start", BY_STEP[1][0] == "meter_source" and BY_STEP[2] == ["inverters"] and BY_STEP[3] == ["ui.solar"] and BY_STEP[4][0] == "ui.contract" and BY_STEP[5] == ["peak_cap_w", "peak_forecast_entity"] and BY_STEP[6] == [L.SHADOW[0]] and [WS["steps"][str(n)]["name"] for n in range(1, 7)] == ["Your meter", "Inverters and batteries", "Solar", "Prices", "Peak limit", "Check and start"]) check("battery modes are not a step yet (OPT-11): step 5 says they come later", "later update" in WS["modes_later"]["description"] and not any("mode" in k for k in BY_STEP[5])) SPEC = uipage.setup_spec() check("the spec the page reads is the table's", SPEC["steps"] == 6 and SPEC["final"] == 6 and SPEC["optional"] == [3] and SPEC["kettle"]["min_w"] == L.KETTLE_MIN_W and set(SPEC["kettle"]["sources"]) == set(L.KETTLE) and SPEC["shadow"] == {"field": "shadow_mode", "on": "on", "off": "auto"}) _src = L.field(L.KETTLE_SOURCE) _choices = {c[0] for f in TOP if f["key"] == L.KETTLE_SOURCE for c in f["choices"]} _bad = [] for source, (read, outcomes) in L.KETTLE.items(): model = {L.KETTLE_SOURCE: source} for key, _sign in read: fld = L.field(key) if not (fld and fld.wizard == 1 and fld.control == L.ENTITY and L.visible(fld, model)): _bad.append(key) for _say, sets in outcomes.values(): for key in (sets or {}): fld = L.field(key) if not (fld and fld.wizard == 1 and L.visible(fld, model)): _bad.append(key) check(f"each kettle source is a meter choice, and reads and sets only step-1 " f"fields shown for it ({_bad})", set(L.KETTLE) <= _choices and not _bad and _src.wizard == 1) _shadow = next((f for f in TOP if f["key"] == L.SHADOW[0]), {}) check("the shadow switch writes two of shadow mode's own choices", {L.SHADOW[1], L.SHADOW[2]} <= {c[0] for c in _shadow.get("choices", [])}) # --------------------------------------------------------------------------- # print("\n2. the kettle test's sign logic") # --------------------------------------------------------------------------- # K = uipage.kettle check("one grid entity: up keeps the sign, down flips it", K("off", -10, 1990) == ("ok", {"meter_invert": False}, 2000) and K("off", 400, -1600) == ("flip", {"meter_invert": True}, -2000)) check("the direction is the change, not the sign: exporting 3 kW, a kettle " "still moves the reading up", K("off", -3000, -1000)[0] == "ok" and K("off", 3000, 1000)[0] == "flip") check(f"a change under {L.KETTLE_MIN_W:.0f} W sets nothing and is not passed", K("off", 0, 499)[:2] == ("none", None) and K("off", 0, -499)[:2] == ("none", None) and K("off", 0, 500)[0] == "ok") check("a net entity that goes down cannot be fixed by an option, so the step " "is not passed", K("ha_signed", 0, 2000) == ("ok", {}, 2000) and K("ha_signed", 0, -2000)[:2] == ("wrong_signed", None)) check("import and export are read as their difference", L.KETTLE["ha_dsmr"][0] == (("p1_import_entity", 1), ("p1_export_entity", -1)) and K("ha_dsmr", -3000, -1000)[0] == "ok" and K("ha_dsmr", 0, -2000)[0] == "wrong_dsmr") check("a meter the add-on reads itself has no kettle test on the page", K("homewizard_local", 0, 2000) is None and K("mqtt_p1", 0, 2000) is None) _js = re.search(r"function kettleOutcome\(src, before, after\) \{(.*?)\n \}", JS, re.S) check("the page reaches the same verdict: settings.js kettleOutcome() is " "uipage.kettle() line for line", _js and "const d = after - before;" in _js.group(1) and "Math.abs(d) < M.setup.kettle.min_w" in _js.group(1) and 'src[d > 0 ? "up" : "down"]' in _js.group(1)) check("...and it applies what the verdict sets, and passes the step only on " "a verdict that sets something", "if (o.sets) { Object.assign(V, o.sets); queueValidate(); }" in JS and "ok: o.sets != null" in JS) check("a changed meter or entity forgets the result, so the test runs again", "if (WZ.kettle && WZ.kettle.sig !== sig) WZ.kettle = null;" in JS) check("each verdict has words", all(say in WS["kettle"] for _r, o in L.KETTLE.values() for say, _s in o.values()) and {"none", "unread", "skipped", "self"} <= set(WS["kettle"])) # --------------------------------------------------------------------------- # print("\n3. which refusals block which step") # --------------------------------------------------------------------------- # _m = U.options_to_ui(dict(NEW, meter_source="ha_signed")) _msgs = [{"field": "p1_net_entity", "card": None, "tier": "bad", "text": "a"}, {"field": "meter_entity", "card": None, "tier": "bad", "text": "b"}, {"field": "inverters.setpoint_entity", "card": 0, "tier": "bad", "text": "c"}, {"field": "maintenance_charge_w", "card": None, "tier": "bad", "text": "d"}, {"field": None, "card": None, "tier": "bad", "text": "e"}, {"field": "peak_cap_w", "card": None, "tier": "warn", "text": "f"}] check("a refusal blocks the step of its field, a hidden or non-wizard field's " "and an unplaced one the last step, and a warning blocks nothing", uipage.step_problems(_msgs, _m) == {"1": 1, "2": 1, "3": 0, "4": 0, "5": 0, "6": 3}) class Store: """app/store.py's shape: a dict under .data, and set().""" def __init__(self, data=None): self.data = dict(data or {}) def set(self, key, value): self.data[key] = value def state(eid, value, unit="W", cls=None): return {"entity_id": eid, "state": value, "attributes": {"friendly_name": eid, "unit_of_measurement": unit, "device_class": cls}} class Hass: session = None token = "" def __init__(self, states): self.states_now = states self.reads = [] async def limits(self, _e): return (-5000.0, 5000.0, 1.0) async def states(self): return copy.deepcopy(self.states_now) async def state(self, entity): self.reads.append(entity) return next((copy.deepcopy(s) for s in self.states_now if s["entity_id"] == entity), None) class Supervisor: def __init__(self, options): self.options = copy.deepcopy(options) self.writes = [] self.restarts = 0 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): self.restarts += 1 class Controller: def __init__(self, opts, states, store=None, auto=False, grid=None): self.o, self.hass, self.inverters = dict(opts), Hass(states), [] self.store, self.auto, self.grid = store, auto, grid def log_event(self, text, hint=None): pass def status_payload(self): return {"banner": "", "level": "ok", "rows": [], "checks": [], "plan": {"prices": [{"eur_kwh": 0.2}] * 20 + [{"eur_kwh": None}] * 4}} async def handle_action(self, what): raise AssertionError(f"the wizard reached {what!r}") async def session(ctl, stored, steps): sup = Supervisor(stored) runner = aioweb.AppRunner(web.build_app(ctl, 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() STATES = [state("sensor.grid", "402"), state("number.inv1_power", "0"), state("sensor.inv1_level", "62", "%", "battery"), state("sensor.inv1_battery", "-800"), state("sensor.pv", "-1200", "W", "power"), state("sensor.net", "-3.1", "kW", "power")] async def _steps(s, root, sup): m = await _get(s, root + "config/model?lang=en") model = copy.deepcopy(m["ui"]) model["inverters"][0]["setpoint_entity"] = "" return (await _post(s, root + "config/validate", {"model": model}))[1] _r, _sup = asyncio.run(session(Controller(NEW, STATES), NEW, _steps)) print(f" blank power request entity: steps {_r['steps']}") check("over HTTP, a blank power request entity blocks step 2 and no other", _r["steps"]["2"] >= 1 and all(v == 0 for k, v in _r["steps"].items() if k != "2")) check("the page moves on only when the step's count is 0, the kettle test is " "done on step 1, and the last step saves only with nothing red", "if ((STEPS[n] || 0) > 0) { toast(W.toast.blocked.description); " "firstBad(); return; }" in JS and "if (n === 1 && !kettleOk())" in JS and "if (WZ.blocked) { toast(WS.check.blocked.description); return; }" in JS) # --------------------------------------------------------------------------- # print("\n4. first run: a new install opens the wizard, a working site not") # --------------------------------------------------------------------------- # S = uipage.setup_needed check("a new install: never set up, control never started or stopped", S(NEW, {}, False) == "first_start" and S(NEW, {"phase": "idle", "sign_tripped": False}, False) == "first_start") check("a site that runs control, or ever ran or stopped it, is not sent to " "the wizard", S(NEW, {}, True) is None and S(NEW, {"auto": True}, False) is None and S(NEW, {"auto": False}, False) is None) check("a site the wizard saved is not sent again", S(NEW, {uipage.SETUP_DONE: "2026-09-28T10:00:00+00:00"}, False) is None) check("stored options with no inverter to control always open it, whatever " "the Store says", S(DEFAULTS, {uipage.SETUP_DONE: "x", "auto": True}, True) == "no_power_request" and S(dict(DEFAULTS, inverters=[{"name": "a", "setpoint_entity": " "}]), None, True) == "no_power_request") check("a list roster with a power request entity is set up", S(dict(DEFAULTS, inverters=[{"name": "a", "setpoint_entity": "number.a"}]), {"auto": True}, True) is None) check("without a Store the answer is no (never nag on a guess)", S(NEW, None, False) is None) _sites = [(n, dict(DEFAULTS, **json.load(open(os.path.join( HERE, "scenarios", "ux13", "test23", n))))) for n in sorted(os.listdir( os.path.join(HERE, "scenarios", "ux13", "test23")))] _starting = [(n, o) for n, o in _sites if not validate_startup_config(o)] check(f"the TEST-23 sites that start, and run control, are not sent to the " f"wizard ({[n for n, _o in _starting]})", len(_starting) >= 3 and all(S(o, {"auto": True, "phase": "idle"}, True) is None for _n, o in _starting)) check("...and the one that cannot start (no power request entity) is", [S(o, {"auto": True}, True) for n, o in _sites if (n, o) not in _starting] == ["no_power_request"]) async def _first(s, root, sup): out = {"model": await _get(s, root + "config/model?lang=en")} m = out["model"] model = copy.deepcopy(m["ui"]) model["peak_cap_w"] = 4000 out["plain"] = await _post(s, root + "config/save", { "model": model, "fingerprint": m["fingerprint"]}) out["store_plain"] = dict(_store.data) out["after_plain"] = await _get(s, root + "config/model?lang=en") m = out["after_plain"] model = copy.deepcopy(m["ui"]) out["wizard"] = await _post(s, root + "config/save", { "model": model, "fingerprint": m["fingerprint"], "setup": "done"}) out["after"] = await _get(s, root + "config/model?lang=en") return out _store = Store({"phase": "idle"}) _r, _sup = asyncio.run(session(Controller(NEW, STATES, _store), NEW, _first)) check("over HTTP, /config/model says a new install needs setup, with the spec", _r["model"]["setup_needed"] == "first_start" and _r["model"]["setup"] == json.loads(json.dumps(SPEC))) check("a save from the Installer view does not count as setup", _r["plain"][0] == 200 and _r["after_plain"]["setup_needed"] == "first_start" and uipage.SETUP_DONE not in _r["store_plain"]) check("the wizard's save writes the Store marker, and the wizard stops " "opening by itself", _r["wizard"][0] == 200 and _store.data.get(uipage.SETUP_DONE) and _r["after"]["setup_needed"] is None) _r, _sup = asyncio.run(session(Controller(NEW, STATES, Store({"auto": True}), auto=True), NEW, _first)) check("a working site's /config/model says no", _r["model"]["setup_needed"] is None) check("the page goes to Setup by itself only when opened with no route and " "only while still on Home", "first: !location.hash" in io.open(os.path.join( HERE, "app", "static", "app.js"), encoding="utf-8").read() and 'if (M && M.setup_needed && location.hash === "#home") ' 'location.hash = "#setup";' in JS) # --------------------------------------------------------------------------- # print("\n5. the last step's live checks") # --------------------------------------------------------------------------- # ROWS = {s["entity_id"]: {"entity_id": s["entity_id"], "state": s["state"], "unit": s["attributes"]["unit_of_measurement"]} for s in STATES} SITE = dict(NEW, meter_entity="sensor.grid", soc_entity="sensor.inv1_level", batt_entity="sensor.inv1_battery", pv_entity="sensor.pv") def tiers(notes): return [(n.tier, n.text) for n in notes] _c = tiers(setup_checks(SITE, entities=ROWS)) for t in _c: print(f" {t}") check("the meter reads, as the add-on will see it", ("ok", "The meter reads 402 W taken from the grid now.") in _c) check("with the sign flipped, the same reading is fed in", ("ok", "The meter reads 402 W fed into the grid now.") in tiers( setup_checks(dict(SITE, meter_invert=True), entities=ROWS))) check("the power request entity exists in Home Assistant (a write cannot " "tell: SAFETY-77)", ("ok", "Home Assistant has number.inv1_power, the power request " "entity.") in _c) check("the battery level reads, and the battery power with its direction", ("ok", "The battery level reads 62 %.") in _c and ("info", "The battery reads 800 W charging now, so check the " "inverter shows the same.") in _c) check("a solar entity below zero, not flipped, looks reversed", ("warn", "sensor.pv reads below zero now, so its solar production " "sign looks reversed.") in _c and ("ok", "sensor.pv reads 1 200 W of solar production now.") in tiers(setup_checks(dict(SITE, pv_invert_entities=["sensor.pv"]), entities=ROWS))) _miss = tiers(setup_checks(dict(SITE, setpoint_entity="number.typo"), entities=ROWS)) check("a power request entity Home Assistant does not have is red, so the " "wizard does not save", ("bad", "Home Assistant has no entity number.typo, which “Inverter " "setpoint entity” (setpoint_entity) names.") in _miss) check("a missing battery level entity is a warning, not red (the add-on " "holds 0 W and says so)", [t for t, x in tiers(setup_checks(dict(SITE, soc_entity="sensor.x"), entities=ROWS)) if "sensor.x" in x] == ["warn"]) _two = dict(DEFAULTS, meter_source="ha_signed", p1_net_entity="sensor.net", inverters=[{"name": "a", "setpoint_entity": "number.inv1_power", "soc_entity": "sensor.inv1_level", "max_w": 2000, "bms_soc_floor": 10}, {"name": "b", "setpoint_entity": "number.b", "max_w": 2000, "bms_soc_floor": 10}]) _c2 = tiers(setup_checks(_two, entities=ROWS)) check("with two inverters each is named, and a kW meter is shown as the " "add-on reads it, beside the warning to pick a sensor in W", ("ok", "The meter reads 3 W fed into the grid now.") in _c2 and [t for t, x in _c2 if "reads in kW" in x] == ["warn"] and ("bad", "Home Assistant has no entity number.b, which “Inverter " "setpoint entity” for b (inverters.setpoint_entity) names.") in _c2 and ("ok", "The battery level for a reads 62 %.") in _c2 and ("warn", "There is no battery level entity for b, so the add-on " "holds 0 W.") in _c2) _dsmr = dict(SITE, meter_source="ha_dsmr", p1_import_entity="sensor.grid", p1_export_entity="sensor.inv1_level") check("import and export are read as their difference", ("ok", "The meter reads 340 W taken from the grid now.") in tiers(setup_checks(dict(_dsmr), entities=dict( ROWS, **{"sensor.inv1_level": {"state": "62", "unit": "W"}})))) _hw = dict(SITE, meter_source="homewizard_local", p1_host="192.0.2.5") check("a meter the add-on reads itself: its live reading when it already " "runs on it, else a note that Home shows it after the restart", ("ok", "The meter reads 250 W taken from the grid now.") in tiers( setup_checks(_hw, entities=ROWS, running=_hw, meter_now_w=250)) and [t for t, x in tiers(setup_checks(_hw, entities=ROWS, running=SITE, meter_now_w=250)) if "meter" in x] == ["info"]) check("an unreadable meter is a warning", ("warn", "sensor.grid reads unavailable now, so the meter has no " "reading yet.") in tiers(setup_checks(SITE, entities=dict( ROWS, **{"sensor.grid": {"state": "unavailable", "unit": "W"}})))) _down = tiers(setup_checks(SITE, entities=None)) check("Home Assistant down: one warning, nothing claimed ok about an entity", _down[0] == ("warn", "Home Assistant is not answering, so no entity " "was checked yet.") and not [x for t, x in _down if t in ("ok", "bad") and "entity" in x]) _fixed = U.ui_to_options(dict(U.options_to_ui(SITE), **{ "ui.contract": "fixed", "ui.fallback": "fixed", "ui.fixed_price": 0.3})) _dyn = dict(SITE, price_provider="generic_dayahead") check("prices: known from a contract; for a feed, today's count when it " "already runs, else after the restart; none when none is set", ("ok", "Your prices come from your contract, so they are always known.") in tiers(setup_checks(_fixed, entities=ROWS)) and ("ok", "Prices for today are in, for 20 of 24 hours.") in tiers(setup_checks(_dyn, entities=ROWS, running=_dyn, prices_known=20)) and ("info", "Prices are fetched after the restart, and Home shows " "them.") in tiers(setup_checks(_dyn, entities=ROWS, running=SITE, prices_known=20)) and ("info", "No prices are set, so the batteries do not charge on " "price.") in _c) async def _final(s, root, sup): m = await _get(s, root + "config/model?lang=en") model = copy.deepcopy(m["ui"]) model["inverters"][0]["setpoint_entity"] = "number.typo" out = {"ok": (await _post(s, root + "config/validate", {"model": m["ui"], "setup": True}))[1], "typo": (await _post(s, root + "config/validate", {"model": model, "setup": True}))[1], "plain": (await _post(s, root + "config/validate", {"model": m["ui"]}))[1]} return out _r, _sup = asyncio.run(session(Controller(SITE, STATES, Store(), grid=402.0), SITE, _final)) check("over HTTP, the last step gets the checks beside the refusals, read " "from a fresh entity list", [x["tier"] for x in _r["ok"]["setup"] if "number.inv1_power" in x["text"]] == ["ok"] and not [m for m in _r["ok"]["messages"] if m["tier"] == "bad"] and "setup" not in _r["plain"]) check("...and a typed power request entity HA does not have is red there", ["bad"] == [x["tier"] for x in _r["typo"]["setup"] if "number.typo" in x["text"]]) async def _kettle(s, root, sup): return (await _get(s, root + "ha/entities?entity=sensor.net"), await _get(s, root + "ha/entities?entity=sensor.gone")) (_k, _gone), _sup = asyncio.run(session(Controller(SITE, STATES), SITE, _kettle)) check("the kettle test reads one entity now, in W, and says when it is not " "there", _k == {"entity_id": "sensor.net", "found": True, "state": "-3.1", "unit": "kW", "watts": -3100.0} and _gone["found"] is False and _gone["watts"] is None) # --------------------------------------------------------------------------- # print("\n6. the save goes through the P2 path and writes a valid blob") # --------------------------------------------------------------------------- # _plan = {"meter": "signed", "roster": "two", "solar": "two", "prices": "daynight", "region": "flanders", "peak": "forecast", "shadow": "on"} async def _wizard(s, root, sup): m = await _get(s, root + "config/model?lang=en") model = B.browser(B.apply(m["ui"], _plan)) v = (await _post(s, root + "config/validate", {"model": model, "setup": True}))[1] saved = await _post(s, root + "config/save", { "model": model, "fingerprint": m["fingerprint"], "setup": "done"}) return v, saved _store = Store() (_v, _saved), _sup = asyncio.run(session(Controller(NEW, STATES, _store), NEW, _wizard)) _blob = _sup.writes[-1] if _sup.writes else {} check("the wizard's model validates clean and saves with 200", not [m for m in _v["messages"] if m["tier"] == "bad"] and _saved[0] == 200 and len(_sup.writes) == 1) check("the blob written starts: the shipped validator accepts it", not validate_startup_config(_blob)) check("it has what the six steps set: the net meter, two named inverters, " "two solar rows with one flipped, day and night prices, the Flanders " "levies, the peak limit, and shadow mode on", _blob["meter_source"] == "ha_signed" and [i["name"] for i in _blob["inverters"]] == ["inverter-1", "inverter-2"] and _blob["setpoint_entity"] == "" and _blob["pv_entities"] == ["sensor.pv_a", "sensor.pv_b"] and _blob["pv_invert_entities"] == ["sensor.pv_b"] and json.loads(_blob["static_profile"])["periods"][0]["eur_per_kwh"] == 0.34 and _blob["price_taxes_eur_kwh"] == U.REGION_PRESETS["flanders"]["taxes_eur_kwh"] and _blob["peak_cap_w"] == 4200 and _blob["shadow_mode"] == "on") check("every key config.yaml declares is written, in the stored order", list(_blob)[:len(NEW)] == list(NEW) and set(_blob) == set(NEW)) check("the wizard saved without a restart of its own: the restart is a " "second, confirmed request", _sup.restarts == 0 and 'JSON.stringify({ confirm: "restart" })' in JS.split( "async function finish")[1]) _all = B.blobs() _new = [(n, r) for n, _b, r in _all if n.startswith("new-")] _ok = [n for n, _b, r in _all if not r] print(f" {len(_all)} blobs from {len(B.starts())} starts; " f"{len(_ok)} accepted") check(f"every plan from a new install is accepted by the shipped validator " f"({len(_new)} plans)", len(_new) >= 25 and not [n for n, r in _new if r]) check("every plan from the TEST-23 sites is accepted, except where the site " "trades on price with a plan that has no meter or no prices, which the " "last step then lists", all(not r or ("options.arb" in n and "arbitrage_enabled" in r[0]) for n, _b, r in _all) and len(_ok) >= 80) check("every accepted blob keeps every option key (the Supervisor count, 78)", all(set(b) == set(DEFAULTS) for _n, b, r in _all if not r)) # --------------------------------------------------------------------------- # print("\n7. a stale save names only what changed elsewhere (P2 finding a)") # --------------------------------------------------------------------------- # async def _stale(s, root, sup): m = await _get(s, root + "config/model?lang=en") sup.options["peak_cap_w"] = 5000 # the Configuration tab model = copy.deepcopy(m["ui"]) model["price_timeout_s"] = 12 # the operator's own edit first = await _post(s, root + "config/save", {"model": model, "fingerprint": m["fingerprint"]}) again = await _post(s, root + "config/save", {"model": model, "fingerprint": first[1]["fingerprint"]}) return first, again (_first_r, _again), _sup = asyncio.run(session(Controller(SITE, STATES), SITE, _stale)) print(f" changed_fields {_first_r[1]['changed_fields']}, moved " f"{_first_r[1]['moved']}, changed {_first_r[1]['changed']}") check("the banner's list is the other surface's key only, not the operator's " "own edit", _first_r[0] == 409 and _first_r[1]["changed_fields"] == ["peak_cap_w"] and _first_r[1]["moved"] == ["peak_cap_w"]) check("`changed` (stored against mine, which the classic page shows as " "theirs -> mine) is unchanged", _first_r[1]["changed"] == ["peak_cap_w", "price_timeout_s"]) check("saving again with the new token is the deliberate second press: it " "stores the page as shown (Discard loads theirs instead)", _again[0] == 200 and _sup.options["peak_cap_w"] == 3500 and _sup.options["price_timeout_s"] == 12) async def _two_pages(s, root, sup): """The classic page and this one open the same blob; the classic one saves first (the browser run found the token was dropped then).""" m = await _get(s, root + "config/model?lang=en") classic = await _get(s, root + "config/model?lang=en") theirs = dict(classic["options"], peak_cap_w=5000) await _post(s, root + "config/save", {"options": theirs, "fingerprint": classic["fingerprint"]}) model = copy.deepcopy(m["ui"]) model["price_timeout_s"] = 12 return await _post(s, root + "config/save", {"model": model, "fingerprint": m["fingerprint"]}) _r, _sup = asyncio.run(session(Controller(SITE, STATES), SITE, _two_pages)) check("two pages on one blob: after the other page saved, this one's banner " "still names only the other page's change", _r[0] == 409 and _r[1]["changed_fields"] == ["peak_cap_w"]) async def _unknown(s, root, sup): m = await _get(s, root + "config/model?lang=en") sup.options["peak_cap_w"] = 5000 model = copy.deepcopy(m["ui"]) model["price_timeout_s"] = 12 return await _post(s, root + "config/save", {"model": model, "fingerprint": "not-handed-out"}) _r, _sup = asyncio.run(session(Controller(SITE, STATES), SITE, _unknown)) check("a token the add-on did not hand out (it restarted) falls back to " "every key that differs, as before", _r[0] == 409 and _r[1]["moved"] is None and _r[1]["changed_fields"] == ["peak_cap_w", "price_timeout_s"]) # --------------------------------------------------------------------------- # print("\n8. the wizard'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 def _sentences(t): return len(re.findall(r"[.!?](?:\s|$)", t)) _long = [(p, t) for p, t in _texts(WS) if len(t.split()) > (25 if p.endswith("description") else 6) or (p.endswith("description") and _sentences(t) > 1) or ";" in t] check(f"every wizard string is one sentence of at most 25 words, or a name " f"of at most 6 ({_long})", not _long and len(list(_texts(WS))) > 50) _rendered = [x for notes in ( setup_checks(SITE, entities=ROWS), _c2 and setup_checks(_two, entities=ROWS), setup_checks(SITE, entities=None), setup_checks(_hw, entities=ROWS), setup_checks(_fixed, entities=ROWS), setup_checks(dict( SITE, soc_entity="sensor.gone", batt_entity="sensor.grid_x"), entities=dict(ROWS, **{"sensor.grid": {"state": "x", "unit": "W"}}))) for x in (n.text for n in notes)] _over = [t for t in _rendered if len(t.split()) > 25 or _sentences(t) > 1 or ";" in t] check(f"every live check, rendered with real values, is one sentence of at " f"most 25 words ({len(_rendered)} rendered, over: {_over})", len(_rendered) > 20 and not _over) _dialogs = [WS["dialog"]["description"], WS["dialog_live"]["description"]] check("both restart dialogs name the 0 W, and the shadow one says it stays", all("0 W" in d for d in _dialogs) and "shadow mode keeps them at 0 W" in _dialogs[0]) check("the wizard defaults to shadow mode: the switch starts on", "shadowOn: true" in JS) _leak = sorted(k for k in TL.config_surface(D.CONFIG)[0] if re.search( r"[^A-Za-z0-9_]" + re.escape(k) + r"[^A-Za-z0-9_]", JS)) check(f"settings.js, wizard included, names no option key ({_leak})", not _leak) print() if fails: print(f"{len(fails)} FAILED") sys.exit(1) print("all passed")