laundry: one hold per cycle, so the sticky bound is actually a bound

Review of the previous commit caught that its docstring promised more
than the code did. Not restarting an *open* window still let a progress
that flapped out of and back into Finish re-arm a full fresh window once
the first had expired, so the value could be held well past
sticky_seconds from the first Finish. The new test passed only because
its final read left the sticky condition matching; ending the flap on a
non-matching read re-armed and would have failed it.

Make the guarantee real instead of weakening the claim: arming marks the
hold spent, and only sticky_bypass_fn -- a cycle actually running --
clears it. Expiry on its own no longer re-opens the door, because with no
cycle in between a second Finish is the same Finish, and re-arming on it
strobes the entity Finish -> Idle -> Finish once per window, re-firing
the announcements #345 and #358 are both about.

That subsumes the old _sticky_armed edge-trigger flag, which existed to
stop a stuck field extending the window; "spent until a new cycle" covers
that case and the flap case together, before or after expiry.

Also give the flap test real headroom -- it fitted 0.09s of sleeps into a
0.1s window and would have failed spuriously on a loaded runner.
This commit is contained in:
Marc Billow committed 2026-08-12 14:18:56 +00:00
1 parent 2f7170c448
commit f12f67b2b3
3 files changed
+67 -32

No files matched your search

+43 -10
View File
@@ -234,31 +234,64 @@ def test_a_paused_new_cycle_is_left_to_the_window_rather_than_released():
assert sensor.native_value == "Wash"
def test_a_flapping_finish_cannot_ratchet_the_window_forward():
"""Edge-triggering stops a *stuck* Finish from extending the hold, but
a device whose progress flaps out of and back into Finish re-arms --
an already-open window must not restart on that, or the bound stops
being a bound."""
desc = replace(_PROGRESS_DESC, sticky_seconds=0.1)
def test_a_flapping_finish_cannot_ratchet_an_open_window_forward():
"""Edge-triggering stops a *stuck* Finish from extending the hold; a
progress that flaps out of and back into Finish must not restart it
either, or the bound stops being a bound."""
desc = replace(_PROGRESS_DESC, sticky_seconds=0.3)
sensor, coordinator = _sensor(desc)
_replace(coordinator, state="Ready", progress="Finish")
assert sensor.native_value == "Finish"
for _ in range(3):
time.sleep(0.03)
time.sleep(0.05)
_replace(coordinator, state="Ready", progress="None")
assert sensor.native_value == "Finish"
_replace(coordinator, state="Ready", progress="Finish")
assert sensor.native_value == "Finish"
# 0.09s of flapping so far; past 0.1s from the *first* Finish it ends,
# rather than 0.1s from the most recent re-arm.
time.sleep(0.03)
# 0.15s of flapping so far -- the window still ends 0.3s after the
# first Finish, not 0.3s after the most recent re-entry.
time.sleep(0.2)
_replace(coordinator, state="Ready", progress="Finish")
assert sensor.native_value == "Idle"
def test_a_finish_after_the_window_closes_does_not_re_arm_it():
"""Expiry doesn't re-open the door: with no new cycle in between, a
second Finish is the same Finish. Re-arming on it would strobe the
entity Finish -> Idle -> Finish once per window, re-firing exactly the
announcements #345 is about -- so it takes a bypass (a cycle actually
running) to make the hold available again.
Distinct from the flap test above: there the window is still open, and
the last read before expiry leaves the sticky condition *matching*.
Here it has already closed, and the flap ends on a non-matching read,
which is the state a spent-on-arm-only guard would let re-arm."""
desc = replace(_PROGRESS_DESC, sticky_seconds=0.05)
sensor, coordinator = _sensor(desc)
_replace(coordinator, state="Ready", progress="Finish")
assert sensor.native_value == "Finish"
time.sleep(0.1)
assert sensor.native_value == "Idle"
_replace(coordinator, state="Ready", progress="None")
assert sensor.native_value == "Idle"
_replace(coordinator, state="Ready", progress="Finish")
assert sensor.native_value == "Idle"
# A real cycle in between is what makes it available again.
_replace(coordinator, state="Run", progress="Drying")
assert sensor.native_value == "Drying"
_replace(coordinator, state="Run", progress="Finish")
assert sensor.native_value == "Finish"
_replace(coordinator, state="Ready", progress="None")
assert sensor.native_value == "Finish"
def test_hold_expires_after_sticky_seconds():
# A fresh desc (frozen dataclass -- replace(), not mutation, so the
# module-level _PROGRESS_DESC other tests share stays untouched) with