119 lines
4.1 KiB
Python
119 lines
4.1 KiB
Python
"""Runnable check for the maintenance state machine. `python3 test_maintenance.py`
|
|
|
|
Walks a whole cycle in simulated time, which is the test the YAML version never
|
|
had - and it is exactly the phase transitions that a real cycle takes ten hours
|
|
to exercise once a month.
|
|
|
|
⚠️ 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, and a size-preserving
|
|
edit reverted inside one second is invisible to CPython's staleness test. That
|
|
is TEST-15, and it has already turned one live mutation into a false survivor.
|
|
Read MUTATION.md before you report a survivor.
|
|
"""
|
|
|
|
import sys
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
from app.maintenance import CHARGE, DRAIN, HOLD, IDLE, MaintConfig, Maintenance
|
|
|
|
fails = []
|
|
|
|
|
|
def check(name, cond):
|
|
print(f" {'ok ' if cond else 'FAIL'} {name}")
|
|
if not cond:
|
|
fails.append(name)
|
|
|
|
|
|
class FakeStore:
|
|
"""Same surface as Store, no disk."""
|
|
|
|
def __init__(self):
|
|
self.data = {"phase": IDLE, "phase_started": None,
|
|
"last_completed": None, "last_start_attempt": None}
|
|
|
|
def set(self, k, v):
|
|
self.data[k] = v
|
|
|
|
def set_time(self, k, when):
|
|
self.data[k] = when
|
|
|
|
def get_time(self, k):
|
|
return self.data.get(k)
|
|
|
|
def save(self):
|
|
pass
|
|
|
|
|
|
cfg = MaintConfig(enabled=True, interval_days=28, start_hour=10,
|
|
discharge_w=2500, charge_w=2500,
|
|
soc_floor=11, soc_target=99, hold_min=120)
|
|
|
|
t0 = datetime(2026, 8, 23, 10, 0, tzinfo=timezone.utc)
|
|
|
|
print("scheduling")
|
|
m = Maintenance(cfg, FakeStore())
|
|
check("never run before is due", m.due(t0))
|
|
r = m.tick(t0, soc=80)
|
|
check("starts at the start hour when due", r.phase == DRAIN and r.owns_setpoint)
|
|
check("drain commands the discharge rate", r.setpoint_w == 2500)
|
|
|
|
m2 = Maintenance(cfg, FakeStore())
|
|
r = m2.tick(t0.replace(hour=11), soc=80)
|
|
check("does not start outside the start hour", r.phase == IDLE)
|
|
|
|
m3 = Maintenance(cfg, FakeStore())
|
|
m3.store.set_time("last_completed", t0 - timedelta(days=3))
|
|
check("not due three days after a cycle", not m3.due(t0))
|
|
check("due 29 days after a cycle",
|
|
Maintenance(cfg, FakeStore()).due(t0) and
|
|
(t0 - (t0 - timedelta(days=29))) >= timedelta(days=28))
|
|
|
|
print("ownership per phase")
|
|
m = Maintenance(cfg, FakeStore())
|
|
m.force_start(t0)
|
|
r = m.tick(t0, soc=80)
|
|
check("drain: maintenance owns the setpoint", r.owns_setpoint and not r.charge_only)
|
|
r = m.tick(t0 + timedelta(hours=3), soc=10.5)
|
|
check("drain exits at the floor", r.phase == CHARGE)
|
|
check("charge: the LOOP owns the setpoint", not r.owns_setpoint and r.charge_only)
|
|
r = m.tick(t0 + timedelta(hours=20), soc=99)
|
|
check("charge exits at the target", r.phase == HOLD)
|
|
check("hold: maintenance owns and commands zero",
|
|
r.owns_setpoint and r.setpoint_w == 0.0)
|
|
r = m.tick(t0 + timedelta(hours=21), soc=100)
|
|
check("hold keeps holding before hold_min", r.phase == HOLD)
|
|
r = m.tick(t0 + timedelta(hours=22, minutes=5), soc=100)
|
|
check("hold completes after hold_min", r.phase == IDLE)
|
|
check("completion is recorded", m.store.get_time("last_completed") is not None)
|
|
check("no longer due right after completing", not m.due(t0 + timedelta(hours=22)))
|
|
|
|
print("failure handling")
|
|
m = Maintenance(cfg, FakeStore())
|
|
m.force_start(t0)
|
|
r = m.tick(t0 + timedelta(hours=13), soc=60)
|
|
check("drain aborts on timeout", r.phase == IDLE)
|
|
check("abort commands zero", r.setpoint_w == 0.0)
|
|
|
|
m = Maintenance(cfg, FakeStore())
|
|
m.force_start(t0)
|
|
r = m.tick(t0 + timedelta(minutes=5), soc=None)
|
|
check("no SoC: holds the phase rather than guessing", r.phase == DRAIN)
|
|
check("no SoC: still commands the drain rate", r.setpoint_w == 2500)
|
|
|
|
m = Maintenance(cfg, FakeStore())
|
|
m.store.set("phase", "banana")
|
|
r = m.tick(t0, soc=50)
|
|
check("unknown phase recovers to idle at 0 W", r.phase == IDLE and r.setpoint_w == 0.0)
|
|
|
|
print("disabled schedule")
|
|
m = Maintenance(MaintConfig(enabled=False), FakeStore())
|
|
check("disabled never starts", m.tick(t0, soc=80).phase == IDLE)
|
|
|
|
print()
|
|
if fails:
|
|
print(f"{len(fails)} FAILED: {', '.join(fails)}")
|
|
sys.exit(1)
|
|
print("all checks passed")
|