admin/goodwe-addon
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

TEST-06: close the three gaps that survived SAFETY-18's suite green

Browse Source 
None of the three is a bug in the shipped code. They are gaps in what the
suite could detect, and the next person to touch freeze.py had less
protection than the passing count implied.

1. WHERE check_meter_frozen() runs in cycle(). Moving it to the top left
   the whole file green. Pinned by CONSEQUENCE, not by source order, so a
   refactor that keeps the semantics stays green:
     - the alarm must land on the SAME cycle the command reaches the rail
       (at the top it grades the previous command against this cycle's
       meter and arrives a cycle late);
     - the dead-poll-task composition: poll task dies, stale_input_s is
       crossed, the arbiter commands 0 W, and on THAT cycle the alarm is
       already clear - because the detector ran after the 0 W write.
   wired() now returns a per-cycle trace, which is what makes "on which
   cycle" an observation rather than a hope.

2. The NOT COVERED status row. The old check only asserted that an
   uncovered transport is SILENT, which a page claiming `armed` on a guard
   that cannot run also satisfies. The string is now asserted in both
   directions: the uncovered row must say NOT COVERED, name the
   meter_source and homewizard_local, and warn; the covered row must still
   say armed, or "does not say armed" would be satisfied by deleting the
   armed branch.

3. meter_freeze_s. Pinned in test_validate.py by MAINT-04's regex
   technique (no PyYAML in the image), and in test_freeze.py by its
   consequences: the shipped value drives the replays, must stay inside
   the 900 s billing quarter, and 3600/hold_s must stay inside the shipped
   deadband. A Controller built with the key ABSENT must land on the same
   number, which pins main.py's fallback without touching CFG-03.

