"""UX-15: setup-only mode. `python3 test_setup_only.py` A new install names no inverter, and instead of exiting 78 it serves the setup wizard and commands nothing (DECISIONS/UX-15-setup-only-mode.md). Part 1 checks when it applies. Part 2 the write path refuses in that mode. Part 3 boots the REAL `app.main.main()` in a child process (this file with `--driver`) against a fake Supervisor and Home Assistant over HTTP, and logs every request they get: the shipped defaults serve the wizard with no service call, also with `auto_start` on, a Store that says control was started and a Start pressed on the page; the wizard saves; the next boot passes the normal gate and runs in shadow. Part 4: a refused but complete configuration still exits 78 and serves nothing. Takes ~25 s. Mutating a module this file grades? Read MUTATION.md first. """ import asyncio import copy import json import os import shutil import signal import socket import subprocess import sys import tempfile import time HERE = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, os.path.join(HERE, "tools")) DEADLINE_S = 60 fails = [] def check(name, cond): print(f" {'ok ' if cond else 'FAIL'} {name}") if not cond: fails.append(name) def defaults(): import ux13_demo as D return D.defaults(D.CONFIG) def state(eid, value, unit="W", cls=None, **attrs): return {"entity_id": eid, "state": value, "attributes": dict({"friendly_name": eid, "unit_of_measurement": unit, "device_class": cls}, **attrs)} STATES = [state("sensor.grid", "402"), state("number.inv1_power", "0", min=-2500, max=2500, step=1), state("sensor.inv1_level", "62", "%", "battery"), state("sensor.inv1_battery", "0")] # --------------------------------------------------------------------------- # # The child: the real main(), with only the outside world faked over HTTP. # --------------------------------------------------------------------------- # def driver(scenario, work): """`work` holds options.json, state.json and the files the parent reads. Every request the fakes get is streamed as `EV [...]` lines.""" import aiohttp from aiohttp import web as aioweb from aiohttp.abc import AbstractResolver import app.main as M import app.store import app.web import ux13_demo as D import ux13_wizard_blobs as B import translations_lint as TL def event(*e): print("EV " + json.dumps(list(e)), flush=True) opts = json.load(open(os.path.join(work, "options.json"))) stored = {"options": copy.deepcopy(opts)} schema = D.schema_ui(D.CONFIG) translations = {"en": TL.parse_translation_file( os.path.join(HERE, "translations", "en.yaml"))} # --- the fake Supervisor and Home Assistant --------------------------- async def h_probe(_r): event("GET", "/core/api/") return aioweb.Response(text="API running.") async def h_states(_r): event("GET", "/core/api/states") return aioweb.json_response(STATES) async def h_state(r): eid = r.match_info["eid"] event("GET", "/core/api/states/" + eid) found = next((s for s in STATES if s["entity_id"] == eid), None) return (aioweb.json_response(found) if found else aioweb.json_response({"message": "not found"}, status=404)) async def h_service(r): body = await r.json() event("SERVICE", r.match_info["domain"], r.match_info["service"], body) return aioweb.json_response([]) async def h_mqtt(_r): event("GET", "/services/mqtt") return aioweb.json_response({"result": "error"}, status=400) async def h_info(_r): return aioweb.json_response({"data": {"slug": "local_goodwe", "homeassistant": "2026.9.1"}}) async def h_self_info(_r): return aioweb.json_response({"data": { "slug": "local_goodwe", "options": stored["options"], "schema": schema, "translations": translations}}) async def h_options(r): body = await r.json() stored["options"] = body["options"] event("OPTIONS") with open(os.path.join(work, "saved.json"), "w") as f: json.dump(body["options"], f) return aioweb.json_response({"result": "ok", "data": {}}) async def h_restart(_r): event("RESTART") return aioweb.json_response({"result": "ok", "data": {}}) async def h_other(r): event("OTHER", r.method, r.path) return aioweb.json_response({}, status=404) fake = aioweb.Application() fake.add_routes([ aioweb.get("/core/api/", h_probe), aioweb.get("/core/api/states", h_states), aioweb.get("/core/api/states/{eid}", h_state), aioweb.post("/core/api/services/{domain}/{service}", h_service), aioweb.get("/services/mqtt", h_mqtt), aioweb.get("/info", h_info), aioweb.get("/addons/self/info", h_self_info), aioweb.post("/addons/self/options", h_options), aioweb.post("/addons/self/restart", h_restart), aioweb.route("*", "/{tail:.*}", h_other)]) fake_port = free_port() class ToFake(AbstractResolver): """`http://supervisor` is the fake, on its port.""" async def resolve(self, host, port=0, family=socket.AF_INET): return [{"hostname": host, "host": "127.0.0.1", "port": fake_port, "family": socket.AF_INET, "proto": 0, "flags": 0}] async def close(self): pass Real = aiohttp.ClientSession def session(*a, **k): k.setdefault("connector", aiohttp.TCPConnector(resolver=ToFake())) return Real(*a, **k) aiohttp.ClientSession = session os.environ["SUPERVISOR_TOKEN"] = "test-token" M.OPTIONS_PATH = os.path.join(work, "options.json") app.store.STATE_PATH = os.path.join(work, "state.json") M.INGRESS_PORT = free_port() kw = M.Controller.__init__.__kwdefaults__ kw["peak_db_path"] = os.path.join(work, "tariff.db") kw["price_history_path"] = os.path.join(work, "price_history.json") served = [] real_start = app.web.start async def start(controller, port=8099, selfcfg=None): runner = await real_start(controller, port, selfcfg) served.append(type(controller).__name__) # listening now event("SERVED", type(controller).__name__, getattr(controller.hass, "read_only", None)) return runner app.web.start = start real_amain = M.amain root = f"http://127.0.0.1:{M.INGRESS_PORT}/" async def ask(s, method, path, body=None): async with s.request(method, root + path, json=body) as r: try: out = await r.json(content_type=None) except ValueError: out = None event("PAGE", method, path, r.status, out) return r.status, out async def scenario_steps(): async with Real() as s: await ask(s, "GET", "status") _st, model = await ask(s, "GET", "config/model?lang=en") event("STATE_FILE", os.path.exists(app.store.STATE_PATH)) await ask(s, "POST", "action", {"action": "auto_toggle"}) if scenario.startswith("setup"): await ask(s, "POST", "action", {"action": "maint_start"}) wiz = B.apply(model["ui"], B.BASE) wiz = B.browser(json.loads(json.dumps(wiz))) await ask(s, "POST", "config/validate", {"model": wiz, "setup": True}) event("STATE_FILE", os.path.exists(app.store.STATE_PATH)) await ask(s, "POST", "config/save", {"model": wiz, "base": model["ui"], "fingerprint": model["fingerprint"], "setup": "done"}) await ask(s, "POST", "config/restart", {"confirm": "restart"}) await asyncio.sleep(3.0) # room for any stray task to write await ask(s, "GET", "status") async def wrapped(): runner = aioweb.AppRunner(fake) await runner.setup() await aioweb.TCPSite(runner, "127.0.0.1", fake_port).start() task = asyncio.create_task(real_amain()) try: t0 = time.monotonic() while not served and not task.done() \ and time.monotonic() - t0 < 20: await asyncio.sleep(0.05) if task.done(): event("ENDED_BEFORE_SERVING") await task return await scenario_steps() os.kill(os.getpid(), signal.SIGTERM) # the Supervisor's stop await asyncio.wait_for(task, 20) finally: await runner.cleanup() M.amain = wrapped M.main() def free_port(): with socket.socket() as s: s.bind(("127.0.0.1", 0)) return s.getsockname()[1] def run_child(scenario, options, state=None): work = tempfile.mkdtemp(prefix="ux15-") with open(os.path.join(work, "options.json"), "w") as f: json.dump(options, f) if state is not None: with open(os.path.join(work, "state.json"), "w") as f: json.dump(state, f) env = dict(os.environ, PYTHONDONTWRITEBYTECODE="1") try: p = subprocess.run([sys.executable, "-B", __file__, "--driver", scenario, work], cwd=HERE, env=env, capture_output=True, text=True, timeout=DEADLINE_S) rc, out, err = p.returncode, p.stdout, p.stderr except subprocess.TimeoutExpired as e: rc, out, err = None, e.stdout or "", e.stderr or "" out = out.decode() if isinstance(out, bytes) else out err = err.decode() if isinstance(err, bytes) else err ev = [json.loads(ln[3:]) for ln in out.splitlines() if ln.startswith("EV ")] files = {} for name in ("saved.json", "state.json"): path = os.path.join(work, name) if os.path.exists(path): files[name] = json.load(open(path)) shutil.rmtree(work, ignore_errors=True) return rc, ev, err, files def page(ev, method, path): """(status, body) of each page request, in order.""" return [(e[3], e[4]) for e in ev if e[0] == "PAGE" and e[1] == method and e[2].startswith(path)] def tail(text, n=12): return "\n".join(" | " + ln for ln in text.splitlines()[-n:]) if len(sys.argv) == 4 and sys.argv[1] == "--driver": driver(sys.argv[2], sys.argv[3]) sys.exit(0) # --------------------------------------------------------------------------- # print("1. when setup-only mode applies") # --------------------------------------------------------------------------- # from app import setuponly # noqa: E402 from app.validate import validate_startup_config # noqa: E402 SHIPPED = defaults() V = validate_startup_config check("the shipped defaults are refused only for the missing power request " "entity, and that is setup-only mode", setuponly.applies(SHIPPED, V(SHIPPED)) and len(V(SHIPPED)) == 1) _blank_bad = dict(SHIPPED, maintenance_charge_w=2500) check("no power request entity AND another refusal: not setup-only, exit 78", len(V(_blank_bad)) >= 2 and not setuponly.applies(_blank_bad, V(_blank_bad))) _complete_bad = dict(SHIPPED, setpoint_entity="number.inv1_power", maintenance_charge_w=2500) check("a complete configuration that is refused: not setup-only, exit 78", V(_complete_bad) and not setuponly.applies(_complete_bad, V(_complete_bad))) _list_blank = dict(SHIPPED, inverters=[{ "name": "inverter-a", "setpoint_entity": "", "soc_entity": "sensor.s", "batt_entity": "sensor.b", "max_w": 2000, "bms_soc_floor": 10}]) check("an inverter list whose entry has no power request entity: not " "setup-only (someone edited it), exit 78", V(_list_blank) and not setuponly.applies(_list_blank, V(_list_blank))) _ok = dict(SHIPPED, setpoint_entity="number.inv1_power") check("a startable configuration: nothing refused, the normal start", V(_ok) == [] and not setuponly.applies(_ok, V(_ok))) # --------------------------------------------------------------------------- # print("\n2. in setup-only mode the write path refuses") # --------------------------------------------------------------------------- # from app import hass as H # noqa: E402 import app.main as M # noqa: E402 class Session: """Records every request a HomeAssistant would send.""" def __init__(self): self.sent = [] def post(self, url, **k): self.sent.append(("POST", url)) raise AssertionError("a request was sent") def get(self, url, **k): self.sent.append(("GET", url)) raise AssertionError("a request was sent") def _refused(coro): """True only for the read-only refusal; a request that reached the session (AssertionError there) or no refusal at all is False.""" try: asyncio.run(coro) except H.ReadOnly: return True except Exception: # noqa: BLE001 return False return False _s = Session() _ro = H.HomeAssistant(_s, "t", read_only=True) check("a read-only HomeAssistant refuses number.set_value before it sends " "anything", _refused(_ro.set_number("number.inv1_power", 0.0)) and _s.sent == []) check("...and any other service call, e.g. a notification", _refused(_ro.call("persistent_notification", "create", {"message": "x"})) and _s.sent == []) try: _ro.read_only = False _flipped = True except AttributeError: _flipped = False check("...and read-only cannot be switched off afterwards", not _flipped and _ro.read_only is True) _rw = H.HomeAssistant(Session(), "t") try: asyncio.run(_rw.set_number("number.inv1_power", 0.0)) _sent = False except AssertionError: _sent = True check("control: a normal HomeAssistant does send the write", _sent and _rw.read_only is False) try: M.Controller(dict(_ok), _ro, None, None, stop=asyncio.Event()) _built = True except ValueError as err: _built = "read-only" not in str(err) except Exception: # noqa: BLE001 _built = True # it got past the check and failed on something else check("no controller can be built on a read-only HomeAssistant", not _built) try: setuponly.SetupOnly(SHIPPED, _rw, None) _took = True except ValueError: _took = False check("setup-only mode refuses a HomeAssistant that can write", not _took) _so = setuponly.SetupOnly(SHIPPED, _ro, None) check("setup-only mode holds no inverter and never says control is started", list(_so.inverters) == [] and _so.auto is False and _so.status_payload()["setup_only"] is True) # --------------------------------------------------------------------------- # print("\n3. the real main(): the shipped defaults serve the wizard, 0 writes") # --------------------------------------------------------------------------- # _runs = { "setup": (dict(SHIPPED), None), # A started-control attempt three ways: auto_start on, a Store that says # control was started, and the page's own Start (the driver presses it). "setup_started": (dict(SHIPPED, auto_start=True), {"auto": True}), } SAVED = None for name, (options, st) in _runs.items(): rc, ev, err, files = run_child(name, options, st) services = [e for e in ev if e[0] == "SERVICE"] print(f" {name}: exit {rc}, {len(ev)} requests logged, " f"{len(services)} service calls") check(f"{name}: the process served the page in setup-only mode on a " "read-only HomeAssistant, and exited 0 when stopped", rc == 0 and ["SERVED", "SetupOnly", True] in ev) if rc != 0: print(tail(err)) check(f"{name}: 0 service calls to Home Assistant (no set_value, no " "notification), and MQTT was never asked for", services == [] and ["GET", "/services/mqtt"] not in ev) _status = page(ev, "GET", "status") check(f"{name}: /status says setup_only, with the setup banner", _status and all(b and b.get("setup_only") is True and b.get("mode") == "setup_only" and b.get("banner", "").startswith("Setup") for _c, b in _status)) _model = page(ev, "GET", "config/model") check(f"{name}: the page model opens the wizard (no power request) and " "says setup-only", _model and _model[0][1]["setup_needed"] == "no_power_request" and _model[0][1]["setup_only"] is True) _acts = page(ev, "POST", "action") check(f"{name}: Start and the battery cycle are refused with 409", len(_acts) == 2 and all(c == 409 for c, _b in _acts)) check(f"{name}: setup-only mode writes no Store file before the wizard " "saves, so the first-run rule still sees a new install", [e for e in ev if e[0] == "STATE_FILE"][:2] == [["STATE_FILE", st is not None]] * 2) _val = page(ev, "POST", "config/validate") check(f"{name}: the wizard's live checks run and find nothing red", _val and _val[0][0] == 200 and not [m for m in _val[0][1]["messages"] if m["tier"] == "bad"] and not [m for m in _val[0][1]["setup"] if m["tier"] == "bad"]) _save = page(ev, "POST", "config/save") check(f"{name}: the wizard saves (200) and asks for the restart", _save and _save[0][0] == 200 and ["RESTART"] in ev and "saved.json" in files) check(f"{name}: the Store gets setup_done from the save", bool(files.get("state.json", {}).get("setup_done"))) if name == "setup": SAVED, SAVED_STATE = files.get("saved.json"), files.get("state.json") check("the saved blob passes the normal gate: no refusal, not setup-only", SAVED is not None and V(SAVED) == [] and not setuponly.applies(SAVED, V(SAVED)) and SAVED.get("shadow_mode") == "on") print("\n after the restart, on the saved settings:") rc, ev, err, files = run_child("after", SAVED or {}, SAVED_STATE) # Not the dismiss of PRICE-01's old fixed-tariff card, sent at every start. # Only that dismiss: a `create` with the old id would still be counted. services = [e for e in ev if e[0] == "SERVICE" and not (e[1:3] == ["persistent_notification", "dismiss"] and e[3].get("notification_id") == "goodwe_fixed_tariff_unused")] writes = [e[3].get("value") for e in services if e[1:3] == ["number", "set_value"]] print(f" after: exit {rc}, {len(services)} service calls, set_value " f"values {sorted(set(writes))}") check("the next boot passes the normal gate: a Controller serves the page, " "and it exits 0 when stopped", rc == 0 and ["SERVED", "Controller", False] in ev) if rc != 0: print(tail(err)) check("...where Start is accepted (200), unlike in setup-only mode", [c for c, _b in page(ev, "POST", "action")] == [200]) _status = page(ev, "GET", "status") check("...its /status is not setup-only", _status and all( b and "setup_only" not in b for _c, b in _status)) _model = page(ev, "GET", "config/model") check("...and the wizard does not open again: setup_done is in the Store", _model and _model[0][1]["setup_needed"] is None and _model[0][1]["setup_only"] is False) check("...and it runs in shadow, even after Start: every service call is " "number.set_value 0 W", writes and len(writes) == len(services) and all(w == 0.0 for w in writes)) # --------------------------------------------------------------------------- # print("\n4. a refused configuration that is not only missing setup: exit 78") # --------------------------------------------------------------------------- # for name, options in (("refused", _complete_bad), ("refused_blank", _blank_bad)): rc, ev, err, _files = run_child(name, options) print(f" {name}: exit {rc}") check(f"{name}: exit 78 and nothing served, no service call", rc == 78 and not [e for e in ev if e[0] in ("SERVED", "SERVICE")] and ["ENDED_BEFORE_SERVING"] in ev) print() if fails: print(f"{len(fails)} FAILED") sys.exit(1) print("all passed")