@@ -4,7 +4,8 @@
No framework , asserts only , same house style as test_freeze . py .
Five parts , and parts 3 and 4 are the ones that carry the acceptance criteria :
Six parts , and parts 3 and 4 are the ones that carry SAFETY - 03 ' s own
acceptance criteria ; part 6 is TEST - 11 ' s:
1. THE CLAMP ARITHMETIC , on the shipped ` clamp_to_rating ( ) ` - boundaries ,
symmetry , the device second stage , and the non - finite cases . Table driven .
@@ -23,6 +24,13 @@ Five parts, and parts 3 and 4 are the ones that carry the acceptance criteria:
5. NO PATH BYPASSES IT . write_setpoint ( ) is the only route to the wire , so it
is called directly with an out - of - range value the way the heartbeat and the
operator ' s buttons call it.
6. TEST - 11. THE GUARD READS THE WRITE THAT JUST HAPPENED . ` self . clamp_last `
is assigned by ` write_setpoint ( ) ` and by nothing else , so a
` check_output_clamp ( ) ` moved above the write keeps its loop - side evidence
and silently loses its write - side half . Graded as a CONSEQUENCE - what the
watchdog could see when it was consulted - so that a rename or a helper
does not trip it , and with the write - path half exercised on a stimulus of
its own , because parts 3 - 5 never produce one through cycle ( ) .
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
WHY PART 4 ALSO ASSERTS ON ` Decision ` ' s BOOLEANS AND NOT ON `reason`
@@ -550,6 +558,240 @@ check("an in-range value is untouched", hass_d.writes[3] == 1500.0)
check ( " every emitted frame satisfies |emitted| <= the rating " ,
all ( abs ( w ) < = 2000.0 for w in hass_d . writes ) )
# ==============================================================================
print ( )
print ( " 6. TEST-11: the guard ' s evidence is THIS cycle ' s write, not the last one " )
# WHY THIS PART EXISTS. `check_output_clamp()` sits one line below
# `await self.write_setpoint(self.target)` in cycle(), and moving it above the
# write left this file, test_freeze.py and test_loop.py ALL GREEN. That is not a
# missing assertion about line numbers - it is a missing assertion about
# EVIDENCE. The guard has two sources:
#
# * `self.loop_clamped`, set from the Decision well ABOVE the write. Still
# seen wherever the call is put, which is why the move is nearly invisible:
# on the §10.3 replay above, every engagement the alarm reports is this one.
# * `self.clamp_last`, which ONLY `write_setpoint()` assigns (main.py) and
# which is None until it has run. This half is what loses its evidence.
#
# So the check below does not pin the call's POSITION IN THE SOURCE, and must
# not: the comment at that line warns against tidying SAFETY-03's ordering and
# SAFETY-18's into one rule, and a check a rename breaks is worse than no check.
# It pins the CONSEQUENCE - that when the guard is consulted, `self.clamp_last`
# is the result of the write of THIS cycle - and part 6b runs the identical
# assertions through a deliberately refactored controller to prove it.
class _SpyWatchdog ( ClampWatchdog ) :
""" The shipped watchdog, plus a record of what the guard could SEE.
⚠ ️ IT HOOKS THE CONSUMER , NOT THE GUARD ' S NAME. Wrapping
` check_output_clamp ` by name would make a rename - AC 2 ' s legitimate
refactor - look identical to the defect , and this ticket is explicitly not
allowed to buy its check with a nuisance . ` clamp_wd . note ( ) ` is the one call
the guard exists to make , so it is hooked instead : it is reached from the
guard under any name , from any helper , as long as that helper runs after the
write .
"""
def __init__ ( self ) :
super ( ) . __init__ ( )
self . owner = None
self . seen = [ ] # clamp_last.requested_w at each consultation
self . seen_write = [ ] # was the WRITE-path half engaged, per cycle
def note ( self , engaged , reason , now ) :
cl = self . owner . clamp_last
self . seen . append ( None if cl is None else cl . requested_w )
self . seen_write . append ( bool ( cl and cl . engaged ) )
return super ( ) . note ( engaged , reason , now )
def probed ( cls = Controller , grid = 9000.0 , cycles = 5 , * * over ) :
""" `wired()`, with the watchdog replaced by the spy and the per-cycle
command returned . A RAMP is used on purpose : the slew limiter makes every
cycle ' s command a different number, which is what makes " this cycle ' s
evidence " and " last cycle ' s evidence " distinguishable at all. " " "
async def go ( ) :
# `track=True` - the plant follows, so the saturation freeze never
# engages and the slew limiter keeps the command MOVING; against a
# dead plant the freeze pins it flat after three cycles and the ramp
# stops being a ramp.
hass = _Hass ( grid , track = True )
c = cls ( 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 " : 0 ,
" step_w " : 10 , " heartbeat_s " : 10 , " stale_input_s " : 15 } ,
* * over ) , hass , _Store ( ) , _Pub ( ) , stop = asyncio . Event ( ) )
c . auto = True
c . dev_min , c . dev_max = - LIVE_MAX_W , LIVE_MAX_W
c . clamp_wd = _SpyWatchdog ( )
c . clamp_wd . owner = c
asked = [ ]
for _ in range ( cycles ) :
await c . read_inputs ( )
await c . cycle ( )
asked . append ( c . target )
return hass , c , asked
return asyncio . run ( go ( ) )
def evidence_checks ( label , asked , wd ) :
""" The same four claims, run against the shipped controller (6a) and against
a refactored one ( 6 b ) . Sharing them is the point : if a rename or a helper
changed the answer , this function would say so in the same words . """
seen = wd . seen
print ( f " { label } : asked "
f " { [ f ' { w : +.0f } ' for w in asked ] } " )
print ( f " { label } : the guard saw "
f " { [ ' None ' if s is None else f ' { s : +.0f } ' for s in seen ] } " )
check ( f " { label } : PRECONDITION - the command MOVES on every cycle, so this "
f " cycle ' s evidence and last cycle ' s are different numbers " ,
len ( set ( asked ) ) == len ( asked ) and len ( asked ) > = 4 )
check ( f " { label } : the guard was consulted exactly once per cycle " ,
len ( seen ) == len ( asked ) )
check ( f " { label } : AC 1 - on EVERY cycle self.clamp_last held the result of "
f " the write THAT CYCLE. write_setpoint() is the only line that "
f " assigns self.clamp_last, so a check_output_clamp() moved above the "
f " write grades the PREVIOUS cycle ' s write instead " ,
seen == asked )
check ( f " { label } : ...and on the FIRST cycle self.clamp_last was already set "
f " - before the run ' s first write it is None, i.e. the write-path half "
f " of the guard has no evidence at all " ,
bool ( seen ) and seen [ 0 ] is not None )
hass_e , ctl_e , asked_e = probed ( )
evidence_checks ( " shipped " , asked_e , ctl_e . clamp_wd )
# ⚠️ AND HERE IS WHY AC 3 NEEDS A STIMULUS OF ITS OWN, MEASURED RATHER THAN
# ASSERTED. Through the real cycle() the write-path half never engages: the
# arbiter files the device range and max_w as P_SAFETY LIMIT claims, so
# `resolution.target_w` is already inside both bounds before write_setpoint()
# re-applies them (main.py calls that duplication defence in depth, and it is).
# Every engagement the alarm reports on this path is therefore the loop-side
# one - which is exactly what makes the move nearly invisible, and exactly why
# 6c below drives the write path directly instead of hoping cycle() will.
check ( f " MEASURED: the write-path half engaged 0 times in { len ( asked_e ) } real "
f " cycles ( { sum ( ctl_e . clamp_wd . seen_write ) } ), so no cycle()-driven check "
f " can be watching it " , not any ( ctl_e . clamp_wd . seen_write ) )
class _RefactoredController ( Controller ) :
""" AC 2, as a fixture rather than as a promise: the two refactors a P3 tidy
would actually make , applied at once .
* the guard is RENAMED to ` _clamp_guard ( ) ` , the old name left as the thin
delegate such a rename leaves behind ;
* the write MOVES INTO A HELPER , ` _emit_setpoint ( ) ` .
Neither changes when the guard runs relative to the write , so every check in
` evidence_checks ( ) ` must stay green . If one of them goes red here , the check
is grading the text of main . py and has to be thrown away .
"""
async def _emit_setpoint ( self , value , * , force = False ) :
await Controller . write_setpoint ( self , value , force = force )
async def write_setpoint ( self , value , * , force = False ) :
await self . _emit_setpoint ( value , force = force )
async def _clamp_guard ( self ) :
await Controller . check_output_clamp ( self )
async def check_output_clamp ( self ) :
await self . _clamp_guard ( )
# ⚠️ VERIFIED AGAINST main.py ITSELF, NOT ONLY AGAINST THIS SUBCLASS. Renaming
# `check_output_clamp` throughout main.py AND moving both the write and the guard
# into a new `_emit_and_guard()` helper called from cycle() leaves every check in
# this part green; moving the guard above the write turns the two AC 1 lines
# below red. The one thing a repo-wide rename does cost is the direct call in 6c,
# which is a mechanical find/replace and fails as an AttributeError naming the
# new method - not as an assertion claiming the ordering is wrong, which is the
# false signal this part was written to avoid.
hass_f , ctl_f , asked_f = probed ( cls = _RefactoredController )
evidence_checks ( " renamed + helper " , asked_f , ctl_f . clamp_wd )
check ( " AC 2: the refactored controller behaves identically to the shipped one "
" - same commands, same evidence, so the check above is not pinning "
" either name " ,
asked_f == asked_e and ctl_f . clamp_wd . seen == ctl_e . clamp_wd . seen )
# --------------------------------------------------------------------------- #
# 6c. AC 3 - the WRITE-PATH half, exercised
# --------------------------------------------------------------------------- #
# The stimulus is SAFETY-20's shape and that is not a coincidence: `max_w` 5000
# on an inverter that reports +-2500 is the one configuration where the value
# put on the wire DIFFERS from the value the loop asked for while every loop-side
# flag stays clean. That difference is the whole content of `clamp_last`, and a
# DEVICE-bound engagement is the only kind `loop_clamped` cannot also see.
#
# It is driven through `write_setpoint()` and the guard directly rather than
# through cycle(), because the measurement above says cycle() cannot produce it
# today. That is stated as a limitation, not hidden: what is being graded here is
# the guard's own contract - given a write that clamped, alarm - and the ordering
# in cycle() is what 6a grades.
async def _write_path ( moved = False ) :
""" `moved` runs the two calls in the order the ticket asks about. """
hass = _Hass ( 0.0 )
c = Controller ( { " setpoint_entity " : " number.sp " , " meter_source " : " off " ,
" max_w " : LIVE_MAX_W } ,
hass , _Store ( ) , _Pub ( ) , stop = asyncio . Event ( ) )
c . auto = True
# The inverter is SMALLER than the configured ceiling - SAFETY-20's install.
c . dev_min , c . dev_max = - 2500.0 , 2500.0
c . target = 3000.0
if moved :
await c . check_output_clamp ( )
await c . write_setpoint ( c . target )
else :
await c . write_setpoint ( c . target )
await c . check_output_clamp ( )
return hass , c
hass_w , ctl_w = asyncio . run ( _write_path ( ) )
notes_w = clamp_notes ( hass_w )
print ( f " max_w { LIVE_MAX_W : .0f } on a +-2500 W device, asked for "
f " { ctl_w . target : +.0f } W: the device saw { hass_w . writes [ - 1 ] : +.0f } W, "
f " bound= { ctl_w . clamp_last . bound !r} , loop_clamped= { ctl_w . loop_clamped [ 0 ] } " )
check ( " PRECONDITION: the write really did clamp, on the DEVICE bound " ,
ctl_w . clamp_last is not None and ctl_w . clamp_last . engaged
and ctl_w . clamp_last . bound == " device "
and hass_w . writes [ - 1 ] == 2500.0 )
check ( " ...and the loop-side half is SILENT here, so self.clamp_last is the "
" only evidence that this happened at all " ,
ctl_w . loop_clamped == ( False , " " ) )
check ( " AC 3: the guard alarmed on write-path evidence alone " ,
ctl_w . clamp_wd . engagements == 1 and len ( notes_w ) == 1 )
check ( " ...and the operator is told WHICH bound bit, so a device-rating clamp "
" is not reported as a max_w one " ,
bool ( notes_w ) and " device bound " in notes_w [ 0 ] [ " message " ] )
check ( " ...and the action log carries the row, written by write_setpoint() " ,
any ( REASON_CODE in e and " device bound " in e for e in ctl_w . events ) )
# The same stimulus, the same clamp, the two calls swapped. This is the ticket
# in one pair of runs.
hass_m , ctl_m = asyncio . run ( _write_path ( moved = True ) )
print ( f " the same stimulus with the guard consulted BEFORE the write: "
f " { ctl_m . clamp_wd . engagements } engagement(s), "
f " { len ( clamp_notes ( hass_m ) ) } notification(s), and the clamp still "
f " happened (engaged= { ctl_m . clamp_last . engaged } ) " )
check ( " THE DEFECT, DEMONSTRATED: consulted before the write the guard is "
" SILENT - self.clamp_last is still None at that point - while the wire "
" carries a clamped value either way " ,
not clamp_notes ( hass_m ) and ctl_m . clamp_wd . engagements == 0
and ctl_m . clamp_last . engaged and hass_m . writes [ - 1 ] == 2500.0 )
check ( " ...and main.py:856 ' s consumer is the line that goes quietly false " ,
bool ( ctl_w . clamp_last and ctl_w . clamp_last . engaged ) )
print ( )
if fails :
print ( f " { len ( fails ) } FAILED: { ' , ' . join ( fails ) } " )