"""UX-13 P0: stored options <-> page model, exactly. `python3 test_uimodel.py` app/uimodel.py derives the keys the page does not show (the flat inverter, the solar keys, static_profile, the levies figure, arbitrage_enabled). Part 1 round-trips every options blob the test suite hands the validator (scenarios/ux13/, recorded by tools/ux13_validator_calls.py at 652097a), the TEST-23 option files and config.yaml's defaults: options -> model -> JSON -> options must be identical. Parts 2-6 edit one page field and check the blob. Mutating a module this file grades? Read MUTATION.md first (purge __pycache__). """ import glob import gzip import io import json import logging import os import re import sys HERE = os.path.dirname(os.path.abspath(__file__)) logging.disable(logging.WARNING) # the validator's own log lines from app import uimodel as U # noqa: E402 from app.price import (FLEMISH_TAXES_COMPONENTS_EUR_KWH, # noqa: E402 PRICE_OPTION_DEFAULTS) from app.static_tariff import parse_static_profile # 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) def trip(options): return U.ui_to_options(json.loads(json.dumps(U.options_to_ui(options)))) def edit(options, **fields): model = json.loads(json.dumps(U.options_to_ui(options))) for k, v in fields.items(): model[k.replace("__", ".")] = v return U.ui_to_options(model) def raises(fn): try: fn() except U.DeriveError: return True return False def defaults(): """config.yaml's `options:` block, read without PyYAML (not in CI).""" text = io.open(os.path.join(HERE, "config.yaml"), encoding="utf-8").read() block = text[text.index("\noptions:"):text.index("\nschema:")] out = {} for key, raw in re.findall(r"^ ([a-z_0-9]+):(.*)$", block, re.M): raw = raw.split(" #")[0].strip() try: out[key] = json.loads(raw) except ValueError: out[key] = raw return out DEFAULTS = defaults() SITE = dict(DEFAULTS, setpoint_entity="number.goodwe_setpoint", soc_entity="sensor.soc", batt_entity="sensor.batt") print("1. every known blob survives the round trip unchanged") _calls = [json.loads(line) for line in gzip.open(os.path.join( HERE, "scenarios", "ux13", "validator-calls-652097a.jsonl.gz"), "rt")] _test23 = sorted(glob.glob(os.path.join(HERE, "scenarios", "ux13", "test23", "*.json"))) _blobs = ([("config.yaml", DEFAULTS)] + [(c["source"], c["opts"]) for c in _calls if isinstance(c["opts"], dict)] + [(os.path.basename(f), json.load(open(f))) for f in _test23]) _bad = [] for _src, _o in _blobs: try: _back = trip(_o) except Exception as e: # noqa: BLE001 _bad.append(f"{_src}: {type(e).__name__} {e}") continue if json.dumps(_back, sort_keys=True) != json.dumps(_o, sort_keys=True): _bad.append(f"{_src}: {sorted(set(_back) ^ set(_o)) or 'values'}") print(f" {len(_calls)} validator calls, {len(_test23)} TEST-23 files, " f"config.yaml defaults") check(f"all {len(_blobs)} are identical after options -> model -> JSON -> " f"options (not: {_bad[:3]})", not _bad and len(_calls) >= 200 and len(_test23) >= 5 and len(DEFAULTS) >= 78) check("config.yaml was read whole: the defaults hold every option", "inverters" in DEFAULTS and DEFAULTS["inverters"] == [] and DEFAULTS["price_taxes_eur_kwh"] == 0.11736) _m = U.options_to_ui(DEFAULTS) check("the page model never shows a derived key: it keeps them aside", not {"setpoint_entity", "pv_entity", "pv_entities", "static_profile", "price_taxes_profile"} & set(_m) and set(_m["keep"]) >= {"static_profile", "pv_entity", "max_w"}) check("a blob that lacks keys gains none", trip({"max_w": 900}) == {"max_w": 900}) print("\n2. one inverter or a list") _one = edit(SITE, inverters=[dict(U.options_to_ui(SITE)["inverters"][0], max_w=1500)]) check("editing the one inverter writes the flat keys and keeps the list empty", _one["max_w"] == 1500 and _one["inverters"] == [] and _one["setpoint_entity"] == "number.goodwe_setpoint") _card = U.options_to_ui(SITE)["inverters"][0] _two = edit(SITE, inverters=[ dict(_card, name="inverter-a", capacity_kwh=None), dict(_card, name="inverter-b", setpoint_entity="number.goodwe_b", soc_entity="sensor.soc_b", batt_entity="sensor.batt_b")]) check("a second inverter writes the list and clears the flat power request " "entity, which the validator refuses beside a list", len(_two["inverters"]) == 2 and _two["setpoint_entity"] == "" and "capacity_kwh" not in _two["inverters"][0] and not validate_startup_config(_two)) _listed = dict(SITE, setpoint_entity="", inverters=[ {"name": "garage", "setpoint_entity": "number.g", "max_w": 2000, "bms_soc_floor": 10}]) _l1 = edit(_listed, inverters=[dict(_listed["inverters"][0], max_w=1800)]) check("a list of one stays a list, so the inverter keeps its name and stored " "state", _l1["inverters"] == [dict(_listed["inverters"][0], max_w=1800)]) check("editing a listed inverter keeps the unread flat keys as stored", all(_l1[k] == _listed[k] for k in ("setpoint_entity", "soc_entity", "batt_entity", "max_w"))) check("a name on the flat inverter is refused rather than dropped", raises(lambda: edit(SITE, inverters=[dict(_card, name="garage")]))) _raw = dict(SITE, inverters="garage") check("a roster no card can show is kept when another field is saved, and " "cannot be edited here", edit(_raw, gain=0.5)["inverters"] == "garage" and raises(lambda: edit(_raw, inverters=[_card]))) print("\n3. solar rows") check("a stored flip reads as a reversed row", U.options_to_ui(dict(DEFAULTS, pv_entity="sensor.pv", pv_invert_entities=["sensor.pv"]))["ui.solar"] == [{"entity": "sensor.pv", "reversed": True}]) check("one row writes pv_entity", edit(DEFAULTS, ui__solar=[{"entity": "sensor.pv", "reversed": False}]) ["pv_entity"] == "sensor.pv") _pv2 = edit(dict(DEFAULTS, pv_entity="sensor.a"), ui__solar=[ {"entity": "sensor.a", "reversed": False}, {"entity": "sensor.b", "reversed": True}]) check("two rows write pv_entities, clear pv_entity and flip the reversed one", _pv2["pv_entities"] == ["sensor.a", "sensor.b"] and _pv2["pv_entity"] == "" and _pv2["pv_invert_entities"] == ["sensor.b"]) _stray = dict(DEFAULTS, pv_entities=["sensor.a", "sensor.b"], pv_invert_entities=["sensor.b", "sensor.old"]) _pv1 = edit(_stray, ui__solar=[{"entity": "sensor.a", "reversed": False}]) check("removing a reversed row drops its flip, and a flip that was never a row " "stays as stored", _pv1["pv_entities"] == ["sensor.a"] and _pv1["pv_invert_entities"] == ["sensor.old"]) print("\n4. contract and prices") _fixed = edit(DEFAULTS, ui__contract="fixed", ui__fixed_price=0.31) check("a fixed price writes a profile the tariff parser reads as one price", _fixed["price_provider"] == "off" and [p.eur_per_kwh for p in parse_static_profile( _fixed["static_profile"]).periods] == [0.31] and U.options_to_ui(_fixed)["ui.contract"] == "fixed") _dn = edit(SITE, ui__contract="daynight", ui__day_price=0.34, ui__night_price=0.26) check("day and night write a profile that starts and reads back the same", not validate_startup_config(_dn) and U.options_to_ui(_dn)["ui.day_price"] == 0.34 and U.options_to_ui(_dn)["ui.night_price"] == 0.26) _dyn = edit(_dn, ui__contract="dynamic", price_provider="generic_dayahead") check("a dynamic contract keeps the day and night profile as its fallback", _dyn["price_provider"] == "generic_dayahead" and _dyn["static_profile"] == _dn["static_profile"]) check("a dynamic contract without a price source is refused here", raises(lambda: edit(DEFAULTS, ui__contract="dynamic"))) _odd = json.dumps({"periods": [ {"from": "00:00", "to": "12:00", "days": "mon-sun", "eur_per_kwh": 0.2}, {"from": "12:00", "to": "24:00", "days": "mon-sun", "eur_per_kwh": 0.3}]}) _om = U.options_to_ui(dict(DEFAULTS, static_profile=_odd)) check("odd windows show as 'custom' and survive saving another field", _om["ui.contract"] == "custom" and edit(dict(DEFAULTS, static_profile=_odd), gain=0.5) ["static_profile"] == _odd) print("\n5. region presets") _fl = U.REGION_PRESETS["flanders"] check("the Flanders preset is the sum of app/price.py's components, and is the " "shipped default", _fl["taxes_eur_kwh"] == round(sum( v for _n, v in FLEMISH_TAXES_COMPONENTS_EUR_KWH), 5) == PRICE_OPTION_DEFAULTS["price_taxes_eur_kwh"] == DEFAULTS[ "price_taxes_eur_kwh"]) check("every preset carries its year", all( isinstance(p["year"], int) and p["year"] >= 2026 for p in U.REGION_PRESETS.values())) _own = dict(DEFAULTS, price_taxes_eur_kwh=0.06) check("a figure that is no preset reads as 'own', and choosing Flanders " "writes the preset", U.options_to_ui(_own)["ui.region"] == U.OWN and edit(_own, ui__region="flanders")["price_taxes_eur_kwh"] == 0.11736) _prof = dict(DEFAULTS, price_taxes_eur_kwh=0.12, price_taxes_profile='{"day": 0.14, "night": 0.1}') check("'own' keeps a stored levies table when the figure is edited", edit(_prof, price_taxes_eur_kwh=0.13)["price_taxes_profile"] == _prof["price_taxes_profile"]) check("'own' stores the typed figure", edit(DEFAULTS, ui__region=U.OWN, price_taxes_eur_kwh=0.13) ["price_taxes_eur_kwh"] == 0.13) print("\n6. arbitrage_enabled: the OPT-11 hook changes nothing yet") check("the hook answers None while battery modes do not exist", U.arbitrage_from_batteries(U.options_to_ui(DEFAULTS)) is None) check("so the toggle is stored as the page has it", edit(DEFAULTS, arbitrage_enabled=True)["arbitrage_enabled"] is True) print("\n7. the time-window editor (UX-13 P5)") def rows(*spec): return [{"from": a, "to": b, "days": d, "price": p} for a, b, d, p in spec] def periods(r, base=SITE): return edit(base, ui__contract="custom", ui__fallback="custom", ui__periods=r) def refused(r): """(code, values) of the DeriveError the rows give, or None.""" try: periods(r) except U.DeriveError as err: return err.code, err.values return None # Five windows the day/night preset cannot say: a weekday peak, a weekday # night, Saturday split at noon, and Sunday. _five = rows(("06:30", "21:15", "mon-fri", 0.31), ("21:15", "06:30", "mon-fri", 0.19), ("00:00", "12:00", "sat", 0.2), ("12:00", "00:00", "sat", 0.25), ("00:00", "24:00", "sun", 0.18)) _p5 = periods(_five) _prof5 = parse_static_profile(_p5["static_profile"]) check("five odd windows write a profile the tariff parser reads, with every " "price in place", [p.eur_per_kwh for p in _prof5.periods] == [0.31, 0.19, 0.2, 0.25, 0.18] and _prof5.periods[3].end_min == 1440) check("...that the shipped start checks accept, and that reads back as the " "same rows", not validate_startup_config(_p5) and U.options_to_ui(_p5)["ui.contract"] == "custom" and [(r["from"], r["to"], r["days"], r["price"]) for r in U.options_to_ui(_p5)["ui.periods"]] == [("06:30", "21:15", "mon-fri", 0.31), ("21:15", "06:30", "mon-fri", 0.19), ("00:00", "12:00", "sat", 0.2), ("12:00", "24:00", "sat", 0.25), ("00:00", "24:00", "sun", 0.18)]) _named = dict(SITE, static_profile=json.dumps({"periods": [ {"name": "peak", "from": "08:00", "to": "20:00", "days": "mon-sun", "eur_per_kwh": 0.33}, {"name": "offpeak", "from": "20:00", "to": "08:00", "days": "mon-sun", "eur_per_kwh": 0.21}]})) _rows = U.options_to_ui(_named)["ui.periods"] _rows[1]["price"] = 0.2 check("a stored profile's window names are kept when a price changes", [p.name for p in parse_static_profile(periods(_rows, base=_named) ["static_profile"]).periods] == ["peak", "offpeak"]) check("the day and night preset shows as three rows, so 'Other time windows' " "starts from what is stored", [r["days"] for r in U.options_to_ui( edit(SITE, ui__contract="daynight", ui__day_price=0.3, ui__night_price=0.2))["ui.periods"]] == ["mon-fri", "mon-fri", "sat,sun"]) check("an overlap is named: the two windows, the day and the times", refused(rows(("07:00", "23:00", "mon-fri", 0.3), ("22:00", "07:00", "mon-fri", 0.2), ("00:00", "24:00", "sat,sun", 0.2))) == ("period_overlap", {"day": "mon", "from": "22:00", "to": "23:00", "a": 1, "b": 2})) check("a gap is named: the day and the times", refused(rows(("07:00", "22:00", "mon-sun", 0.3), ("22:00", "24:00", "mon-sun", 0.2))) == ("period_gap", {"day": "mon", "from": "00:00", "to": "07:00"})) check("an overlap that changes partners is named piece by piece: windows 1 " "and 2 first, over their own times only", refused(rows(("00:00", "24:00", "mon-sun", 0.3), ("06:00", "08:00", "mon-sun", 0.2), ("08:00", "10:00", "mon-sun", 0.2))) == ("period_overlap", {"day": "mon", "from": "06:00", "to": "08:00", "a": 1, "b": 2})) check("a gap on one day only is found on that day", refused(rows(("00:00", "24:00", "mon-fri", 0.3), ("00:00", "24:00", "sat", 0.2))) == ("period_gap", {"day": "sun", "from": "00:00", "to": "24:00"})) check("bad times are named by window: '7h', '25:00', '07:60', '24:00' as a " "start", all(refused(rows(("00:00", "24:00", "mon-sun", 0.2), (a, b, "mon", 0.1))) == ("period_time", {"row": 2}) for a, b in (("7h", "08:00"), ("25:00", "08:00"), ("07:60", "08:00"), ("24:00", "08:00"), ("07:00", ""))) and refused(rows(("7:00", "07:00", "mon-sun", 0.2))) == ("period_zero", {"row": 1})) check("an empty price, and a price in cents, are named by window", refused(rows(("00:00", "24:00", "mon-sun", None))) == ("period_price", {"row": 1}) and refused(rows(("00:00", "24:00", "mon-sun", 34))) == ("period_price_range", {"row": 1})) check("no windows at all is refused", refused([]) == ("no_periods", {})) from app import uipage # noqa: E402 _msg = uipage.derive_message(U.DeriveError("x", "period_overlap", "ui.periods", day="sat", a=1, b=3, **{ "from": "06:00", "to": "07:00"})) _nl = uipage.derive_message(U.DeriveError("x", "period_gap", "ui.periods", day="sun", **{"from": "00:00", "to": "07:00"}), "nl") check(f"the page shows the error beside the editor, with the day by name " f"({_msg['text']!r}, {_nl['text']!r})", _msg["field"] == "ui.periods" and "Windows 1 and 3" in _msg["text"] and "Saturday 06:00 to 07:00" in _msg["text"] and "zondag 00:00 tot 07:00" in _nl["text"]) _raw = json.dumps({"periods": [ {"from": "00:00", "to": "24:00", "days": "mon,wed", "eur_per_kwh": 0.2}, {"from": "00:00", "to": "24:00", "days": "tue,thu-sun", "eur_per_kwh": 0.3}]}) check("days the editor has no choice for stay in the Configuration tab, and " "survive a save", U.options_to_ui(dict(SITE, static_profile=_raw)) ["ui.periods"] is None and edit(dict(SITE, static_profile=_raw), gain=0.5) ["static_profile"] == _raw) print("\n8. UX-43: tariff prices take five decimals") from app import uilayout as _L # noqa: E402 _PX = 0.34512 _TARIFF = ("ui.fixed_price", "ui.day_price", "ui.night_price", "ui.periods") _steps = {k: _L.field(k).step for k in _TARIFF} check("the four tariff price fields step 0.00001 EUR/kWh", sorted(_steps) == sorted(_TARIFF) and all(v == 0.00001 for v in _steps.values())) _f5 = edit(DEFAULTS, ui__contract="fixed", ui__fixed_price=_PX) check("fixed price 0.34512 is saved exactly and read back exactly", '"eur_per_kwh": 0.34512}' in _f5["static_profile"] and U.options_to_ui(_f5)["ui.fixed_price"] == _PX and trip(_f5) == _f5) _d5 = edit(SITE, ui__contract="daynight", ui__day_price=_PX, ui__night_price=0.26789) check("day price 0.34512 and night price 0.26789 are saved and read back exactly", '"eur_per_kwh": 0.34512' in _d5["static_profile"] and '"eur_per_kwh": 0.26789' in _d5["static_profile"] and U.options_to_ui(_d5)["ui.day_price"] == _PX and U.options_to_ui(_d5)["ui.night_price"] == 0.26789 and not validate_startup_config(_d5)) _n5 = edit(SITE, ui__contract="daynight", ui__day_price=0.3, ui__night_price=_PX) check("night price 0.34512 alone is saved and read back exactly", U.options_to_ui(_n5)["ui.night_price"] == _PX) _r3 = rows(("00:00", "07:00", "mon-sun", 0.12345), ("07:00", "22:00", "mon-sun", _PX), ("22:00", "24:00", "mon-sun", 0.05001)) _p3 = periods(_r3) check("three windows with five-decimal prices save and read back exactly", [p.eur_per_kwh for p in parse_static_profile(_p3["static_profile"]).periods] == [0.12345, _PX, 0.05001] and [r["price"] for r in U.options_to_ui(_p3)["ui.periods"]] == [0.12345, _PX, 0.05001] and trip(_p3) == _p3 and "e-" not in _p3["static_profile"]) check("a two-decimal price still works", U.options_to_ui( edit(DEFAULTS, ui__contract="fixed", ui__fixed_price=0.31)) ["ui.fixed_price"] == 0.31) _old = dict(SITE, price_provider="off", static_profile=json.dumps({"periods": [ {"name": "day", "from": "07:00", "to": "22:00", "days": "mon-fri", "eur_per_kwh": 0.34}, {"name": "night", "from": "22:00", "to": "07:00", "days": "mon-fri", "eur_per_kwh": 0.26}, {"name": "night", "from": "00:00", "to": "24:00", "days": "sat,sun", "eur_per_kwh": 0.26}]})) check("a stored profile with two-decimal prices is unchanged by load and save", trip(_old) == _old) _js = open(os.path.join(HERE, "app", "static", "settings.js"), encoding="utf-8").read() check("the window editor's price input takes its step from the field, not a " "hard-coded 0.01", 'step: "0.01"' not in _js and "step: String(f.step || 0.01)" in _js) for _lang in ("en", "nl"): with open(os.path.join(HERE, "app", "ui_text", f"{_lang}.json"), encoding="utf-8") as _fh: _flds = json.load(_fh)["fields"] _hint = "34.512" if _lang == "en" else "34,512" _eq = "0.34512" if _lang == "en" else "0,34512" check(f"[{_lang}] the c€/kWh hint sits on all four tariff price fields", all(_hint in _flds[k]["description"] and _eq in _flds[k]["description"] for k in ("fixed_price", "day_price", "night_price", "periods"))) print() if fails: print(f"{len(fails)} FAILED") sys.exit(1) print("all passed")