"""Runnable check for TEL-07 - the SECOND MQTT device, and A's history. `python3 test_mqtt.py` WHAT THIS FILE IS FOR, in one sentence: to prove by DIFFING THE ANNOUNCED PAYLOADS - not by reading the code - that giving inverter B its own MQTT device changes nothing about inverter A's. ⚠️ WHY THAT IS THE WHOLE CARD. `app/mqtt.py`'s own note records the finding: Home Assistant composes an entity_id from the DEVICE name plus the ENTITY name, and `object_id` in the discovery payload did NOT override it (tested on HA 2026.8). So re-scoping the device renames every entity - and because HA keeps the id it composed at FIRST discovery, what a rename actually produces is a NEW entity beside a DEAD one. The recorded history stays behind on the old id, and every dashboard, template and automation that names it - including the three live watchdog automations `goodwe_master_watchdog_held_*` and `goodwe_master_parked_at_0_w_*` - goes on pointing at an entity that has quietly stopped updating. Nothing errors. That is the failure this file exists to make impossible to ship. ⚠️ AND THE TEMPTING DESIGN IS THE DESTRUCTIVE ONE. "One device per inverter" - A becomes `inverter-a`, B becomes `inverter-b` - is symmetrical and tidy and is exactly the change described above. A KEEPS THE DEVICE IT ALREADY HAS; only B gets a new one. Part 8 below writes that tidy version and shows it going red. THE SHAPE OF THE EVIDENCE. Part 1 records what `release/1.0` actually publishes - generated by running THAT commit's module, not transcribed - and every later part is a diff against it, reported as `N topics, M missing, C changed, A added`. Part 1 The recorded baseline, and the diff at a roster of one and of none. Part 2 `_announce()` publishes EXACTLY `discovery_payloads()` and nothing else, so parts 1 and 3-8 are diffing what goes on the wire. Part 3 A roster of two: A's nine messages byte-identical, eight added. Part 4 Zero overlap between A's entities and B's - topics, unique_ids, object_ids, state topics, and devices. Part 5 Site-scoped sensors stay site-scoped, at every roster length 1-6. Part 6 The composed entity_ids, modelled INDEPENDENTLY of `app/mqtt.py`, and anchored on the one composed id that was verified against the live system (`sensor.goodwe_p1_sample_age`, mqtt.py's own note). Part 7 ⚠️ THE ANNOUNCEMENT AND THE PUBLICATION AGREE, on a real `Controller`, with the two units carrying DIFFERENT numbers - so a swap is visible. Part 8 ⚠️ VERIFIED BY BREAKING IT. Twelve mutations of the shipped module, re-executed from mutated source, each one named with the property it breaks. A property nothing breaks is a property nothing is watching. Part 9 What these checks CANNOT see, enumerated and sized. Part 10 OPT-11 P2: each battery's select and switch, commands, removal. ⚠️ THE FIXTURE GIVES A AND B DIFFERENT NAMES, DIFFERENT RATINGS, DIFFERENT SoC, DIFFERENT BATTERY POWER AND DIFFERENT LATCH STATES, ON PURPOSE. A fixture whose two units are equal cannot observe a bug that swaps them - MUTATION.md habit 4, and the defect this repository has now shipped six cards in a row. ⚠️ INVERTER B DOES NOT PHYSICALLY EXIST (~2026-09-12). Every number attributed to it below is SIMULATED, from the datasheet and serial recorded in NOTES.md §1.1b (GW5048D-ES, serial 55048ESU22AW0992, max_w 4600 W). Nothing here is measured, and nothing here should ever be quoted as if it were. ⚠️ MUTATING THE MODULE THIS FILE GRADES? PURGE `__pycache__` FIRST - or run `tools/mutation_run.sh`, which does. `python3 -B` alone does NOT save you: it stops bytecode being WRITTEN, not a stale one being READ. Read MUTATION.md. Part 8 sidesteps the whole problem by exec'ing mutated SOURCE into a fresh namespace, so no `.pyc` is involved at any point. """ import ast import asyncio import hashlib import json import re import sys from pathlib import Path from app import mqtt from app.inverter import build_roster, roster_names from app.main import Controller HERE = Path(__file__).resolve().parent MQTT_SRC = (HERE / "app" / "mqtt.py").read_text(encoding="utf-8") fails = [] def check(name, cond): print(f" {'ok ' if cond else 'FAIL'} {name}") if not cond: fails.append(name) # --------------------------------------------------------------------------- # The recorded baseline. ⚠️ GENERATED, NOT TRANSCRIBED. Reproduce it with: # # git show release/1.0:goodwe_controller/app/mqtt.py > /tmp/mqtt_base.py # python3 - <<'EOF' # import sys; sys.path.insert(0, "/tmp"); import mqtt_base as M # class F: # def __init__(self): self.m = [] # def publish(self, t, p=None, retain=False): self.m.append((t, p, retain)) # p = M.MqttPublisher.__new__(M.MqttPublisher) # p.omit, p.client, p.enabled = set(), F(), True # p._announce() # for row in p.client.m: print(row) # EOF # # A transcription would be a check that grades this branch against MY MEMORY of # release/1.0 rather than against release/1.0. # --------------------------------------------------------------------------- _DEV = ('{"identifiers": ["goodwe_rs485_controller"], "name": "GoodWe", ' '"manufacturer": "GoodWe (via RS485 meter emulation)", ' '"model": "ES/BP series"}') BASELINE = [ ('homeassistant/sensor/goodwe_ctl_setpoint/config', '{"name": "Setpoint", "object_id": "goodwe_setpoint", "unique_id": "goodwe_ctl_setpoint", ' '"state_topic": "goodwe_ctl/setpoint", "availability_topic": "goodwe_ctl/availability", ' '"device": ' + _DEV + ', "unit_of_measurement": "W", "device_class": "power", ' '"state_class": "measurement"}', True), ('homeassistant/sensor/goodwe_ctl_grid/config', '{"name": "Grid power", "object_id": "goodwe_grid_power", "unique_id": "goodwe_ctl_grid", ' '"state_topic": "goodwe_ctl/grid", "availability_topic": "goodwe_ctl/availability", ' '"device": ' + _DEV + ', "unit_of_measurement": "W", "device_class": "power", ' '"state_class": "measurement"}', True), ('homeassistant/sensor/goodwe_ctl_battery/config', '{"name": "Battery power", "object_id": "goodwe_battery_power", "unique_id": "goodwe_ctl_battery", ' '"state_topic": "goodwe_ctl/battery", "availability_topic": "goodwe_ctl/availability", ' '"device": ' + _DEV + ', "unit_of_measurement": "W", "device_class": "power", ' '"state_class": "measurement"}', True), ('homeassistant/sensor/goodwe_ctl_soc/config', '{"name": "Battery SoC", "object_id": "goodwe_battery_soc", "unique_id": "goodwe_ctl_soc", ' '"state_topic": "goodwe_ctl/soc", "availability_topic": "goodwe_ctl/availability", ' '"device": ' + _DEV + ', "unit_of_measurement": "%", "device_class": "battery", ' '"state_class": "measurement"}', True), ('homeassistant/sensor/goodwe_ctl_circulation/config', '{"name": "Circulating power", "object_id": "goodwe_circulating_power", ' '"unique_id": "goodwe_ctl_circulation", "state_topic": "goodwe_ctl/circulation", ' '"availability_topic": "goodwe_ctl/availability", "device": ' + _DEV + ', ' '"unit_of_measurement": "W", "device_class": "power", "state_class": "measurement", ' '"icon": "mdi:sync-alert"}', True), ('homeassistant/sensor/goodwe_ctl_phase/config', '{"name": "Maintenance phase", "object_id": "goodwe_maintenance_phase", ' '"unique_id": "goodwe_ctl_phase", "state_topic": "goodwe_ctl/phase", ' '"availability_topic": "goodwe_ctl/availability", "device": ' + _DEV + ', ' '"icon": "mdi:battery-sync"}', True), ('homeassistant/sensor/goodwe_ctl_status/config', '{"name": "Controller status", "object_id": "goodwe_controller_status", ' '"unique_id": "goodwe_ctl_status", "state_topic": "goodwe_ctl/status", ' '"availability_topic": "goodwe_ctl/availability", "device": ' + _DEV + ', ' '"icon": "mdi:heart-pulse"}', True), ('homeassistant/sensor/goodwe_ctl_p1_age/config', '{"name": "P1 sample age", "object_id": "p1_sample_age_s", "unique_id": "goodwe_ctl_p1_age", ' '"state_topic": "goodwe_ctl/p1_age", "availability_topic": "goodwe_ctl/availability", ' '"device": ' + _DEV + ', "unit_of_measurement": "s", "device_class": "duration", ' '"state_class": "measurement"}', True), ('goodwe_ctl/availability', 'online', True), ] # --- TEL-10's three, and they are DELIBERATELY NOT IN `BASELINE` ------------- # # ⚠️ `BASELINE` IS WHAT release/1.0 PUBLISHES AND NOTHING MAY BE ADDED TO IT. # Every strong claim in this file rests on that list being a record of another # tree; append a row this branch invented and "byte-identical to release/1.0" # quietly becomes "identical to itself", which is the defect MUTATION.md's # habit 4 is about - a check whose data cannot reach the failure it names. # # So TEL-10's arrival is TWO claims and gets two checks: the nine release/1.0 # messages did not move, and EXACTLY these three arrived after them. # # ⚠️ AND THIS LIST IS A REGRESSION PIN, NOT EVIDENCE OF CORRECTNESS - the same # status as part 3's sha256 over the per-unit payloads, for the same reason: # there is no released version of these bytes to diff against, so all it can # say is "these three payloads have not changed since the PR was reviewed". _TEL10_NUM = ('"availability_topic": "goodwe_ctl/availability", "device": ' + _DEV + ', "unit_of_measurement": "W", "device_class": "power", ' '"state_class": "measurement"}') TEL10 = [ ('homeassistant/sensor/goodwe_ctl_loop_i/config', '{"name": "Loop integrator", "object_id": "goodwe_loop_integrator", ' '"unique_id": "goodwe_ctl_loop_i", "state_topic": "goodwe_ctl/loop_i", ' + _TEL10_NUM, True), ('homeassistant/sensor/goodwe_ctl_loop_lead/config', '{"name": "Loop integrator lead", "object_id": "goodwe_loop_integrator_lead", ' '"unique_id": "goodwe_ctl_loop_lead", "state_topic": "goodwe_ctl/loop_lead", ' + _TEL10_NUM, True), ('homeassistant/sensor/goodwe_ctl_loop_reason/config', '{"name": "Loop reason", "object_id": "goodwe_loop_reason", ' '"unique_id": "goodwe_ctl_loop_reason", "state_topic": "goodwe_ctl/loop_reason", ' '"availability_topic": "goodwe_ctl/availability", "device": ' + _DEV + ', ' '"icon": "mdi:sigma"}', True), ] # What THIS branch announces at a roster of one: release/1.0's eight sensor # configs, then TEL-10's three, then the availability message - which stays # LAST, because `_announce()` publishes it after every config and a broker # replaying the retained set must see availability arrive after the entities it # is the availability OF. SITE_NOW = BASELINE[:-1] + TEL10 + BASELINE[-1:] class FakeClient: """Captures what `_announce()` actually hands paho.""" def __init__(self): self.msgs = [] def publish(self, topic, payload=None, retain=False): self.msgs.append((topic, payload, retain)) def announce(units=(), omit=(), module=mqtt, include=()): """The messages `_announce()` publishes, through the SHIPPED method. ⚠️ Not `discovery_payloads()` directly. Part 2 proves the two agree; every other part goes through this one, so what is diffed is what paho is told. `include` is TEL-11's optional-row set and DEFAULTS TO EMPTY here for the same reason it does in the shipped constructor: every check in this file written before that key existed must go on grading the announcement an install that never set `pv_entity` receives, which is all of them. """ pub = module.MqttPublisher.__new__(module.MqttPublisher) pub.omit = set(omit) pub.units = [str(u) for u in units] pub.include = [str(k) for k in include] pub.enabled = True pub.client = FakeClient() pub._announce() return pub.client.msgs def diff(got, want=SITE_NOW): """`(missing, changed, added)` between two announcement lists, by topic.""" g, w = dict((t, (p, r)) for t, p, r in got), dict((t, (p, r)) for t, p, r in want) missing = [t for t in w if t not in g] changed = [t for t in w if t in g and g[t] != w[t]] added = [t for t in g if t not in w] return sorted(missing), sorted(changed), sorted(added) def report(label, got, want=SITE_NOW): m, c, a = diff(got, want) print(f" {label}: {len(want)} topics, {len(m)} missing, {len(c)} changed, " f"{len(a)} added") for t in m: print(f" MISSING {t}") for t in c: print(f" CHANGED {t}") for t in a: print(f" ADDED {t}") return m, c, a def cfgs(msgs): """The discovery payloads, parsed, without the availability message.""" return [json.loads(p) for t, p, _ in msgs if t.startswith("homeassistant/")] # The roster this card exists for. ⚠️ SIMULATED: inverter B does not physically # exist. Its rating is the datasheet's (NOTES.md §1.1b, GW5048D-ES, serial # 55048ESU22AW0992) and has never been measured. OPTS2 = { "meter_source": "off", "inverters": [ {"name": "inverter-a", "setpoint_entity": "number.sp_a", "soc_entity": "sensor.soc_a", "batt_entity": "sensor.batt_a", "max_w": 2000, "bms_soc_floor": 10}, {"name": "inverter-b", "setpoint_entity": "number.sp_b", "soc_entity": "sensor.soc_b", "batt_entity": "sensor.batt_b", "max_w": 4600, "bms_soc_floor": 10}, ], } NAMES2 = roster_names(OPTS2) # --------------------------------------------------------------------------- print("\n1. the recorded baseline, and a roster of one") # --------------------------------------------------------------------------- # AC 1 and AC 4. ⚠️ THE COMPARISON IS BYTE-FOR-BYTE ON THE PAYLOAD STRING, not # on a parsed dict. `json.dumps` writes keys in insertion order, so a reordered # payload is a different retained message on the broker even though the parsed # dicts compare equal - and a reordered payload is a re-announcement of every # entity on a live install. m, c, a = report("roster of one (`inverters:` empty, i.e. every install today)", announce(units=roster_names({}))) # ⚠️ THE HEADLINE SAYS EIGHT, NOT NINE, AND THE SLICE UNDER IT IS WHY. # release/1.0 announces NINE messages; the ninth is `goodwe_ctl/availability`, # which `_announce()` publishes LAST and which TEL-10 therefore MOVES - it is # byte-identical but no longer at index 8. So what is asserted here is the eight # entity CONFIGS, `[:len(BASELINE) - 1] == BASELINE[:-1]`, and the word "nine" # above it named an assertion this file does not make. The ninth is not # unguarded: the `== SITE_NOW` on the next check pins it byte-for-byte in its # new last position. Two sentences, two checks. check("⭐ AT THE SHIPPED ROSTER OF ONE EVERY ONE OF release/1.0's EIGHT ENTITY " "CONFIGS IS STILL BYTE-IDENTICAL AND STILL FIRST: same topics, same " "unique_ids, same payload bytes, same retain flags, in the same order - " "no second device, no renamed entity, nothing re-ordered. TEL-10 adds " "rows AFTER them; it may not move one of them. (release/1.0's NINTH " "message, `availability`, is byte-identical too and deliberately moves " "to LAST; the `== SITE_NOW` below is what pins it)", announce(units=roster_names({}))[:len(BASELINE) - 1] == BASELINE[:-1]) # ⚠️ AND THE ADDITION IS ITS OWN CLAIM, WITH A COUNT. "Nothing moved" and # "exactly three arrived" are different sentences and a single `== SITE_NOW` # would let a fourth row hide inside whichever of the two a reader had in mind. _m1, _c1, _a1 = diff(announce(units=roster_names({})), BASELINE) check("⭐ TEL-10 ADDS EXACTLY THREE TOPICS AND CHANGES NONE: the integrator, " "the lead and the loop's own reason string, all NEW, all on the SITE " "device - nothing of release/1.0's is missing and nothing of it changed", (_m1, _c1) == ([], []) and sorted(_a1) == sorted(t for t, _p, _r in TEL10) and announce(units=roster_names({})) == SITE_NOW) check("...and the roster-of-one path is the one every install actually reads: " "an empty `inverters:` names exactly one unit", roster_names({}) == ["inverter-a"]) check("...and a publisher told NOTHING about the roster announces the same " "thing, so a caller that has not been taught about units cannot change " "what a live install sees", announce(units=()) == SITE_NOW) m, c, a = report("roster of one, P1 off (`omit=('p1_age',)`)", announce(units=roster_names({}), omit=("p1_age",)), [row for row in SITE_NOW if "p1_age" not in row[0]]) check("...and the `omit` path is unchanged too: with P1 ingestion off the " "watchdog input is still not announced, and nothing else moved", (m, c, a) == ([], [], [])) # --- TEL-11 ------------------------------------------------------------------ # ⚠️ THE OPTIONAL ROW'S DEFAULT IS THE CLAIM, AND IT IS THE OPPOSITE DEFAULT # FROM `omit`'s. `p1_age` is announced unless withdrawn; `pv` does not exist # unless asked for. Every check above this line passes `include` nothing, so # they are all grading the announcement an install that never set `pv_entity` # receives - which is MOST installs, and is the acceptance criterion "unset # behaves exactly as today". check("⭐ TEL-11 ADDS NOTHING AT THE SHIPPED DEFAULT: a publisher that is told " "nothing about `pv_entity` announces the same list, in the same order, " "with the same bytes - so an install that never sets the option sees no " "new entity, no new topic and no re-announcement of an old one", announce(units=roster_names({})) == SITE_NOW) _PV_TOPIC = "homeassistant/sensor/goodwe_ctl_pv/config" _PV_PAYLOAD = ('{"name": "PV power", "object_id": "goodwe_pv_power", ' '"unique_id": "goodwe_ctl_pv", "state_topic": "goodwe_ctl/pv", ' '"availability_topic": "goodwe_ctl/availability", "device": ' + _DEV + ', "unit_of_measurement": "W", "device_class": ' '"power", "state_class": "measurement"}') _pv_one = announce(units=roster_names({}), include=("pv",)) _m2, _c2, _a2 = diff(_pv_one, SITE_NOW) check("⭐ AND WITH `pv_entity` SET IT IS EXACTLY ONE NEW TOPIC AND NOTHING " "ELSE MOVES: same eleven configs first, same bytes, same order - the " "additions this card makes are additions, not a re-composition", (_m2, _c2) == ([], []) and _a2 == [_PV_TOPIC] and _pv_one[:len(SITE_NOW) - 1] == SITE_NOW[:-1]) check("...and the payload is the site device, in watts, with " "`state_class: measurement` - which is what puts it in the recorder's " "statistics, and a week of statistics is the entire point of the card", (_PV_TOPIC, _PV_PAYLOAD, True) in _pv_one) # ⚠️ AND IT GOES LAST, WHICH IS WHY A ROSTER OF TWO IS ASKED THE SAME QUESTION. # Placed among the site rows it would shift every per-unit entity's position in # the retained announcement - a zero-byte edit that re-announces inverter B. _pv_two = announce(units=NAMES2, include=("pv",)) check("...and at a roster of two it is STILL the last config, after every " "per-unit row, so no existing entity's position in the retained " "announcement moves when the option is turned on", [t for t, _p, _r in _pv_two if t.startswith("homeassistant/")][-1] == _PV_TOPIC and _pv_two == announce(units=NAMES2)[:-1] + [ (_PV_TOPIC, _PV_PAYLOAD, True)] + announce(units=NAMES2)[-1:]) check("...and `omit` still wins over `include`, so the two arguments cannot " "disagree about one key", [t for t, _p, _r in announce(units=roster_names({}), include=("pv",), omit=("pv",)) if t == _PV_TOPIC] == []) # --------------------------------------------------------------------------- print("\n2. what is diffed is what goes on the wire") # --------------------------------------------------------------------------- # ⚠️ WITHOUT THIS PART, PARTS 1 AND 3-8 GRADE A FUNCTION THE ADD-ON MIGHT NOT # USE. `_announce()` used to build its payloads inline; if it still did, every # diff here would be a diff of a second implementation that happens to sit in # the same file. This is the join. _direct = [(t, json.dumps(cfg), True) for t, cfg in mqtt.discovery_payloads(NAMES2)] + \ [(mqtt.AVAILABILITY, "online", True)] check("`_announce()` publishes EXACTLY `discovery_payloads()`, in order, with " "retain set, plus the availability message - so every diff in this file " "is a diff of the bytes paho is handed", announce(units=NAMES2) == _direct) # --------------------------------------------------------------------------- print("\n3. a roster of two: A untouched, B added") # --------------------------------------------------------------------------- TWO = announce(units=NAMES2) m, c, a = report("roster of two (inverter-a + inverter-b, SIMULATED)", TWO) check("⭐ AC 1: AT A ROSTER OF TWO, EVERY ONE OF A's NINE MESSAGES IS STILL " "BYTE-IDENTICAL - nothing missing, nothing changed. This is the criterion " "the live install's recorded history depends on, and it is a diff of the " "payloads, not a reading of the code", (m, c) == ([], [])) check("...and the eight additions are all B's and A's own per-unit rows, and " "all of them are NEW topics", len(a) == 8) check("...and A's eight entity configs are still the FIRST eight, in the same " "order, so a broker replaying retained messages sees them unchanged " "(the ninth, `availability`, is byte-identical and last - the check " "above is the one that covers it, as a diff rather than a prefix)", TWO[:len(BASELINE) - 1] == BASELINE[:-1]) # ⚠️ THE SITE SET IS `SITE_NOW`, NOT `BASELINE`. Read from `BASELINE` it would # be short by TEL-10's three, and those three would then be counted as PER-UNIT # rows by every line below that says "not a site id" - which silently changes # what the sha256 pin, the uniqueness checks and part 7's fed/announced set are # taken over. site_ids = {c_["unique_id"] for c_ in cfgs(SITE_NOW)} unit_cfgs = [c_ for c_ in cfgs(TWO) if c_["unique_id"] not in site_ids] a_cfgs = [c_ for c_ in unit_cfgs if "inverter_a" in c_["unique_id"]] b_cfgs = [c_ for c_ in unit_cfgs if "inverter_b" in c_["unique_id"]] check("A's per-unit rows stay on THE DEVICE A ALREADY HAS - no new device is " "created for the first roster entry, which is the whole asymmetry", len(a_cfgs) == 4 and all(c_["device"] == mqtt.DEVICE for c_ in a_cfgs)) check("⭐ AC 2: B's rows are on B's OWN device, with its own identifiers", len(b_cfgs) == 4 and all(c_["device"]["identifiers"] == ["goodwe_rs485_controller_inverter_b"] for c_ in b_cfgs)) check("...and B's device hangs off the controller's via `via_device`, so HA " "shows one controller driving two inverters rather than two strangers", all(c_["device"].get("via_device") == mqtt.DEVICE["identifiers"][0] for c_ in b_cfgs)) check("...and B's device carries the SAME manufacturer and model as the site " "device - it is the same installation, not a second product", all(c_["device"]["manufacturer"] == mqtt.DEVICE["manufacturer"] and c_["device"]["model"] == mqtt.DEVICE["model"] for c_ in b_cfgs)) # A regression pin on the NEW payloads. ⚠️ THIS IS NOT EVIDENCE OF CORRECTNESS - # it is generated from this branch, so it can only say "these bytes have not # changed since the PR was reviewed". It exists because B's payloads have no # released version to be diffed against, and a later edit that quietly changes # a per-unit `device_class` or state topic should have to be deliberate. _pin = hashlib.sha256( json.dumps(unit_cfgs, sort_keys=False).encode()).hexdigest()[:16] print(" the eight per-unit payloads, in announcement order:") for c_ in unit_cfgs: print(f" {c_['unique_id']:<40} {c_['object_id']:<34} " f"{c_['state_topic']}") print(f" regression pin (sha256 of the eight payloads): {_pin}") check("the per-unit payloads are the ones this PR was reviewed with " "(REGRESSION PIN, not a correctness claim - see the comment)", _pin == "eb5022b0455158d7") # --------------------------------------------------------------------------- print("\n4. zero overlap between A's entities and B's") # --------------------------------------------------------------------------- def field(rows, key): return [r[key] for r in rows] for key in ("unique_id", "object_id", "state_topic"): av, bv = set(field(a_cfgs, key)), set(field(b_cfgs, key)) sv = set(field(cfgs(SITE_NOW), key)) check(f"⭐ AC 2: no `{key}` is shared between A's entities and B's, and " f"none of B's collides with a SITE entity that already exists", len(av) == 4 and len(bv) == 4 and not (av & bv) and not (bv & sv) and not (av & sv)) all_ids = field(cfgs(TWO), "unique_id") check("...and across the WHOLE announcement every unique_id is distinct - an " "HA discovery collision is not an error, it DROPS the second entity, so " "one inverter's telemetry would silently stop existing", len(all_ids) == len(set(all_ids)) == 19) all_topics = [t for t, _, _ in TWO] check("...and so is every topic", len(all_topics) == len(set(all_topics)) == 20) # ⚠️ THE SLUG COLLISION rule 9(c) DOES NOT REFUSE. `inverter-b` and `Inverter B` # are two different names by the validator's test and one string once slugged. COLLIDE = announce(units=["inverter-b", "Inverter B"]) col_ids = field(cfgs(COLLIDE), "unique_id") print(" two names that slug alike ('inverter-b', 'Inverter B') ->", [c_["unique_id"] for c_ in cfgs(COLLIDE) if c_["unique_id"] not in site_ids]) check("two roster names that differ only in case and punctuation - which " "`validate.py` rule 9(c) accepts, because it refuses only EXACT " "duplicates - still get distinct unique_ids", len(col_ids) == len(set(col_ids))) # --------------------------------------------------------------------------- print("\n5. site-scoped sensors stay site-scoped") # --------------------------------------------------------------------------- # ⚠️ `p1_sample_age_s` IS A WATCHDOG INPUT (mqtt.py's `omit` note and the ESP32 # firmware's ingest guard). A second copy of it is a second answer to "is the # meter fresh?", one of which is always the one nobody is looking at. `grid` is # the same argument without the safety edge: there is ONE grid connection. # ⚠️ TEL-10's THREE ARE IN HERE, AND THE ARGUMENT IS `grid`'s. There is ONE # integrator - §6.2's architecture is one integrator on the P1 meter whose # output is then SPLIT - so a per-inverter copy of it would invent N of a thing # there is one of, and two of them would disagree the moment the allocator gave # the units different shares. SITE_ONLY = ("goodwe_ctl_grid", "goodwe_ctl_p1_age", "goodwe_ctl_circulation", "goodwe_ctl_phase", "goodwe_ctl_status", "goodwe_ctl_loop_i", "goodwe_ctl_loop_lead", "goodwe_ctl_loop_reason") rows = [] for n in range(1, 7): names = [f"inverter-{chr(ord('a') + i)}" for i in range(n)] ids = field(cfgs(announce(units=names)), "unique_id") rows.append((n, [ids.count(s) for s in SITE_ONLY], len(ids))) print(f" roster of {n}: {len(ids)} entities, " + ", ".join(f"{s}x{ids.count(s)}" for s in SITE_ONLY)) check("⭐ AC 3: `grid` and `p1_sample_age_s` are announced EXACTLY ONCE at every " "roster length from 1 to 6 - announcing them per inverter would invent " "two of a thing there is one of, and would duplicate a safety signal", all(counts == [1] * len(SITE_ONLY) for _n, counts, _t in rows)) check("...and the site set does not grow with the roster while the per-unit " "set does: 11 site entities always, 4 more per unit past the first", [t for _n, _c, t in rows] == [11, 19, 23, 27, 31, 35]) site_devices = {json.dumps(c_["device"], sort_keys=True) for c_ in cfgs(announce(units=NAMES2)) if c_["unique_id"] in site_ids} check("...and every site entity is still on the SITE device, at a roster of two", site_devices == {json.dumps(mqtt.DEVICE, sort_keys=True)}) # --------------------------------------------------------------------------- print("\n6. the composed entity_ids") # --------------------------------------------------------------------------- # ⚠️ THIS SLUGIFIER IS DELIBERATELY NOT `mqtt._slug`. Importing the module's own # spelling would make the model and the subject move together, so a mutation in # the slugifier would change both sides equally and be invisible - "a check # whose data cannot reach the failure it names", written in one import. def compose(device_name, entity_name): """HA's entity_id for a discovery payload, per mqtt.py's own note.""" s = re.sub(r"[^a-z0-9]+", "_", f"{device_name} {entity_name}".lower()) return "sensor." + s.strip("_") # The ONE anchor there is: mqtt.py records that the live entity is # `sensor.goodwe_p1_sample_age`, verified against the live system 2026-08-26, # and that the firmware subscribes to that composed name. If this model does not # reproduce that id, the model is wrong and every claim below it is worthless. p1 = [c_ for c_ in cfgs(BASELINE) if c_["unique_id"] == "goodwe_ctl_p1_age"][0] check("the model reproduces the ONE composed entity_id that was verified " "against the live system - `sensor.goodwe_p1_sample_age`, which the ESP32 " "ingest guard subscribes to - from the device name and the entity name", compose(mqtt.DEVICE["name"], p1["name"]) == "sensor.goodwe_p1_sample_age") check("...and that id is NOT the `object_id` in the payload " "(`p1_sample_age_s`), which is the whole reason the device name is " "load-bearing and mqtt.py says so in capitals", p1["object_id"] != "goodwe_p1_sample_age") composed = [compose(c_["device"]["name"], c_["name"]) for c_ in cfgs(TWO)] print(" every composed entity_id at a roster of two:") for c_, eid in zip(cfgs(TWO), composed): print(f" {eid:<46} <- device {c_['device']['name']!r} + " f"name {c_['name']!r}") check("⭐ every composed entity_id at a roster of two is DISTINCT: no per-unit " "entity can land on top of a site entity that already has history", len(composed) == len(set(composed)) == 19) base_composed = [compose(c_["device"]["name"], c_["name"]) for c_ in cfgs(BASELINE)] check("...and the eight ids that exist today are unchanged by adding B, which " "is AC 1 restated in the units HA actually uses", composed[:8] == base_composed) check("...and the per-unit ids come out SYMMETRIC even though the devices are " "not - sensor.goodwe_inverter_a_battery_power beside " "sensor.goodwe_inverter_b_battery_power - which is the entire benefit of " "the 'one device per inverter' design, without its cost", "sensor.goodwe_inverter_a_battery_power" in composed and "sensor.goodwe_inverter_b_battery_power" in composed) check("...and for every per-unit entity `object_id` equals the composed id, so " "the two mechanisms agree and an HA that DID honour object_id would " "produce the same ids as one that does not", all("sensor." + c_["object_id"] == compose(c_["device"]["name"], c_["name"]) for c_ in unit_cfgs)) # --------------------------------------------------------------------------- print("\n7. the announcement and the publication agree") # --------------------------------------------------------------------------- # ⚠️ THE HAZARD mqtt.py's `omit` NOTE NAMES: an entity that exists and is never # fed is a worse signal than one that does not exist. So the set of per-unit # topics ANNOUNCED and the set `Controller.publish()` actually writes have to be # the same set - and, because a matching set of KEYS is not a matching set of # VALUES, the two units carry DIFFERENT numbers here so that a swap is red. class _Store: data = {} def set(self, *a): pass def get_time(self, *a): return None class _Pub: def __init__(self): self.last = {} self.units = [] def publish(self, values): self.last = dict(values) def close(self): pass def controlled(opts, soc, batt, tripped): """A real Controller with per-unit readings planted. SIMULATED values.""" # ⚠️ `stop` is a REQUIRED keyword (CFG-01): a Controller that cannot stop # itself used to be the default one. These cases only call publish(), so any # Event will do - but they may not skip it. pub = _Pub() ctl = Controller(dict(opts), None, _Store(), pub, stop=asyncio.Event()) for inv, s, b, t in zip(ctl.inverters, soc, batt, tripped): inv.soc, inv.batt, inv.target = s, b, -1.0 * b inv.sign_wd.tripped = t ctl.publish() return ctl, pub.last # A 41 / 88 %, A discharging 1200 W, B charging 300 W, and B latched, not A. # Every one of those six numbers differs from its sibling, which is what makes # the assertions below able to see a swap. ctl2, vals2 = controlled(OPTS2, [41.0, 88.0], [1200.0, -300.0], [False, True]) announced_unit_topics = {c_["state_topic"] for c_ in unit_cfgs} published_unit_topics = {f"{mqtt.BASE}/{k}" for k in vals2 if k.startswith("unit/")} print(f" announced per-unit topics {len(announced_unit_topics)}, " f"published {len(published_unit_topics)}, " f"announced-but-never-fed {sorted(announced_unit_topics - published_unit_topics)}, " f"fed-but-never-announced {sorted(published_unit_topics - announced_unit_topics)}") check("⭐ every per-unit entity announced is one the controller actually " "publishes to, and every per-unit topic published is one that was " "announced - no entity that nothing feeds, no topic nothing declared", announced_unit_topics == published_unit_topics) check("⭐ AND THE VALUES LAND ON THE RIGHT UNIT: A's 41 % on A's topic and B's " "88 % on B's, not the other way round", vals2["unit/inverter_a/soc"] == 41.0 and vals2["unit/inverter_b/soc"] == 88.0) check("...and the same for battery power, whose SIGNS differ between the two " "units here (A discharging, B charging) so a swap cannot hide in a " "magnitude", vals2["unit/inverter_a/battery"] == 1200.0 and vals2["unit/inverter_b/battery"] == -300.0) check("...and the SAFETY-02 sign latch is per unit: B is latched and A is not, " "and the two rows say so separately", vals2["unit/inverter_a/sign_latch"] == "clear" and vals2["unit/inverter_b/sign_latch"] == "tripped") check("...and each unit's setpoint row is its OWN share, not the site target", vals2["unit/inverter_a/setpoint"] == -1200.0 and vals2["unit/inverter_b/setpoint"] == 300.0) check("...and the SITE rows still mean what they meant: `soc` is the EMPTIEST " "pack (41 %, not 88 and not the mean) and `battery` is the SUM (900 W), " "so this card added rows beside them and did not re-point them", vals2["soc"] == 41.0 and round(vals2["battery"]) == 900) _, vals1 = controlled({"meter_source": "off", "setpoint_entity": "number.sp"}, [55.0], [700.0], [False]) print(f" roster of one publishes keys: {sorted(vals1)}") check("⭐ AC 4 on the PUBLISH side: at a roster of one not one `unit/...` topic " "is written either, so the live install's broker sees no new topic at all", not [k for k in vals1 if k.startswith("unit/")]) # --- 7b. the wiring, which nothing above reaches ----------------------------- # ⚠️ FOUND BY MUTATION, NOT BY DESIGN. `units=()` in main()'s `MqttPublisher(...)` # call survived every check in this file: parts 1-7 build the publisher # themselves, so a controller publishing eight per-unit topics beside a # publisher announcing none of them was GREEN. That is this repository's # recurring defect - a check whose data cannot reach the failure it names - in # the one line that joins the two halves together. check("`roster_names()` and `build_roster()` give the same names in the same " "order, on every shape - the flat keys, a declared roster, a nameless " "entry and a non-dict entry - so the names the publisher is handed are " "the names the units answer to", all([inv.name for inv in build_roster(o, None)] == roster_names(o) for o in ({}, {"setpoint_entity": "number.sp"}, OPTS2, {"inverters": [{}, {}]}, {"inverters": [{"name": "roof west"}, "junk"]}))) check("...and a nameless entry gets a POSITIONAL name, not `inverter-a` twice: " "two units under one name share one Store namespace, so one unit's " "SAFETY-02 latch would be the other's", roster_names({"inverters": [{}, {}]}) == ["inverter-1", "inverter-2"]) _call = [n for n in ast.walk(ast.parse((HERE / "app" / "main.py").read_text())) if isinstance(n, ast.Call) and isinstance(n.func, ast.Name) and n.func.id == "MqttPublisher"] _kw = [k for c_ in _call for k in c_.keywords if k.arg == "units"] print(f" main.py builds MqttPublisher {len(_call)} time(s); " f"units= is " + (ast.unparse(_kw[0].value) if _kw else "ABSENT")) check("⚠️ A SOURCE READ, AND THE WEAKEST CHECK IN THIS FILE - said so rather " "than dressed up. main()'s one `MqttPublisher(...)` hands it " "`roster_names(opts)`; nothing else here can reach that line, because " "main() opens a broker, a web server and a control loop", len(_call) == 1 and len(_kw) == 1 and ast.unparse(_kw[0].value) == "roster_names(opts)") # --- 7c. a name that is not ASCII ------------------------------------------- # HA transliterates before it slugs; this module cannot, so it treats anything # outside ASCII as a separator. Either way the id must be an id. _odd = announce(units=["caf\u00e9-a", "caf\u00e9-b"]) _odd_unit = [c_ for c_ in cfgs(_odd) if c_["unique_id"] not in site_ids] print(" a roster named with non-ASCII ->", sorted({c_["unique_id"] for c_ in _odd_unit})) check("every unique_id, object_id and state topic this module produces is " "lowercase ASCII, even from a roster name that is not - an id HA would " "not have composed is an entity nothing can subscribe to", _odd_unit and all(re.fullmatch(r"[a-z0-9_]+", c_[k]) for c_ in _odd_unit for k in ("unique_id", "object_id")) and all(re.fullmatch(r"[a-z0-9_/]+", c_["state_topic"]) for c_ in _odd_unit)) # The topics the SHIPPED controller actually feeds at a roster of two. Part 8 # grades every mutant announcement against this, which is what turns "the # announcement changed" into "the announcement now claims an entity nothing # will ever publish to". FED = set(published_unit_topics) # --------------------------------------------------------------------------- print("\n8. verified by breaking it") # --------------------------------------------------------------------------- # ⚠️ MUTATED SOURCE, EXEC'D INTO A FRESH NAMESPACE. No file is edited and no # `.pyc` is consulted, so TEST-15's stale-bytecode failure cannot produce a # false survivor here. `app/mqtt.py` imports nothing from this package AT MODULE # SCOPE, which is what makes exec'ing it standalone honest rather than a stub. # ⚠️ TEL-12 IS WHY THAT NOW SAYS "AT MODULE SCOPE". `optional_keys()` reads # `pv.entities()` - one spelling of the roster predicate for the announcement and # for the SUM alike - and imports it INSIDE the function precisely so this # harness keeps working; its own comment says so. Nothing graded below calls # `optional_keys()`, so no mutant here is stubbed by that. def load(pairs): """`app/mqtt.py` with `pairs` applied, executed into a fresh module. ⚠️ EVERY ANCHOR MUST OCCUR EXACTLY ONCE. A mutation whose anchor has drifted out of the file would otherwise apply to nothing and be reported as a survivor - a false red on the instrument rather than on the module. """ src = MQTT_SRC for old, new in pairs: if src.count(old) != 1: raise AssertionError(f"mutation anchor not unique: {old!r}") src = src.replace(old, new) mod = type(sys)("mqtt_mutant") exec(compile(src, "", "exec"), mod.__dict__) return mod def properties(mod): """Every claim this file makes, as booleans, over ONE module.""" out = {} try: base = announce(units=["inverter-a"], module=mod) two = announce(units=["inverter-a", "inverter-b"], module=mod) none = announce(units=(), module=mod) col = announce(units=["inverter-b", "Inverter B"], module=mod) tc = cfgs(two) uc = [c_ for c_ in tc if c_["unique_id"] not in site_ids] bc = [c_ for c_ in uc if "inverter_b" in c_["unique_id"]] ac = [c_ for c_ in uc if "inverter_a" in c_["unique_id"]] comp = [compose(c_["device"]["name"], c_["name"]) for c_ in tc] out["A byte-identical at N=1"] = base == SITE_NOW out["A byte-identical at N=2"] = diff(two)[:2] == ([], []) out["N=1 adds no topic"] = len(base) == len(none) == len(SITE_NOW) out["unique_ids all distinct"] = ( len(set(c_["unique_id"] for c_ in tc)) == len(tc)) out["slug-alike names still distinct"] = ( len(set(c_["unique_id"] for c_ in cfgs(col))) == len(cfgs(col))) out["no composed entity_id collision"] = len(set(comp)) == len(comp) out["object_id == composed id"] = all( "sensor." + c_["object_id"] == compose(c_["device"]["name"], c_["name"]) for c_ in uc) out["A keeps the device it has"] = bool(ac) and all( c_["device"] == mqtt.DEVICE for c_ in ac) out["B has its own device"] = bool(bc) and all( c_["device"]["identifiers"] != mqtt.DEVICE["identifiers"] for c_ in bc) out["B is via_device the site"] = bool(bc) and all( c_["device"].get("via_device") == mqtt.DEVICE["identifiers"][0] for c_ in bc) out["site sensors announced once"] = all( [c_["unique_id"] for c_ in cfgs(announce(units=[f"i{j}" for j in range(n)], module=mod))].count(s) == 1 for n in (1, 2, 3) for s in SITE_ONLY) # ⚠️ TWO PROPERTIES THAT WERE MISSING, AND THE INSTRUMENT FOUND THEIR # ABSENCE ITSELF: on the first run of this table, "the unit is dropped # from the state topic" and "`grid` is moved to the per-unit set" were # both GREEN. The first collapses A's and B's four topics onto one set # of four, so both inverters read one number; the second announces a # per-inverter grid sensor, which is the exact thing AC 3 forbids. The # ids stayed distinct through both, so every property above missed them. out["per-unit state topics all distinct"] = ( len({c_["state_topic"] for c_ in uc}) == len(uc)) # ⚠️ GRADED AGAINST THE SHIPPED `Controller.publish()`, on purpose. An # announcement is only honest if something feeds it - mqtt.py's `omit` # note - so a per-unit entity the controller will never publish to is a # defect however tidy its id looks. (A rename applied to BOTH halves at # once is invisible here and is disclosed in part 9; a rename applied to # one half is what actually happens, and is red.) out["every announced per-unit topic is fed"] = ( {c_["state_topic"] for c_ in uc} == FED) except Exception as err: # a mutation that CRASHES proves nothing; say so out["did not raise"] = False out["__error__"] = repr(err)[:70] return out SHIPPED = properties(mqtt) check("the shipped module satisfies every property this instrument grades - " "so a red below is the mutation and not the instrument", all(v is True for k, v in SHIPPED.items())) MUTATIONS = [ ("⭐ RE-SCOPE A's DEVICE NAME (the destructive change, AC 5)", [('"name": "GoodWe",', '"name": "GoodWe Inverter A",')]), ("⭐ RE-SCOPE A's DEVICE IDENTIFIERS", [('"identifiers": ["goodwe_rs485_controller"],', '"identifiers": ["goodwe_inverter_a"],')]), ("⭐ THE TEMPTING TIDY DESIGN: one device per inverter, A included", [("device = DEVICE if i == 0 else unit_device(name, slug)", "device = unit_device(name, slug)"), ('prefix = f"{_title(name)} " if i == 0 else ""', 'prefix = ""')]), ("B's rows land on A's device instead of B's own", [("device = DEVICE if i == 0 else unit_device(name, slug)", "device = DEVICE")]), ("A's per-unit entity names lose the unit prefix", [('prefix = f"{_title(name)} " if i == 0 else ""', 'prefix = ""')]), ("per-unit entities announced at a roster of ONE", [("PER_UNIT_MIN_ROSTER = 2", "PER_UNIT_MIN_ROSTER = 1")]), ("the unit is dropped from the unique_id", [('return f"{BASE}_unit_{slug}_{key}"', 'return f"{BASE}_unit_{key}"')]), ("the unit is dropped from the state topic", [('return f"unit/{slug}/{key}"', 'return f"unit/{key}"')]), ("the slug de-collision is removed (rule 9(c) does not cover this)", [("while slug in seen:", "while False and slug in seen:")]), ("every unit takes the FIRST unit's slug", [("for i, (name, slug) in enumerate(zip(names, slugs)):", "for i, (name, slug) in enumerate(zip(names, [slugs[0]] * len(slugs))):")]), ("`grid` is moved to the per-unit set (site sensor announced per inverter)", [(' ("setpoint", "setpoint", "Setpoint", "W", "power", "measurement", None),\n', ' ("setpoint", "setpoint", "Setpoint", "W", "power", "measurement", None),\n' ' ("grid", "grid_power", "Grid power", "W", "power", "measurement", None),\n')]), ("B's device drops `via_device`", [(' "via_device": DEVICE["identifiers"][0],\n', "")]), # --- TEL-10 ------------------------------------------------------------ ("TEL-10's integrator row is DROPPED (published to a topic nothing " "announced, which AC 4 forbids in the other direction)", [(' ("loop_i", "goodwe_loop_integrator", "Loop integrator", "W", "power", "measurement", None),\n', "")]), ("TEL-10's integrator is announced PER INVERTER - two answers to a " "question §6.2 says there is one integrator for", [(' ("sign_latch", "sign_latch", "Sign latch", None, None, None, "mdi:flash-alert"),\n', ' ("sign_latch", "sign_latch", "Sign latch", None, None, None, "mdi:flash-alert"),\n' ' ("loop_i", "loop_integrator", "Loop integrator", "W", "power", "measurement", None),\n')]), # ⚠️ THE ONE THAT CHANGES NO BYTE OF ANY PAYLOAD. Every id, name, unit and # topic is identical; only the POSITION of the integrator row moves. A # broker replaying the retained set then hands HA a different sequence, and # part 1's claim is about order as much as about bytes - which is why it is # asserted as a LIST and not as a set. ("TEL-10's integrator row is announced BEFORE release/1.0's nine", [(' ("loop_i", "goodwe_loop_integrator", "Loop integrator", "W", "power", "measurement", None),\n', ""), (' ("setpoint", "goodwe_setpoint", "Setpoint", "W", "power", "measurement", None),\n', ' ("loop_i", "goodwe_loop_integrator", "Loop integrator", "W", "power", "measurement", None),\n' ' ("setpoint", "goodwe_setpoint", "Setpoint", "W", "power", "measurement", None),\n')]), ] survivors = [] for label, pairs in MUTATIONS: got = properties(load(pairs)) broke = sorted(k for k, v in SHIPPED.items() if v is True and got.get(k) is not True) if not broke: survivors.append(label) print(f" {'RED ' if broke else 'GREEN'} {label}") if broke: print(f" breaks: {', '.join(broke)}") elif got.get("__error__"): print(f" raised: {got['__error__']}") check(f"⭐ ALL {len(MUTATIONS)} MUTATIONS ARE RED, each on a NAMED property: " "re-scoping A's device, the tidy one-device-per-inverter design, B " "landing on A's device, per-unit entities at a roster of one, a unit " "dropped from an id or a topic, two units sharing a slug, and a site " "sensor announced per inverter. A mutation nothing catches is a claim " "nothing is watching", not survivors) if survivors: print(" SURVIVED:", "; ".join(survivors)) # ⚠️ AC 6, THE FIXTURE'S OWN NON-VACUITY. If A and B were given the same name # the whole of part 4 would still be green and would be observing nothing. same = announce(units=["inverter-a", "inverter-a"]) swapped = announce(units=["inverter-b", "inverter-a"]) check("...and the fixture's two units are genuinely distinguishable: swapping " "their order changes the announcement, and a roster of two IDENTICAL " "names - which rule 9(c) refuses, and which this file therefore never " "relies on - would still not collide", swapped != TWO and len({c_["unique_id"] for c_ in cfgs(same)}) == len(cfgs(same))) # --------------------------------------------------------------------------- print("\n9. what this file CANNOT see, enumerated and sized") # --------------------------------------------------------------------------- for line in [ "1. HA ITSELF IS NOT IN THE LOOP. `compose()` in part 6 is a MODEL of how " "HA 2026.8 builds an entity_id, anchored on exactly ONE measured id " "(sensor.goodwe_p1_sample_age, mqtt.py's note, verified 2026-08-26). If HA " "changes that rule, or if it differs for a device with `via_device`, every " "id claim here moves with it and this file stays green. SIZE: the 8 " "per-unit ids at N=2. It cannot be closed from this repository - it needs a " "broker, an HA, and the owner present.", "2. THE TICKET AND THE MODULE DISAGREE ABOUT THE LIVE IDs. The card quotes " "`sensor.goodwe_rs485_controller_goodwe_battery_soc`; mqtt.py records " "`sensor.goodwe_p1_sample_age`. Those cannot both come from one composition " "rule - the first is the id an EARLIER device name composed, and HA keeps " "whatever it composed at first discovery. This card does not need to " "resolve it, because it renames no device and no entity, and the byte-diff " "protects both spellings equally. SIZE: unresolved, 8 live entity ids; it " "wants one look at the live registry, which this branch must not take.", "3. THE PER-UNIT PAYLOADS HAVE NO RELEASED BASELINE, AND NEITHER DO " "TEL-10's THREE. A's are diffed against release/1.0; B's can only be " "diffed against this branch (the sha256 pin in part 3), and `TEL10` above " "is the same kind of pin for the three site rows this branch adds. A wrong " "`device_class` or unit on any of them is therefore caught as a CHANGE " "from review, not as an error. SIZE: 4 per-unit sensors x 4 optional " "fields, plus 3 site sensors x 4.", "4. NO CHECK HERE RUNS A BROKER. `_announce()` is graded through a fake " "paho client, so QoS, retain semantics on a real broker, and what HA does " "with a retained config for a device it has never seen are all out of " "scope. SIZE: the whole transport.", "5. THE SLUG DE-COLLISION IS ORDER-DEPENDENT. `inverter-b` before " "`Inverter B` gives the SECOND one the `_2` suffix; reordering the roster " "swaps which unit's ids move. Rule 9(c) does not refuse slug-alike names, " "so this is reachable from a config an operator can write. SIZE: 4 ids on " "one unit, and only when two names slug alike. Filed as a follow-up rather " "than fixed in validate.py here, because a new refusal at startup on a " "LIVE install is a bigger change than this card is allowed to be.", "7. PART 8 MUTATES `app/mqtt.py` ONLY. `Controller.publish()`'s four " "per-unit lines are graded by part 7 - with different SoC, different " "battery sign and different latch states on the two units, so a swap or a " "mis-paired slug is red - but no automated table mutates main.py here. " "Those mutations were run by hand through `tools/mutation_run.sh` and are " "recorded in the PR. SIZE: 4 lines. Also invisible: a rename of " "`unit_state_key()` applied to BOTH halves at once, which is harmless " "(the topics are internal) but would move every per-unit state topic " "silently.", "6. `inverter AC power` AND PER-UNIT HEARTBEAT AGE ARE NOT ANNOUNCED. The " "card lists both as per-unit; neither has a source in the add-on today " "(nothing reads an AC-power register per unit, and `inv.last_beat` is a " "monotonic loop clock). Announcing them would create entities nothing " "feeds. SIZE: 2 sensors x N units, deliberately absent, named in the PR.", ]: print(" " + line) # --------------------------------------------------------------------------- print("\n10. OPT-11 P2: each battery's switch and mode as HA entities") # --------------------------------------------------------------------------- import logging # noqa: E402 import os # noqa: E402 import tempfile # noqa: E402 from app import messages, packmode # noqa: E402 from app.store import Store, unit_key # noqa: E402 def announce_p2(units=(), omit=(), include=(), modes=(), retract=(), module=mqtt): """What `_on_connect()` sends: the announcement, then any subscriptions.""" pub = module.MqttPublisher.__new__(module.MqttPublisher) pub.omit, pub.units, pub.include = set(omit), [str(u) for u in units], \ [str(k) for k in include] pub.modes, pub.retract = tuple(modes), list(retract) pub.enabled, pub.client = True, FakeClient() subs = [] pub.client.subscribe = lambda topic, *a: subs.append(topic) pub._on_connect() return pub.client.msgs, subs # --- 10a. modes off: byte-identical to ed739d4 -------------------------------- # The digest below is GENERATED from release/1.0 at ed739d4 (the base of this # PR): the same matrix run through that commit's `_announce()`. Reproduce with # `git show ed739d4:goodwe_controller/app/mqtt.py` and this loop. ROSTERS = [[], ["inverter-a"], ["inverter-a", "inverter-b"], ["a", "b", "c"], ["inverter-b", "Inverter B", "x", "y"], ["café"]] ED739D4_OFF = "501a7d93782a260b" _h, _n, _subs_off = hashlib.sha256(), 0, [] for _r in ROSTERS: for _om in [(), ("p1_age",), ("p1_age", "grid")]: for _inc in [(), ("pv",)]: _m, _s = announce_p2(_r, _om, _inc) _h.update(repr(_m).encode()) _subs_off += _s _n += 1 print(f" modes off, {_n} configurations: digest {_h.hexdigest()[:16]} " f"(ed739d4: {ED739D4_OFF})") check("modes off: every configuration announces byte-for-byte what ed739d4 " "announces, in the same order, and subscribes to nothing", _h.hexdigest()[:16] == ED739D4_OFF and not _subs_off) # --- 10b. modes on: the ordered payload diff ---------------------------------- MODES = packmode.MODES ON1, SUB1 = announce_p2(["inverter-a"], modes=MODES) m, c, a = report("roster of one, modes on", ON1) # P2 announced the first two; OPT-11 P3 appended "Cheap hours only". _opt = '["Solar first", "Leftover solar", "Cheap hours only"]' MODE_N1 = [ ("homeassistant/select/goodwe_ctl_unit_inverter_a_mode/config", '{"name": "Inverter A Mode", "object_id": "goodwe_inverter_a_mode", ' '"unique_id": "goodwe_ctl_unit_inverter_a_mode", ' '"state_topic": "goodwe_ctl/unit/inverter_a/mode", ' '"command_topic": "goodwe_ctl/unit/inverter_a/mode/set", ' '"options": ' + _opt + ', "retain": false, ' '"availability_topic": "goodwe_ctl/availability", "device": ' + _DEV + ', ' '"icon": "mdi:battery-sync"}', True), ("homeassistant/switch/goodwe_ctl_unit_inverter_a_use/config", '{"name": "Inverter A Use", "object_id": "goodwe_inverter_a_use", ' '"unique_id": "goodwe_ctl_unit_inverter_a_use", ' '"state_topic": "goodwe_ctl/unit/inverter_a/use", ' '"command_topic": "goodwe_ctl/unit/inverter_a/use/set", ' '"payload_on": "ON", "payload_off": "OFF", "retain": false, ' '"availability_topic": "goodwe_ctl/availability", "device": ' + _DEV + ', ' '"icon": "mdi:power"}', True), ] check("modes on at a roster of one: every existing message byte-identical and " "in place, then exactly the select and the switch, then availability", ON1 == SITE_NOW[:-1] + MODE_N1 + SITE_NOW[-1:]) check("...and it subscribes to the two command topics, and nothing else", SUB1 == ["goodwe_ctl/unit/+/mode/set", "goodwe_ctl/unit/+/use/set"]) ON2, _ = announce_p2(NAMES2, modes=MODES) m, c, a = report("roster of two, modes on", ON2, TWO) check("modes on at a roster of two: nothing missing or changed, four added, " "and the modes-off list is an unchanged prefix", (m, c) == ([], []) and len(a) == 4 and ON2[:len(TWO) - 1] == TWO[:-1] and ON2[-1] == TWO[-1]) _mc = cfgs(ON2)[-4:] check("...A's two stay on the site device with its name in front; B's two are " "on B's own device, as TEL-07's rule has it", [c_["name"] for c_ in _mc] == ["Inverter A Mode", "Inverter A Use", "Mode", "Use"] and _mc[0]["device"] == mqtt.DEVICE == _mc[1]["device"] and _mc[2]["device"] == _mc[3]["device"] == mqtt.unit_device("inverter-b", "inverter_b")) check("...unique_ids are stable per battery and distinct from every other id", [c_["unique_id"] for c_ in _mc] == ["goodwe_ctl_unit_inverter_a_mode", "goodwe_ctl_unit_inverter_a_use", "goodwe_ctl_unit_inverter_b_mode", "goodwe_ctl_unit_inverter_b_use"] and len({c_["unique_id"] for c_ in cfgs(ON2)}) == len(cfgs(ON2))) _ids = [compose(c_["device"]["name"], c_["name"]).split(".", 1)[1] for c_ in _mc] print(f" composed ids: {_ids}") check("...and HA composes the ids the design names: " "goodwe_inverter__mode and goodwe_inverter__use", _ids == ["goodwe_inverter_a_mode", "goodwe_inverter_a_use", "goodwe_inverter_b_mode", "goodwe_inverter_b_use"]) check("every mode has a plain label, in MODES order, and a later mode only " "appends an option: no existing option or id changes", all(k in mqtt.MODE_LABELS for k in MODES) and len(set(mqtt.MODE_LABELS.values())) == len(mqtt.MODE_LABELS) and json.loads(MODE_N1[0][1])["options"] == [mqtt.MODE_LABELS[k] for k in MODES]) # Review of #152: the HA options are what automations send, so they are kept # apart from the page's words on purpose. They start equal; if the page's # English changes, this goes red and the author decides: keep the HA option # (and change this pin) or rename it (a breaking change, in DECISIONS). _en = json.loads((HERE / "app" / "ui_text" / "en.json").read_text()) check("each HA mode option is the page's English name for that mode, in " "both places the page names it", all(mqtt.MODE_LABELS[k] == _en["home_words"]["modes"][k] == _en["home"]["batteries"]["choice"][k]["name"] for k in MODES)) # OPT-11 P3 added `cheap_hours`: against P2's two modes (7a00f12), only the # options list changes, by one entry at its end; no topic, id or label moves. _p3 = mqtt.mode_payloads(NAMES2, MODES) _orig = mqtt.MODE_LABELS mqtt.MODE_LABELS = {k: v for k, v in _orig.items() if k != "cheap_hours"} try: _p2 = mqtt.mode_payloads(NAMES2, MODES[:2]) finally: mqtt.MODE_LABELS = _orig check("...P3 appended `cheap_hours`: against P2's list only the options " "change, by one entry at the end, and P2's two labels are unchanged", MODES[:2] == ("solar_first", "leftover_solar") and _p2[0][1]["options"] == ["Solar first", "Leftover solar"] and [t for t, _ in _p3] == [t for t, _ in _p2] and all({k: v for k, v in x.items() if k != "options"} == {k: v for k, v in y.items() if k != "options"} for (_, x), (_, y) in zip(_p3, _p2)) and _p3[0][1]["options"] == _p2[0][1]["options"] + ["Cheap hours only"]) # --- 10c. modes turned off: removal ------------------------------------------ _cur2 = [t for t, _ in mqtt.mode_payloads(NAMES2, MODES)] _ret = mqtt.mode_retractions(_cur2, []) OFF2, _subs = announce_p2(NAMES2, retract=_ret) check("modes turned off: the modes-off list, then an EMPTY retained config " "for each of the four, then availability, and no subscription", OFF2 == TWO[:-1] + [(t, "", True) for t in _cur2] + TWO[-1:] and not _subs) check("...the store keeps what was announced, so a roster that shrinks removes " "only the battery that left", mqtt.mode_retractions(_cur2, [t for t, _ in mqtt.mode_payloads(["inverter-a"], MODES)]) == _cur2[2:]) check("...and nothing from the store is removed that is not one of our mode " "configs: a non-list, a number, a site sensor, another add-on's select", mqtt.mode_retractions("homeassistant/select/x/config", []) == [] and mqtt.mode_retractions( [3, None, "homeassistant/sensor/goodwe_ctl_grid/config", "homeassistant/select/other_mode/config", "homeassistant/select/goodwe_ctl_unit_a_mode/config/x", "homeassistant/select/goodwe_ctl_unit_a_mode/config"], []) == ["homeassistant/select/goodwe_ctl_unit_a_mode/config"]) _on1 = [t for t, _ in mqtt.mode_payloads(["inverter-a"], MODES)] _k, _r = mqtt.mode_configs(["inverter-a"], MODES, _cur2) check("start-up (`mode_configs`): a battery that left is removed, and stays " "in the store, so a start with no broker does not forget it", _r == _cur2[2:] and _k == _on1 + _cur2[2:]) _k, _r = mqtt.mode_configs(NAMES2, (), _cur2 + _cur2[:1]) check("...modes turned off: all four are removed, once each, and kept", _r == _cur2 and _k == _cur2) check("...never turned on: nothing to store and nothing to remove, so the " "store is not written", mqtt.mode_configs(NAMES2, (), None) == ([], [])) _k, _r = mqtt.mode_configs(NAMES2, MODES, _cur2) check("...turned on again: nothing removed, and the list does not grow", _r == [] and _k == _cur2) from app.main import mode_announcement # noqa: E402 _st = Store(os.path.join(tempfile.mkdtemp(), "ann.json")) _OPTS1M = dict(OPTS2, battery_modes_enabled=True, inverters=OPTS2["inverters"][:1]) _seen = [mode_announcement(dict(OPTS2), _st), dict(_st.data)] _seen += [mode_announcement(dict(OPTS2, battery_modes_enabled=True), _st), list(_st.data.get(mqtt.MODE_CONFIGS_KEY) or [])] _seen += [mode_announcement(_OPTS1M, _st), list(_st.data.get(mqtt.MODE_CONFIGS_KEY) or [])] _seen += [mode_announcement(dict(OPTS2, battery_modes_enabled=False), _st), list(_st.data.get(mqtt.MODE_CONFIGS_KEY) or [])] check("main's start-up, through a real Store: never on announces and stores " "nothing; on stores the four; a roster of one removes B's two and keeps " "them; off removes all four and the store keeps them", _seen[0] == ((), []) and mqtt.MODE_CONFIGS_KEY not in _seen[1] and _seen[2] == (MODES, []) and _seen[3] == _cur2 and _seen[4] == (MODES, _cur2[2:]) and sorted(_seen[5]) == sorted(_cur2) and _seen[6] == ((), _seen[5]) and sorted(_seen[7]) == sorted(_cur2)) _st = Store(os.path.join(tempfile.mkdtemp(), "ren.json")) mode_announcement(dict(OPTS2, battery_modes_enabled=True), _st) _ren = dict(OPTS2, battery_modes_enabled=True, inverters=[ OPTS2["inverters"][0], dict(OPTS2["inverters"][1], name="inverter-c")]) _r = mode_announcement(_ren, _st)[1] _kept = list(_st.data.get(mqtt.MODE_CONFIGS_KEY) or []) check("...B renamed C while on: B's two are removed and stay stored beside " "A's and C's, so a start with no broker does not forget them", _r == _cur2[2:] and _kept[-2:] == _cur2[2:] and len(_kept) == 6 and mode_announcement(_ren, _st)[1] == _cur2[2:]) check("battery modes are on only for exactly `true`: not \"true\", 1 or " "\"false\"", [packmode.enabled({"battery_modes_enabled": v}) for v in (True, "true", 1, "false", None)] == [True, False, False, False, False]) # --- 10d. state topics, from the store ---------------------------------------- _tmp = tempfile.mkdtemp() OPTS2M = dict(OPTS2, battery_modes_enabled=True) def ctl_for(opts, path): pub = _Pub() pub.sent = [] pub.publish = lambda values, _p=pub: (_p.sent.append(dict(values)), setattr(_p, "last", dict(values))) return Controller(dict(opts), None, Store(path), pub, stop=asyncio.Event()), pub _path = os.path.join(_tmp, "state.json") ctlm, pubm = ctl_for(OPTS2M, _path) ctlm.publish() _fed = {f"{mqtt.BASE}/{k}" for k in pubm.last if k.endswith("/mode") or k.endswith("/use")} check("every announced mode entity is fed, and nothing else is: the state " "topics published are exactly the ones announced", _fed == {c_["state_topic"] for c_ in _mc}) check("...defaults with nothing in the store: both On, both Solar first", [pubm.last[k] for k in ("unit/inverter_a/mode", "unit/inverter_a/use", "unit/inverter_b/mode", "unit/inverter_b/use")] == ["Solar first", "ON", "Solar first", "ON"]) _c_off, _p_off = ctl_for(dict(OPTS2, battery_modes_enabled=False), os.path.join(_tmp, "off2.json")) _c_off.publish() _c_abs, _p_abs = ctl_for(OPTS2, os.path.join(_tmp, "off3.json")) _c_abs.publish() check("modes off (false or absent): the controller publishes the same keys as " "before, with no mode or switch topic", set(_p_off.last) == set(_p_abs.last) == set(vals2) and not [k for k in _p_off.last if k.endswith(("/mode", "/use"))]) # --- 10e. commands: the /action path, validated ------------------------------- class _Logs(logging.Handler): def __init__(self): super().__init__(logging.DEBUG) self.rows = [] def emit(self, record): self.rows.append((record.levelno, record.getMessage())) _logs = _Logs() logging.getLogger("goodwe").addHandler(_logs) logging.getLogger("goodwe").setLevel(logging.DEBUG) _calls = [] _real_handle = ctlm.handle_action async def _spy(what, **kw): _calls.append((what, kw)) return await _real_handle(what, **kw) ctlm.handle_action = _spy def cmd(topic, payload, retained=False, ctl=None): ctl = ctl or ctlm asyncio.run(ctl.mqtt_command(topic, payload, retained)) def stored(ctl=None): d = (ctl or ctlm).store.data return {k: d.get(k) for k in (unit_key("inverter-a", "pack_use"), unit_key("inverter-a", "pack_mode"), unit_key("inverter-b", "pack_use"), unit_key("inverter-b", "pack_mode"))} T_A = "goodwe_ctl/unit/inverter_a" T_B = "goodwe_ctl/unit/inverter_b" pubm.sent.clear() cmd(f"{T_B}/use/set", b"OFF") cmd(f"{T_A}/mode/set", b"Leftover solar") check("good commands go through handle_action() with the /action arguments", _calls == [("pack_use", {"unit": "inverter-b", "value": False}), ("pack_mode", {"unit": "inverter-a", "value": "leftover_solar"})]) check("...and reach the store under P1's keys, and nowhere else", stored() == {unit_key("inverter-a", "pack_use"): None, unit_key("inverter-a", "pack_mode"): "leftover_solar", unit_key("inverter-b", "pack_use"): False, unit_key("inverter-b", "pack_mode"): None}) check("...and the state topics say so at once, from the store", pubm.sent and pubm.sent[-1] == { "unit/inverter_a/mode": "Leftover solar", "unit/inverter_a/use": "ON", "unit/inverter_b/mode": "Solar first", "unit/inverter_b/use": "OFF"}) check("...and the action log records each change", any("inverter-b switched off" in e for e in ctlm.events) and any("inverter-a mode set to leftover solar" in e for e in ctlm.events)) _before = stored() _n_calls = len(_calls) BAD = [ (f"{T_A}/use/set", b"on"), (f"{T_A}/use/set", b"ON\n"), (f"{T_A}/use/set", b"true"), (f"{T_A}/use/set", b"1"), (f"{T_A}/use/set", b""), (f"{T_A}/mode/set", b"solar_first"), # a key, not the option (f"{T_A}/mode/set", b"Sell only"), # not an offered mode (f"{T_A}/mode/set", b"solar first"), (f"{T_A}/mode/set", "Sοlar first".encode()), # Greek omicron (f"{T_A}/mode/set", b"Solar first\x00"), (f"{T_A}/mode/set", b'{"action": "pack_use", "value": false}'), (f"{T_A}/mode/set", b"Solar first; rm -rf /"), (f"{T_A}/mode/set", b"' OR 1=1 --"), (f"{T_A}/use/set", b"\xff\xfe"), # not UTF-8 (f"{T_A}/mode/set", b"Solar first" * 10), # too long ("goodwe_ctl/unit/inverter_z/use/set", b"OFF"), # no such battery ("goodwe_ctl/unit/../use/set", b"OFF"), ("goodwe_ctl/unit/inverter_a/soc/set", b"OFF"), ("goodwe_ctl/unit/inverter_a/use/set/x", b"OFF"), ("goodwe_ctl/unit/Inverter_A/use/set", b"OFF"), (f"{T_A}/use/set", None), ] for _t, _p in BAD: cmd(_t, _p) check(f"all {len(BAD)} bad commands are refused: the store is unchanged", stored() == _before) _passed = [c_ for c_ in _calls[_n_calls:]] check("...those that reach handle_action() carry a value /action also refuses " "(None) or no known battery; nothing is coerced into a valid choice", all(kw["value"] is None or kw["unit"] is None for _, kw in _passed)) # Log once per topic and reason. _logs.rows.clear() for _ in range(5): cmd(f"{T_B}/use/set", b"maybe") _warn = [r for r in _logs.rows if r[0] >= logging.WARNING] check("a refused command repeated 5 times is logged once at WARNING, in plain " "words, and at DEBUG the other times", len(_warn) == 1 and messages.pack_refused("value") in _warn[0][1] and len([r for r in _logs.rows if r[0] == logging.DEBUG]) == 5) _logs.rows.clear() _flood, _ = ctl_for(OPTS2M, os.path.join(_tmp, "flood.json")) for _i in range(100): cmd(f"goodwe_ctl/unit/x{_i}/use/set", b"OFF", ctl=_flood) _w = [r[1] for r in _logs.rows if r[0] >= logging.WARNING] _per = [w for w in _w if "still ignored" not in w] _more = [w for w in _w if "still ignored" in w] print(f" flood: {len(_per)} per-topic WARNINGs, then {_more}") check("...and a flood of 100 different bad topics logs 32 WARNINGs, then the " "rest only as a count, at 1 and 10 more (review of #152)", len(_per) == 32 and len(_more) == 2 and "1 more so far" in _more[0] and "10 more so far" in _more[1]) _logs.rows.clear() for _ in range(100): cmd(f"{T_B}/use/set", b"maybe", ctl=_flood) _more = [r[1] for r in _logs.rows if r[0] >= logging.WARNING and "still ignored" in r[1]] check("...so a repeated attack on ONE topic still shows it goes on: the count " "reaches 100 in 3 lines, not 100", len(_more) == 1 and "100 more so far" in _more[0]) _logs.rows.clear() _inj, _ = ctl_for(OPTS2M, os.path.join(_tmp, "inj.json")) cmd("goodwe_ctl/unit/a\nERROR forged line\x1b[31m/use/set", b"OFF", ctl=_inj) _w = [r[1] for r in _logs.rows if r[0] >= logging.WARNING] check("...and a topic with a newline or an escape code cannot forge a log " "line: it is logged printable only", len(_w) == 1 and "\n" not in _w[0] and "\x1b" not in _w[0] and "a?ERROR forged line?[31m" in _w[0]) # Modes off: refused on the same path, even if a message arrives. cmd(f"{T_A}/use/set", b"OFF", ctl=_c_off) check("with modes off a command is refused by the same check /action makes", stored(_c_off)[unit_key("inverter-a", "pack_use")] is None) # --- 10f. a retained command after a restart ---------------------------------- _logs.rows.clear() _before = stored() _n_calls = len(_calls) cmd(f"{T_A}/use/set", b"OFF", retained=True) cmd(f"{T_B}/use/set", b"ON", retained=True) check("a RETAINED command (the broker replays it on subscribe, e.g. after a " "restart) is never applied, and never reaches handle_action()", stored() == _before and len(_calls) == _n_calls) check("...and it is logged once, in plain words", any(messages.pack_refused("retained") in r[1] for r in _logs.rows if r[0] >= logging.WARNING)) ctl_r, pub_r = ctl_for(OPTS2M, _path) # a restart: same state file ctl_r.publish() check("after a restart the state topics come from the store: A Leftover solar " "and on, B off", [pub_r.last[k] for k in ("unit/inverter_a/mode", "unit/inverter_a/use", "unit/inverter_b/use")] == ["Leftover solar", "ON", "OFF"]) logging.getLogger("goodwe").removeHandler(_logs) # --- 10g. the paho side ---------------------------------------------------------- class _Msg: def __init__(self, topic, payload, retain): self.topic, self.payload, self.retain = topic, payload, retain _got = [] async def _handler(*args): _got.append(args) async def _drive(): pub = mqtt.MqttPublisher.__new__(mqtt.MqttPublisher) pub._on_message(None, None, _Msg("t", b"x", 0)) # not bound: dropped pub.bind(asyncio.get_running_loop(), _handler) pub._on_message(None, None, _Msg(f"{T_A}/use/set", b"OFF", 1)) await asyncio.sleep(0.05) asyncio.run(_drive()) check("paho's thread hands (topic, bytes, retained) to the controller's loop, " "and drops a message that arrives before the controller exists", _got == [(f"{T_A}/use/set", b"OFF", True)]) _kws = {k.arg: ast.unparse(k.value) for c_ in _call for k in c_.keywords} _main_src = (HERE / "app" / "main.py").read_text() check("(source read) main() passes modes and retractions to the publisher, " "and binds its commands to the controller", _kws.get("modes") == "_modes" and _kws.get("retract") == "_retract" and "pub.bind(asyncio.get_running_loop(), controller.mqtt_command)" in _main_src and "_modes, _retract = mode_announcement(opts, store)" in _main_src) # --- 10h. the guards, broken one at a time -------------------------------------- def p2_properties(mod): out = {} try: off, s_off = announce_p2(NAMES2, module=mod) on1, s_on = announce_p2(["inverter-a"], modes=MODES, module=mod) on2, _ = announce_p2(NAMES2, modes=MODES, module=mod) out["off identical"] = off == TWO and not s_off out["on payloads"] = on1 == ON1 and on2 == ON2 out["subscribes when on"] = s_on == SUB1 cur = [t for t, _ in mod.mode_payloads(NAMES2, MODES)] out["retract filters store"] = mod.mode_retractions( [3, "homeassistant/sensor/goodwe_ctl_grid/config", "homeassistant/select/other/config"] + cur, cur) == [] out["retract keeps current"] = mod.mode_retractions(cur, cur) == [] out["retract published"] = announce_p2( NAMES2, retract=cur[:1], module=mod)[0] == \ TWO[:-1] + [(cur[0], "", True)] + TWO[-1:] pc = mod.parse_command out["retained dropped"] = pc(f"{T_A}/use/set", b"OFF", True, MODES)[3] == "retained" out["size capped"] = pc(f"{T_A}/mode/set", b"Solar first" * 10, False, MODES)[3] == "size" out["utf-8 only"] = pc(f"{T_A}/use/set", b"\xff", False, MODES)[3] == "encoding" out["switch exact"] = ( pc(f"{T_A}/use/set", b"on", False, MODES)[2] is None and pc(f"{T_A}/use/set", b"OFF", False, MODES)[2] is False) out["select labels only"] = ( pc(f"{T_A}/mode/set", b"solar_first", False, MODES)[2] is None and pc(f"{T_A}/mode/set", b"Leftover solar", False, MODES)[2] == "leftover_solar") out["select offered only"] = pc( f"{T_A}/mode/set", b"Solar first", False, ("leftover_solar",))[2] \ is None out["topic exact"] = all( pc(t, b"OFF", False, MODES)[3] == "topic" for t in ("goodwe_ctl/unit/../use/set", "goodwe_ctl/unit/a/b/use/set", "goodwe_ctl/unit/a/use/set/x", "x/goodwe_ctl/unit/a/use/set")) except Exception as err: # noqa: BLE001 out["did not raise"] = False out["__error__"] = repr(err)[:70] return out P2_SHIPPED = p2_properties(mqtt) check("the shipped module has every P2 property (so a red below is the " "mutation)", all(v is True for v in P2_SHIPPED.values())) P2_MUTATIONS = [ ("mode entities announced with modes off", [(" if not modes:\n return []\n", " modes = modes or ('solar_first', 'leftover_solar')\n")]), ("subscribed with modes off", [(" if self.modes:\n", " if True:\n")]), ("retained commands applied", [(" if retained:\n", " if False:\n")]), ("no payload size cap", [("or len(payload) > MAX_COMMAND_BYTES", "or False")]), ("bad UTF-8 replaced, not refused", [('.decode("utf-8")', '.decode("utf-8", "replace")')]), ("switch payload case-folded", [("{USE_ON: True, USE_OFF: False}.get(text)", "{USE_ON: True, USE_OFF: False}.get(text.upper())")]), ("select accepts mode keys too", [("if MODE_LABELS.get(k) == text", "if text in (k, MODE_LABELS.get(k))")]), ("select accepts any labelled mode, offered or not", [("(k for k in modes if MODE_LABELS.get(k) == text)", "(k for k in MODE_LABELS if MODE_LABELS.get(k) == text)")]), ("command topic loosened", [('/([a-z0-9_]+)/(mode|use)/set")', '/(.+)/(mode|use)/set")')]), ("command topic matched as a prefix", [("_COMMAND_RE.fullmatch(topic)", "_COMMAND_RE.search(topic)")]), ("store list not filtered", [("and _MODE_CONFIG_RE.fullmatch(t)\n", "and True\n")]), ("current configs also removed", [("and t not in keep))", "))")]), ("commands published retained by HA", [('cfg["retain"] = False', 'cfg["retain"] = True')]), ("retractions not published", [(' self.client.publish(topic, "", retain=True)\n', " pass\n")]), ] _p2_surv = [] for label, pairs in P2_MUTATIONS: got = properties(load(pairs)) got.update(p2_properties(load(pairs))) broke = sorted(k for k, v in {**SHIPPED, **P2_SHIPPED}.items() if v is True and got.get(k) is not True) print(f" {'RED ' if broke else 'GREEN'} {label}" + (f" (breaks: {', '.join(broke)})" if broke else "")) if not broke: _p2_surv.append(label) check(f"all {len(P2_MUTATIONS)} P2 mutations of app/mqtt.py are red", not _p2_surv) # --------------------------------------------------------------------------- print() if fails: print(f"{len(fails)} FAILED:") for f in fails: print(" -", f) sys.exit(1) print("all checks passed")