163 lines
5.2 KiB
Python
163 lines
5.2 KiB
Python
"""Runnable check for MAINT-25. `python3 test_maint25.py`
|
|
|
|
After a failed monthly cycle the add-on waits 7 days (MAINT-13, `backoff_until`).
|
|
The Home plan ignored that and drew a cycle at the start hour every night of
|
|
the wait. Through the plan builder (`status()` as in test_home.py), N=1 and N=2:
|
|
during the back-off no cycle run is drawn and a short line says when it waits
|
|
until, in English and Dutch; after 7 days the run is drawn again; a cycle
|
|
completed after the failure ends the back-off.
|
|
Read MUTATION.md before you report a survivor.
|
|
"""
|
|
|
|
import asyncio
|
|
import os
|
|
import sys
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
sys.path.insert(0, HERE)
|
|
|
|
from app import homeview # noqa: E402
|
|
from app.main import Controller # noqa: E402
|
|
|
|
fails = []
|
|
|
|
|
|
def check(name, cond):
|
|
print(f" {'ok ' if cond else 'FAIL'} {name}")
|
|
if not cond:
|
|
fails.append(name)
|
|
|
|
|
|
class _Store:
|
|
def __init__(self):
|
|
self.data = {}
|
|
|
|
def set(self, k, v):
|
|
self.data[k] = v
|
|
|
|
def get(self, k, d=None):
|
|
return self.data.get(k, d)
|
|
|
|
def set_time(self, k, v):
|
|
self.data[k] = v.isoformat()
|
|
|
|
def get_time(self, k):
|
|
v = self.data.get(k)
|
|
return None if v is None else datetime.fromisoformat(v)
|
|
|
|
|
|
class _Pub:
|
|
enabled = False
|
|
|
|
def publish(self, _v):
|
|
pass
|
|
|
|
def close(self):
|
|
pass
|
|
|
|
|
|
class _Hass:
|
|
session = None
|
|
token = ""
|
|
|
|
async def limits(self, _e):
|
|
return (-5000.0, 5000.0, 1.0)
|
|
|
|
async def number(self, _e, invert=False):
|
|
return None
|
|
|
|
async def state(self, _e):
|
|
return None
|
|
|
|
async def set_number(self, _e, _v):
|
|
return True
|
|
|
|
async def call(self, *_a):
|
|
return True
|
|
|
|
async def states(self):
|
|
return []
|
|
|
|
|
|
NOW = datetime(2026, 9, 27, 9, 40, tzinfo=timezone.utc)
|
|
ONE = {"setpoint_entity": "number.sp", "meter_entity": "sensor.grid",
|
|
"soc_entity": "sensor.soc", "batt_entity": "sensor.batt",
|
|
"meter_source": "off", "max_w": 5000, "shadow_mode": "off",
|
|
"maintenance_enabled": True, "maintenance_start_hour": 12}
|
|
TWO = dict(ONE, inverters=[
|
|
{"name": "Battery A", "setpoint_entity": "number.a", "soc_entity":
|
|
"sensor.soc_a", "batt_entity": "sensor.batt_a", "max_w": 2500,
|
|
"capacity_kwh": 9.6},
|
|
{"name": "Battery B", "setpoint_entity": "number.b", "soc_entity":
|
|
"sensor.soc_b", "batt_entity": "sensor.batt_b", "max_w": 2500,
|
|
"capacity_kwh": 5.4}])
|
|
|
|
|
|
def make(opts, failed=None, done=None):
|
|
c = Controller(dict(opts), _Hass(), _Store(), _Pub(), stop=asyncio.Event())
|
|
c.auto = True
|
|
for inv in c.inverters:
|
|
inv.soc, inv.batt, inv.last_write = 50.0, 0.0, 0.0
|
|
inv.dev_min, inv.dev_max = -5000.0, 5000.0
|
|
c.grid = 10.0
|
|
m = c.maints[0]
|
|
if failed is not None:
|
|
m.store.set_time("last_failed", failed)
|
|
if done is not None:
|
|
m.store.set_time("last_completed", done)
|
|
return c
|
|
|
|
|
|
def plan(c, lang=None):
|
|
s = c.status_payload()
|
|
s.update(homeview.build(c, level=s["level"], checks=s["checks"], now=NOW,
|
|
**({} if lang is None else {"lang": lang})))
|
|
p = s["plan"]
|
|
return [r for r in p["runs"] if r["kind"] == "cycle"], p["lines"]
|
|
|
|
|
|
D = timedelta(days=1)
|
|
print("1. a cycle failed 1 day ago (N=2: Battery A failed, B never ran)")
|
|
runs, lines = plan(make(TWO, failed=NOW - D))
|
|
print(" ", [r["from"] for r in runs], lines)
|
|
check("the run tonight is B's, not A's (A waits out the back-off)",
|
|
len(runs) == 1 and not any("Battery A waits its turn" in x for x in lines))
|
|
check("a plain line says Battery A waits until 3 October",
|
|
any("Battery A: the monthly cycle waits until 3 October." == x
|
|
for x in lines))
|
|
|
|
print("\n2. N=1")
|
|
runs1, lines1 = plan(make(ONE, failed=NOW - D))
|
|
check("no cycle run is drawn, one line says when it waits until",
|
|
runs1 == [] and any("waits until 3 October" in x for x in lines1))
|
|
runs1b, lines1b = plan(make(ONE))
|
|
check("without a failure the run is drawn (non-vacuous)",
|
|
len(runs1b) == 1 and not any("waits until" in x for x in lines1b))
|
|
|
|
print("\n3. after the 7 days, and after a cycle that completed")
|
|
runs2, lines2 = plan(make(ONE, failed=NOW - timedelta(days=7, hours=1)))
|
|
check("7 days and an hour after the failure the run is drawn again, no line",
|
|
len(runs2) == 1 and not any("waits until" in x for x in lines2))
|
|
runs3, lines3 = plan(make(ONE, failed=NOW - D, done=NOW - D / 2))
|
|
check("a cycle completed after the failure ends the back-off (the run is "
|
|
"not drawn only if it is no longer due)",
|
|
not any("waits until" in x for x in lines3))
|
|
runs3b, lines3b = plan(make(ONE, failed=NOW - 2 * D, done=NOW - 3 * D))
|
|
check("a completion BEFORE the failure does not end it",
|
|
runs3b == [] and any("waits until" in x for x in lines3b))
|
|
|
|
print("\n4. Dutch")
|
|
_, lnl = plan(make(ONE, failed=NOW - D), lang="nl")
|
|
print(" ", lnl)
|
|
check("the Dutch line says it plainly",
|
|
any("de maandelijkse cyclus wacht tot 3 oktober" in x for x in lnl))
|
|
|
|
print()
|
|
if fails:
|
|
print(f"{len(fails)} FAILED:")
|
|
for f in fails:
|
|
print(f" - {f}")
|
|
sys.exit(1)
|
|
print("all checks passed")
|