goodwe-addon/goodwe_controller/test_setup_only.py

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