goodwe-addon/goodwe_controller/test_settings.py

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

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]
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 next(f for _g, f in _flat if f["key"] == "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",
      next(f for _g, f in _flat if f["key"] == "peak_cap_w")["scale"] == 1000)
_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 = next(f for _g, f in _flat if f["key"] == "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 = next(f for _g, f in _flat if f["key"] == "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 = next(f for _g, f in _flat if f["key"] == "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 next(f for _g, f in _flat if f["key"] == "ui.solar")["writes"]
      == ["pv_entities", "pv_entity", "pv_invert_entities"])
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("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["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 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",
      _r[0] == 409 and _r[1]["kind"] == "stale"
      and "peak_cap_w" in _r[1]["changed_fields"] 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")