All three demonstrated red before being accepted green, and every new
check was mutated to confirm it can fail; one candidate assertion ("no new
notification while parked at 0 W") was DELETED because no mutation could
break it, and replaced by the exact notification count across a revived
transport, which can.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015VXjA3LWccN1rJsk3okyPC
This commit is contained in:
glenn schrooyen
2026-08-30 15:50:08 +02:00
co-authored by Claude Opus 5
parent 0cdaa3bd3e
commit c34b5a2922
2 changed files with 377 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
goodwe_controller/test_freeze.py
+322 -26
View File
@@ -2,7 +2,10 @@
python3 test_freeze.py
Three parts, and the last two are the ones that matter:
Six parts. Parts 2 and 3 are the acceptance argument; parts 5 and 6 exist
because TEST-06 found three mutations that survived parts 1-4 green - the
evaluation POSITION inside cycle(), the NOT COVERED status string, and the
shipped `meter_freeze_s` itself:
1. the predicate's boundaries and the edge-trigger, table-driven;
2. SILENT ON REAL CONVERGED DATA. `scenarios/p1-age-2026-08-26.jsonl.gz` is
@@ -18,6 +21,14 @@ Three parts, and the last two are the ones that matter:
repo is next to this one (or `$GOODWE_SIM`), and an equivalent local stub
otherwise. Both are exercised; the run says which one it used.
4. THE WIRING: the real `Controller.cycle()`, asserted on the service call
that leaves the process;
5. WHERE IN `cycle()` the detector is evaluated - by consequence, not by
source order. It runs after `write_setpoint()`, and that is what makes two
compositions safe (see the section for both);
6. WHAT THE STATUS PAGE SAYS about an uncovered transport, and the shipped
threshold read back out of `config.yaml`.
⚠️ Part 3 runs in WALL CLOCK - `unchanged_s` is `time.monotonic()` and the
transport really polls - so it drives the detector at `hold_s=10`, not the
shipped 300. The THRESHOLD is config; the MECHANISM is what is under test, and
@@ -57,6 +68,7 @@ import asyncio
import gzip
import json
import os
import re
import subprocess
import sys
import time
@@ -79,6 +91,29 @@ HERE = Path(__file__).resolve().parent
# The live install's numbers, which are the ones the exposure is written at.
LIVE_MAX_W = 5000.0
# --------------------------------------------------------------------------- #
# THE SHIPPED THRESHOLD, READ FROM config.yaml (TEST-06 / AC3)
# --------------------------------------------------------------------------- #
# ⚠️ `meter_freeze_s: 300` used to appear in this file only as the literal 300,
# retyped six times, so `300 -> 3600` in config.yaml left every check here
# green. Everything below is now driven off the SHIPPED value, and the value
# itself is pinned in test_validate.py the way MAINT-04 pins the other defaults.
#
# Regex, not PyYAML: the add-on image installs py3-aiohttp/py3-paho-mqtt only,
# and a check that skips on ImportError is not a check.
_CFG_TXT = (HERE / "config.yaml").read_text(encoding="utf-8")
def _cfg_default(key):
"""The single ` key: <number>` default line in config.yaml's options block."""
found = re.findall(rf"^ {key}: (-?\d+(?:\.\d+)?)\s*$", _CFG_TXT, re.M)
return float(found[0]) if len(found) == 1 else None
SHIPPED_HOLD_S = _cfg_default("meter_freeze_s")
SHIPPED_DEADBAND_W = _cfg_default("deadband_w")
SHIPPED_MAX_W = _cfg_default("max_w")
def check(name, cond):
print(f" {'ok ' if cond else 'FAIL'} {name}")
@@ -135,6 +170,35 @@ check("the reason names both terms",
"unchanged over 300" in wd.reason and "command over 4750" in wd.reason)
# --------------------------------------------------------------------------- #
# 1b. the SHIPPED threshold, and the two properties it has to have (AC3)
# --------------------------------------------------------------------------- #
# ⚠️ This is the number the whole ticket is an argument about, and it was pinned
# nowhere: `meter_freeze_s: 300 -> 3600` used to leave this file and
# test_validate.py both green. test_validate.py pins the literal in MAINT-04's
# manner; here it is pinned by its CONSEQUENCES, so a schema-legal value that
# gives up the argument goes red even though it is a legal value.
print("the shipped meter_freeze_s (AC3)")
check(f"config.yaml's numeric defaults were read (freeze {SHIPPED_HOLD_S}, "
f"deadband {SHIPPED_DEADBAND_W}, max_w {SHIPPED_MAX_W})",
None not in (SHIPPED_HOLD_S, SHIPPED_DEADBAND_W, SHIPPED_MAX_W))
check("the code default and the shipped default are the same number - "
"FreezeWatchdogConfig() is a default, config.yaml is the ship",
FreezeWatchdogConfig().hold_s == SHIPPED_HOLD_S)
check("this file's `tuning()` really is config.yaml's deadband, so the "
"mutual-exclusion arithmetic below is about the shipped law",
tuning().deadband_w == SHIPPED_DEADBAND_W)
# The upper bound is money: config.yaml's own comment says 300 s is "still
# inside the 900 s quarter that bills the capacity tariff". A hold that outlives
# the quarter notifies after the peak is already billed.
QUARTER_S = 900.0
check(f"the alarm still lands inside the {QUARTER_S:.0f} s billing quarter "
f"({SHIPPED_HOLD_S:.0f} s hold)", SHIPPED_HOLD_S < QUARTER_S)
# The lower bound is the conjunction itself - see the derivation below, which
# now runs off the shipped numbers rather than off two retyped literals.
# --------------------------------------------------------------------------- #
# 2. silent on 40 minutes of real converged behaviour
# --------------------------------------------------------------------------- #
@@ -183,10 +247,10 @@ def replay(hold_s, max_w, cycle_floor_s=15.0):
for mw in (2000.0, LIVE_MAX_W):
wd_r, peak_cmd, peak_unchanged, fired = replay(300.0, mw)
wd_r, peak_cmd, peak_unchanged, fired = replay(SHIPPED_HOLD_S, mw)
print(f" max_w {mw:.0f}: worst |command| {peak_cmd:.0f} W (rail "
f"{rail_w(mw):.0f} W), worst unchanged {peak_unchanged:.1f} s "
f"(hold 300 s), episodes {wd_r.episodes}")
f"(hold {SHIPPED_HOLD_S:.0f} s), episodes {wd_r.episodes}")
check(f"max_w {mw:.0f}: never alarms on the converged capture", fired is None)
check(f"max_w {mw:.0f}: and the rail term had margin to spare",
peak_cmd < rail_w(mw))
@@ -194,7 +258,7 @@ for mw in (2000.0, LIVE_MAX_W):
# The margin on the OTHER term, as numbers rather than as trust. Both figures
# are quoted in config.yaml and freeze.py, so they are re-measured every run and
# a capture swapped underneath them cannot leave the prose saying the old thing.
_, _, worst_unchanged, _ = replay(300.0, LIVE_MAX_W)
_, _, worst_unchanged, _ = replay(SHIPPED_HOLD_S, LIVE_MAX_W)
gaps, last, prev = [], rows[0]["t"], None
for r in rows:
if prev is None or r["pwr"] != prev:
@@ -208,7 +272,8 @@ check("worst observed unchanged_s is the 8 s the docs quote",
7.5 < worst_unchanged < 8.5)
check("longest distinct-reading interval is the 15.4 s the docs quote",
15.0 < max(gaps) < 16.0)
check("300 s is at least 15x the worst of the two", 300.0 >= 15 * max(gaps))
check(f"the shipped {SHIPPED_HOLD_S:.0f} s is at least 15x the worst of the two",
SHIPPED_HOLD_S >= 15 * max(gaps))
# --------------------------------------------------------------------------- #
# 2b. TEL-05's case, reconstructed, because the capture cannot contain it
@@ -226,22 +291,26 @@ check("300 s is at least 15x the worst of the two", 300.0 >= 15 * max(gaps))
# -14.2 W) rather than invented: one hour of a perfectly healthy converged meter
# that happens not to move.
print("TEL-05's low-power case - a converged meter that legitimately repeats")
tel05 = FreezeWatchdog(FreezeWatchdogConfig(hold_s=300))
tel05 = FreezeWatchdog(FreezeWatchdogConfig(hold_s=SHIPPED_HOLD_S))
t5 = tuning(LIVE_MAX_W)
tgt, iw, sc = 0.0, 0.0, 0
bare_would_fire = False
tel05_cmd = 0.0
for k in range(3600):
# Long enough that the REJECTED bare threshold has to get its chance even if
# someone raises meter_freeze_s - the mutation must be caught by the pins above,
# not by this case going red for an incidental reason.
TEL05_S = max(3600, int(SHIPPED_HOLD_S) + 300)
for k in range(TEL05_S):
if k % 15 == 0: # cycle_floor_s, the only trigger left
d = compute(prev_w=tgt, grid_w=-14.2, actual_w=tgt, tuning=t5,
sat_count=sc, i_w=iw)
tgt, iw, sc = d.target_w, d.i_w, d.sat_count
tel05_cmd = max(tel05_cmd, abs(tgt))
# The REJECTED design, written out so the case can be shown to be real.
bare_would_fire = bare_would_fire or float(k) >= 300
bare_would_fire = bare_would_fire or float(k) >= SHIPPED_HOLD_S
tel05.evaluate(float(k), tgt, LIVE_MAX_W, T0 + timedelta(seconds=k))
print(f" an hour at -14.2 W: worst |command| {tel05_cmd:.0f} W, "
f"unchanged_s reaches 3599 s, episodes {tel05.episodes}")
print(f" {TEL05_S / 60:.0f} min at -14.2 W: worst |command| {tel05_cmd:.0f} W, "
f"unchanged_s reaches {TEL05_S - 1} s, episodes {tel05.episodes}")
check("the bare unchanged_s threshold TEL-05 rejected WOULD fire here",
bare_would_fire)
check("the conjunction does not - this is what the rail term buys",
@@ -255,8 +324,11 @@ check("...and the command stayed nowhere near the rail",
# figure is inside the deadband, so the command there is parked at 0 W. If
# someone lowers meter_freeze_s or narrows the deadband, this is the check that
# should stop them.
DEADBAND_W = 15.0
slowest_healthy_w = 3600.0 / 300.0
DEADBAND_W = SHIPPED_DEADBAND_W # config.yaml's, not a retyped 15
slowest_healthy_w = 3600.0 / SHIPPED_HOLD_S
print(f" at the shipped {SHIPPED_HOLD_S:.0f} s hold, the slowest healthy "
f"meter that still looks unchanged is {slowest_healthy_w:.1f} W, against "
f"a {DEADBAND_W:.0f} W deadband")
check("a healthy meter slow enough to look unchanged is inside the deadband",
slowest_healthy_w < DEADBAND_W)
_park = compute(prev_w=0.0, grid_w=-slowest_healthy_w, actual_w=0.0,
@@ -454,11 +526,17 @@ class _Hass:
# runaway under test never happens - which would have made this a check
# that passes for the wrong reason.
self.batt = 0.0
# ⚠️ The dead-poll-task case (AC1's second half): every input reads
# None, exactly as it does when the transport task has died while the
# controller keeps cycling.
self.dead = False
async def limits(self, entity):
return (-5000.0, 5000.0, 1.0)
async def number(self, entity, invert=False):
if self.dead:
return None
return {"sensor.grid": self.grid, "sensor.soc": 50.0,
"sensor.batt": self.batt}.get(entity)
@@ -479,29 +557,54 @@ class _Source:
self.unchanged_s = unchanged_s
# ⚠️ meter_freeze_s is deliberately NOT in here. A controller built with the key
# absent must land on the shipped number by itself, and that is asserted below -
# supplying it here would test this dict instead of the shipped default.
STALE_INPUT_S = 15.0
def _opts(**over):
o = {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
"soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
"meter_source": "off", "max_w": LIVE_MAX_W, "gain": 0.6,
"slew_w": 1000, "deadband_w": SHIPPED_DEADBAND_W,
"target_grid_w": -10, "step_w": 10, "heartbeat_s": 10,
"stale_input_s": STALE_INPUT_S}
o.update(over)
return o
def notes_of(hass):
"""The HA persistent notifications that have left the process so far."""
return [d for dom, svc, d in hass.calls
if (dom, svc) == ("persistent_notification", "create")]
def wired(unchanged_s, grid, cycles=60, **over):
"""Returns (hass, controller, trace).
`trace` is one row per cycle - (command after the cycle, the watchdog's
level, notifications raised so far) - which is what AC1 needs: the
QUESTION IS NOT WHETHER THE ALARM HAPPENS BUT ON WHICH CYCLE.
"""
async def go():
hass = _Hass(grid)
c = Controller(dict(
{"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
"soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
"meter_source": "off", "max_w": LIVE_MAX_W, "gain": 0.6,
"slew_w": 1000, "deadband_w": 15, "target_grid_w": -10,
"step_w": 10, "heartbeat_s": 10, "stale_input_s": 15},
**over), hass, _Store(), _Pub(), stop=asyncio.Event())
c = Controller(_opts(**over), hass, _Store(), _Pub(),
stop=asyncio.Event())
c.auto = True
c.dev_min, c.dev_max = -5000.0, 5000.0
c.p1_source = _Source(unchanged_s)
trace = []
for _ in range(cycles):
await c.read_inputs()
await c.cycle()
return hass, c
trace.append((c.target, c.freeze_wd.active, len(notes_of(hass))))
return hass, c, trace
return asyncio.run(go())
hass_f, ctl_f = wired(3600.0, 4000.0)
notes = [d for dom, svc, d in hass_f.calls
if (dom, svc) == ("persistent_notification", "create")]
hass_f, ctl_f, trace_f = wired(3600.0, 4000.0)
notes = notes_of(hass_f)
# The status row lives behind `p1_enabled`, which is what a homewizard_local
# install has. Flipped here rather than plumbing a whole P1 transport in - the
# row's own content is what is under test.
@@ -522,16 +625,209 @@ check("the event log names it", any("METER FROZEN" in e for e in ctl_f.events))
check("and it does NOT command 0 W - this cut is an alarm only",
abs(ctl_f.target) >= rail_w(LIVE_MAX_W))
hass_n, ctl_n = wired(3600.0, -14.2)
hass_n, ctl_n, _ = wired(3600.0, -14.2)
check("wired up, TEL-05's converged case still raises nothing",
not [1 for dom, svc, _ in hass_n.calls if svc == "create"])
hass_m, ctl_m = wired(2.0, 8000.0)
hass_m, ctl_m, _ = wired(2.0, 8000.0)
check("wired up, a real railed load against a moving meter raises nothing",
not [1 for dom, svc, _ in hass_m.calls if svc == "create"]
and abs(ctl_m.target) >= rail_w(LIVE_MAX_W))
check("an uncovered transport (no unchanged_s) raises nothing, and says so",
check("an uncovered transport (no unchanged_s) raises no notification",
not [1 for dom, svc, _ in wired(None, 4000.0)[0].calls if svc == "create"])
# --------------------------------------------------------------------------- #
# 5. WHERE in cycle() the detector is evaluated (TEST-06 / AC1)
# --------------------------------------------------------------------------- #
# ⚠️ WHY THIS IS NOT STYLE. `check_meter_frozen()` runs LAST, after
# `write_setpoint()`, and moving it to the top of `cycle()` used to leave this
# whole file green. It is not a tidiness preference, it is the reason two
# separate compositions are safe, and each is asserted here by an OBSERVABLE
# CONSEQUENCE rather than by the source order - so a refactor that keeps the
# semantics (renaming it, inlining it, calling it from a helper that runs at the
# end) stays green, and one that changes WHICH command it judges goes red:
#
# (a) BOTH TERMS MUST DESCRIBE THE SAME MOMENT. The command it grades has to
# be the one that just went on the wire, judged against how long the meter
# that chose it has stood still. Evaluated at the top it grades the
# PREVIOUS cycle's command against THIS cycle's meter - a pairing that was
# never true of any moment - and the alarm arrives a whole cycle late.
# (b) THE DEAD-POLL-TASK COMPOSITION. When the poll task dies the inputs go
# stale, the arbiter commands 0 W, and the detector - evaluated after that
# write - correctly sees an unrailed command and CLEARS. At the top of the
# cycle it would still be looking at the railed command from before the
# failsafe, and would keep the alarm standing over a controller that is
# already commanding 0 W.
print("WHERE the detector is evaluated in cycle() - the ordering AC1 protects")
_railed = [i for i, (t, _, _) in enumerate(trace_f)
if abs(t) >= rail_w(LIVE_MAX_W)]
_alarmed = [i for i, (_, _, n) in enumerate(trace_f) if n >= 1]
_first_railed = _railed[0] if _railed else None
_first_alarm = _alarmed[0] if _alarmed else None
print(f" ramp to the rail: first railed command on cycle {_first_railed}, "
f"first notification on cycle {_first_alarm}, over {len(trace_f)} cycles")
# Without this the equality below could be satisfied on cycle 0, where "the same
# cycle" and "one cycle later" are indistinguishable.
check("the command takes several cycles to reach the rail, so 'same cycle' and "
"'one cycle later' are different observations",
_first_railed is not None and _first_railed >= 1
and len(trace_f) > _first_railed + 1)
check("the alarm lands on the SAME cycle the command reaches the rail - "
"evaluated after write_setpoint() it grades the command now on the wire; "
"moved to the top of cycle() it would grade the previous one and arrive a "
"cycle late",
_first_alarm is not None and _first_alarm == _first_railed)
def dead_poll(after_cycles=20, dead_cycles=4, revived_cycles=20):
"""Alarm on a railed command, then kill the poll task and keep cycling.
The real thing: the transport task dies, every input reads None, and after
`stale_input_s` the arbiter's P_SAFETY stop commands 0 W. `inputs_bad_since`
is backdated rather than slept through - the clock is wall clock, and the
mechanism rather than the wait is what is under test.
"""
async def go():
hass = _Hass(4000.0)
c = Controller(_opts(), hass, _Store(), _Pub(), stop=asyncio.Event())
c.auto = True
c.dev_min, c.dev_max = -5000.0, 5000.0
c.p1_source = _Source(3600.0)
for _ in range(after_cycles):
await c.read_inputs()
await c.cycle()
before = (c.target, c.freeze_wd.active, len(notes_of(hass)))
hass.dead = True
trace = []
for _ in range(dead_cycles):
await c.read_inputs()
if c.inputs_bad_since is not None:
c.inputs_bad_since = (datetime.now(timezone.utc)
- timedelta(seconds=STALE_INPUT_S + 1))
await c.cycle()
trace.append((c.target, c.freeze_wd.active, len(notes_of(hass))))
# ...and then the transport comes back, against a meter that is still
# frozen. The loop re-rails and this is a NEW episode - the alarm is
# allowed, indeed required, to speak again.
hass.dead = False
revived = []
for _ in range(revived_cycles):
await c.read_inputs()
await c.cycle()
revived.append((c.target, c.freeze_wd.active, len(notes_of(hass))))
return before, trace, revived, notes_of(hass), c
return asyncio.run(go())
_before, _dead, _revived, _all_notes, _ctl_d = dead_poll()
_zeroed = [i for i, (t, _, _) in enumerate(_dead) if t == 0.0]
print(f" poll task killed: alarming at {_before[0]:+.0f} W "
f"(active={_before[1]}, {_before[2]} notification(s)) -> "
+ ", ".join(f"cycle {i}: {t:+.0f} W active={a} notes={n}"
for i, (t, a, n) in enumerate(_dead)))
check("the episode really was standing before the poll task died - otherwise "
"the release below is vacuous",
_before[1] and _before[2] == 1 and abs(_before[0]) >= rail_w(LIVE_MAX_W))
check("a dead poll task drives the command to 0 W",
bool(_zeroed) and _zeroed[0] == 0)
check("...and on THAT SAME cycle the alarm is already clear, because the "
"detector ran after the 0 W write and not before it",
bool(_zeroed) and not _dead[_zeroed[0]][1])
check("...and it stays clear for as long as the failsafe holds it there",
_dead and not any(a for _, a, _ in _dead))
# ⚠️ "and it raises nothing new while parked" is NOT asserted here, and that is
# deliberate: with the alarm cleared there is no mutation of this code that
# could make a notification appear during those cycles, so the assertion could
# not fail and would have been decoration. The real claim - that the outage did
# not add notifications - is carried by the exact count below, which CAN fail.
# ⚠️ AND THE OTHER HALF, or "no second notification" is a claim no mutation can
# break: when the transport comes back against a meter that is STILL frozen the
# loop re-rails, and that is a new episode which MUST speak. What must not
# happen is a second notification stacking up beside the first.
_re_railed = [i for i, (t, _, _) in enumerate(_revived)
if abs(t) >= rail_w(LIVE_MAX_W)]
print(f" transport revived: command re-rails on cycle "
f"{_re_railed[0] if _re_railed else None}, "
f"{_ctl_d.freeze_wd.episodes} episode(s), {len(_all_notes)} notification(s), "
f"ids {sorted({n.get('notification_id') for n in _all_notes})}")
check("a revived transport against a still-frozen meter re-rails the command",
bool(_re_railed))
check("...which is a SECOND episode, and it alarms again on the cycle it rails",
_ctl_d.freeze_wd.episodes == 2
and bool(_re_railed) and _revived[_re_railed[0]][2] == _before[2] + 1)
check("...on the same notification_id, so the second REPLACES the first "
"instead of stacking a second one beside it",
len(_all_notes) == 2
and len({n.get("notification_id") for n in _all_notes}) == 1)
# --------------------------------------------------------------------------- #
# 6. what the status page SAYS about coverage (TEST-06 / AC2)
# --------------------------------------------------------------------------- #
# ⚠️ The check above only ever asserted that an uncovered transport is SILENT,
# which a status page claiming `armed` on a guard that cannot run also satisfies.
# The AC was that the detector reports uncovered transports *and says so*, so
# the string is what gets asserted here, in both directions: the uncovered row
# must not claim `armed`, and the covered row must still claim it - or "does not
# say armed" would be satisfiable by deleting the armed branch outright.
print("what the status page SAYS about coverage")
def freeze_rows(unchanged_s, grid, **over):
"""The freeze-detector rows of the real checks(), on a p1 install."""
_h, ctl, _t = wired(unchanged_s, grid, cycles=3, **over)
# p1_enabled is what a homewizard_local install has; the row's own content
# is what is under test, not the transport that feeds it.
ctl.p1_enabled = True
return ctl, [r for r in ctl.checks()
if r["text"].startswith("freeze detector")]
_ctl_u, _rows_u = freeze_rows(None, 4000.0, meter_source="ha_dsmr")
_ctl_c, _rows_c = freeze_rows(2.0, -14.2)
print(f" uncovered: {_rows_u[0]['text'] if _rows_u else '(no row at all)'}")
print(f" covered: {_rows_c[0]['text'] if _rows_c else '(no row at all)'}")
check("an uncovered transport gets exactly one freeze-detector row",
len(_rows_u) == 1)
check("...and it says NOT COVERED",
bool(_rows_u) and "NOT COVERED" in _rows_u[0]["text"])
check("...and it does NOT claim the guard is armed - that would be the status "
"page lying on behalf of a guard that cannot run",
bool(_rows_u) and "armed" not in _rows_u[0]["text"])
check("...and it names the meter_source it is not covering, so the operator "
"knows which install this is",
bool(_rows_u) and "ha_dsmr" in _rows_u[0]["text"])
check("...and it names homewizard_local as the transport that would be covered",
bool(_rows_u) and "homewizard_local" in _rows_u[0]["text"])
check("...flagged as a warning rather than a silent ok",
bool(_rows_u) and not _rows_u[0]["ok"] and _rows_u[0]["warn"])
check("a COVERED transport gets exactly one freeze-detector row",
len(_rows_c) == 1)
check("...and that one DOES say armed - without this, 'the uncovered row does "
"not say armed' is satisfied by deleting the armed branch",
bool(_rows_c) and "armed" in _rows_c[0]["text"]
and "NOT COVERED" not in _rows_c[0]["text"])
check("...and it quotes the hold and the rail it is armed at, derived from the "
"config the controller was actually built with",
bool(_rows_c)
and f"{_ctl_c.freeze_wd.cfg.hold_s:.0f} s unchanged" in _rows_c[0]["text"]
and f"{rail_w(LIVE_MAX_W):.0f} W" in _rows_c[0]["text"])
check("...as an ok row, since a covered armed guard is not a warning",
bool(_rows_c) and _rows_c[0]["ok"] and not _rows_c[0]["warn"])
# AC3's other half, through the built object rather than through a literal: the
# opts dict does NOT carry meter_freeze_s, so this is the number an install that
# leaves the key at its config.yaml default actually lands on.
print(f" a Controller built with meter_freeze_s ABSENT holds "
f"{_ctl_c.freeze_wd.cfg.hold_s:.0f} s (config.yaml ships "
f"{SHIPPED_HOLD_S:.0f} s)")
check("a Controller built with the key absent lands on the SHIPPED hold, so "
"config.yaml and main.py's fallback cannot drift apart unnoticed",
_ctl_c.freeze_wd.cfg.hold_s == SHIPPED_HOLD_S)
print()
if fails:
print(f"{len(fails)} FAILED: {', '.join(fails)}")
goodwe_controller/test_validate.py
+55
View File
@@ -253,6 +253,61 @@ check("...and the shipped charge rate is at or under the shipped ceiling",
check("...and the shipped discharge rate too, so it warns about nothing",
_shipped["maintenance_discharge_w"] <= _shipped["max_w"])
# --------------------------------------------------------------------------- #
print("the shipped meter_freeze_s, pinned (TEST-06 / AC3)")
# ⚠️ WHY A NUMBER GETS ITS OWN CHECK. `meter_freeze_s: 300` is the threshold
# SAFETY-18's entire argument is about, and before this it was pinned NOWHERE:
# 300 -> 3600 left test_freeze.py AND this file green, because the schema
# accepts int(60,3600) and no check ever read the shipped value. The schema
# bounds what is CONFIGURABLE; this pins what is SHIPPED, which is a different
# question and the one the docs answer.
#
# Read by regex like _shipped above, for the same reason: no PyYAML in the
# image, and a check that skips on ImportError is not a check.
_freeze_s = [float(v) for v in re.findall(
r"^ meter_freeze_s: (\d+(?:\.\d+)?)\s*$", _txt, re.M)]
_freeze_schema = re.findall(r"^ meter_freeze_s: int\((\d+),(\d+)\)\s*$",
_txt, re.M)
check(f"config.yaml carries exactly one meter_freeze_s default {_freeze_s}",
len(_freeze_s) == 1)
check("...and exactly one schema range for it", len(_freeze_schema) == 1)
_hold_s = _freeze_s[0] if _freeze_s else None
_lo, _hi = ((float(_freeze_schema[0][0]), float(_freeze_schema[0][1]))
if _freeze_schema else (None, None))
check("the shipped meter_freeze_s is still the measured 300 s that freeze.py, "
"config.yaml's own comment and test_freeze.py are all written at",
_hold_s == 300.0)
check(f"...and it is inside its own schema range ({_lo}-{_hi} s)",
None not in (_lo, _hi, _hold_s) and _lo <= _hold_s <= _hi)
# The two properties the number has to have, DERIVED rather than restated - so
# they go red on a value that is merely schema-legal. Both come straight from
# config.yaml's own comment on the key.
#
# upper: the alert must land inside the 900 s quarter that bills the capacity
# tariff, or the notification arrives after the money is spent;
# lower: the slowest a HEALTHY meter can tick a 1 Wh register and still look
# unchanged for hold_s is 3600/hold_s W, and that must stay inside the
# deadband or the conjunction's two terms stop being mutually
# exclusive (freeze.py, "WHERE hold_s COMES FROM").
_QUARTER_S = 900.0
_shipped_deadband = [float(v) for v in re.findall(
r"^ deadband_w: (\d+(?:\.\d+)?)\s*$", _txt, re.M)]
check(f"config.yaml carries exactly one deadband_w default {_shipped_deadband}",
len(_shipped_deadband) == 1)
_db = _shipped_deadband[0] if _shipped_deadband else None
_slowest_healthy_w = 3600.0 / _hold_s if _hold_s else None
print(f" shipped meter_freeze_s {_hold_s} s: alert inside the "
f"{_QUARTER_S:.0f} s billing quarter, and a healthy meter slow enough to "
f"look unchanged for it is {_slowest_healthy_w} W against a "
f"{_db} W deadband")
check(f"...the alarm still lands inside the {_QUARTER_S:.0f} s billing quarter",
_hold_s is not None and _hold_s < _QUARTER_S)
check("...and a healthy meter that looks unchanged for it is still inside the "
"deadband, so the two terms stay mutually exclusive",
_slowest_healthy_w is not None and _db is not None
and _slowest_healthy_w < _db)
# --------------------------------------------------------------------------- #
print("the KNOWN cross-file default divergence, pinned rather than commented")
# ⚠️ WHAT THIS GUARDS, AND WHY IT ASSERTS THE BUG RATHER THAN THE FIX.