309 lines
11 KiB
Python
309 lines
11 KiB
Python
|
"""M8-T10 lifecycle tests; all serial I/O is fake and read-only."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import time
|
||
|
|
import threading
|
||
|
|
from datetime import UTC, datetime
|
||
|
|
from queue import Queue
|
||
|
|
|
||
|
|
import anyio
|
||
|
|
from fastapi import FastAPI
|
||
|
|
from sqlalchemy import create_engine
|
||
|
|
from sqlalchemy.orm import sessionmaker
|
||
|
|
|
||
|
|
from app.db import Base
|
||
|
|
from app.models.meter_source import MeterSource
|
||
|
|
from app.services.warmtelink_worker import WarmteLinkWorker, WarmteLinkWorkerManager, open_warmtelink_serial
|
||
|
|
|
||
|
|
|
||
|
|
def _session_factory(tmp_path):
|
||
|
|
engine = create_engine(f"sqlite:///{tmp_path / 'worker.db'}", connect_args={"check_same_thread": False})
|
||
|
|
Base.metadata.create_all(engine)
|
||
|
|
factory = sessionmaker(bind=engine)
|
||
|
|
now = datetime.now(UTC)
|
||
|
|
with factory() as session:
|
||
|
|
session.add_all([
|
||
|
|
MeterSource(name="one", kind="warmtelink_serial", enabled=True, config={"path": "/dev/a"}, created_at=now, updated_at=now),
|
||
|
|
MeterSource(name="two", kind="warmtelink_serial", enabled=True, config={"path": "/dev/b"}, created_at=now, updated_at=now),
|
||
|
|
])
|
||
|
|
session.commit()
|
||
|
|
return factory, engine
|
||
|
|
|
||
|
|
|
||
|
|
class _FakeWorker:
|
||
|
|
made: list["_FakeWorker"] = []
|
||
|
|
|
||
|
|
def __init__(self, source_id, config, **_kwargs):
|
||
|
|
self.source_id, self.config, self.started, self.stopped = source_id, config, False, False
|
||
|
|
self.__class__.made.append(self)
|
||
|
|
|
||
|
|
def start(self): self.started = True
|
||
|
|
def stop(self): self.stopped = True
|
||
|
|
def join(self, timeout=5): return True
|
||
|
|
|
||
|
|
|
||
|
|
def test_manager_reconciles_n_sources_without_duplicates_and_replaces_config(tmp_path):
|
||
|
|
factory, engine = _session_factory(tmp_path)
|
||
|
|
_FakeWorker.made = []
|
||
|
|
manager = WarmteLinkWorkerManager(session_factory=factory, worker_factory=_FakeWorker)
|
||
|
|
manager.reconcile()
|
||
|
|
assert manager.worker_count == 2
|
||
|
|
assert len(_FakeWorker.made) == 2
|
||
|
|
manager.reconcile()
|
||
|
|
assert len(_FakeWorker.made) == 2
|
||
|
|
with factory() as session:
|
||
|
|
one = session.query(MeterSource).filter_by(name="one").one()
|
||
|
|
one.config = {"path": "/dev/new"}
|
||
|
|
session.query(MeterSource).filter_by(name="two").one().enabled = False
|
||
|
|
session.commit()
|
||
|
|
manager.reconcile()
|
||
|
|
assert manager.worker_count == 1
|
||
|
|
assert _FakeWorker.made[0].stopped and _FakeWorker.made[1].stopped
|
||
|
|
assert len(_FakeWorker.made) == 3
|
||
|
|
manager.shutdown()
|
||
|
|
assert _FakeWorker.made[-1].stopped
|
||
|
|
engine.dispose()
|
||
|
|
|
||
|
|
|
||
|
|
class _Clock:
|
||
|
|
def __init__(self): self.delays: list[float] = []
|
||
|
|
def wait(self, event, seconds):
|
||
|
|
self.delays.append(seconds)
|
||
|
|
return event.wait(0.001)
|
||
|
|
|
||
|
|
|
||
|
|
class _BrokenSerial:
|
||
|
|
def __init__(self): self.closed = False
|
||
|
|
def read(self, _size=1): raise OSError("disconnected")
|
||
|
|
def close(self): self.closed = True
|
||
|
|
|
||
|
|
|
||
|
|
class _Ingestor:
|
||
|
|
def __init__(self): self.frames = []
|
||
|
|
def handle_frame(self, source_id, frame, *, session_factory): self.frames.append((source_id, frame))
|
||
|
|
|
||
|
|
|
||
|
|
def test_worker_retries_with_capped_backoff_and_stop_closes_descriptor():
|
||
|
|
clock = _Clock()
|
||
|
|
devices: list[_BrokenSerial] = []
|
||
|
|
worker = WarmteLinkWorker(
|
||
|
|
7, {"path": "/dev/private"}, session_factory=lambda: None,
|
||
|
|
serial_factory=lambda _config: devices.append(_BrokenSerial()) or devices[-1], clock=clock,
|
||
|
|
)
|
||
|
|
worker.start()
|
||
|
|
deadline = time.monotonic() + 1
|
||
|
|
while len(clock.delays) < 8 and time.monotonic() < deadline:
|
||
|
|
time.sleep(0.005)
|
||
|
|
worker.stop()
|
||
|
|
assert worker.join(1)
|
||
|
|
assert clock.delays[:7] == [1, 2, 4, 8, 16, 32, 60]
|
||
|
|
assert all(device.closed for device in devices)
|
||
|
|
|
||
|
|
|
||
|
|
def test_worker_frames_are_read_only_and_success_resets_backoff():
|
||
|
|
class _Device:
|
||
|
|
def __init__(self):
|
||
|
|
self.closed = False
|
||
|
|
self.calls = 0
|
||
|
|
|
||
|
|
def read(self, _size=1):
|
||
|
|
self.calls += 1
|
||
|
|
if self.calls == 1:
|
||
|
|
return b"/X\r\n!\r\n"
|
||
|
|
raise OSError("gone")
|
||
|
|
|
||
|
|
def close(self):
|
||
|
|
self.closed = True
|
||
|
|
|
||
|
|
device, clock, ingestor = _Device(), _Clock(), _Ingestor()
|
||
|
|
worker = WarmteLinkWorker(3, {"path": "/dev/secret"}, session_factory=lambda: None,
|
||
|
|
serial_factory=lambda _config: device, ingestor=ingestor, clock=clock)
|
||
|
|
worker.start()
|
||
|
|
deadline = time.monotonic() + 1
|
||
|
|
while not clock.delays and time.monotonic() < deadline:
|
||
|
|
time.sleep(0.005)
|
||
|
|
worker.stop()
|
||
|
|
assert worker.join(1)
|
||
|
|
assert ingestor.frames == [(3, b"/X\r\n!\r\n")]
|
||
|
|
assert clock.delays[0] == 1
|
||
|
|
assert not hasattr(device, "write")
|
||
|
|
assert device.closed
|
||
|
|
|
||
|
|
|
||
|
|
def test_worker_resets_backoff_after_failures_then_a_complete_frame():
|
||
|
|
class _FrameThenBreak:
|
||
|
|
def __init__(self): self.calls = 0
|
||
|
|
def read(self, _size=1):
|
||
|
|
self.calls += 1
|
||
|
|
if self.calls == 1:
|
||
|
|
return b"/X\r\n!\r\n"
|
||
|
|
raise OSError("gone")
|
||
|
|
def close(self): pass
|
||
|
|
|
||
|
|
clock, ingestor = _Clock(), _Ingestor()
|
||
|
|
devices = iter([_BrokenSerial(), _BrokenSerial(), _FrameThenBreak(), _BrokenSerial()])
|
||
|
|
worker = WarmteLinkWorker(
|
||
|
|
3, {"path": "/dev/private"}, session_factory=lambda: None,
|
||
|
|
serial_factory=lambda _config: next(devices), ingestor=ingestor, clock=clock,
|
||
|
|
)
|
||
|
|
worker.start()
|
||
|
|
deadline = time.monotonic() + 1
|
||
|
|
while len(clock.delays) < 3 and time.monotonic() < deadline:
|
||
|
|
time.sleep(0.005)
|
||
|
|
worker.stop()
|
||
|
|
assert worker.join(1)
|
||
|
|
assert clock.delays[:3] == [1, 2, 1]
|
||
|
|
|
||
|
|
|
||
|
|
def test_worker_is_non_daemon_even_when_constructed_by_a_daemon_parent():
|
||
|
|
result: Queue[bool] = Queue()
|
||
|
|
|
||
|
|
def construct():
|
||
|
|
worker = WarmteLinkWorker(8, {"path": "/dev/private"}, session_factory=lambda: None)
|
||
|
|
result.put(worker.thread.daemon)
|
||
|
|
|
||
|
|
parent = threading.Thread(target=construct, daemon=True)
|
||
|
|
parent.start()
|
||
|
|
parent.join(1)
|
||
|
|
assert result.get_nowait() is False
|
||
|
|
|
||
|
|
|
||
|
|
class _TimedOutWorker(_FakeWorker):
|
||
|
|
def __init__(self, *args, **kwargs):
|
||
|
|
super().__init__(*args, **kwargs)
|
||
|
|
self.live = True
|
||
|
|
self.join_calls = 0
|
||
|
|
|
||
|
|
def join(self, timeout=5):
|
||
|
|
self.join_calls += 1
|
||
|
|
return not self.live
|
||
|
|
|
||
|
|
|
||
|
|
def test_timed_out_worker_is_reaped_and_current_desired_config_restarts(tmp_path):
|
||
|
|
factory, engine = _session_factory(tmp_path)
|
||
|
|
_TimedOutWorker.made = []
|
||
|
|
manager = WarmteLinkWorkerManager(session_factory=factory, worker_factory=_TimedOutWorker)
|
||
|
|
manager.reconcile()
|
||
|
|
old = _TimedOutWorker.made[0]
|
||
|
|
with factory() as session:
|
||
|
|
source = session.query(MeterSource).filter_by(name="one").one()
|
||
|
|
source.config = {"path": "/dev/new"}
|
||
|
|
session.commit()
|
||
|
|
manager.reconcile()
|
||
|
|
assert len(_TimedOutWorker.made) == 2
|
||
|
|
assert old.stopped and manager.worker_count == 2
|
||
|
|
old.live = False
|
||
|
|
deadline = time.monotonic() + 1
|
||
|
|
while len(_TimedOutWorker.made) < 3 and time.monotonic() < deadline:
|
||
|
|
time.sleep(0.005)
|
||
|
|
assert len(_TimedOutWorker.made) == 3
|
||
|
|
assert _TimedOutWorker.made[-1].config == {"path": "/dev/new"}
|
||
|
|
manager.shutdown()
|
||
|
|
engine.dispose()
|
||
|
|
|
||
|
|
|
||
|
|
def test_manager_contains_lifecycle_failures_and_still_starts_other_sources(tmp_path):
|
||
|
|
factory, engine = _session_factory(tmp_path)
|
||
|
|
|
||
|
|
class _FaultyWorker(_FakeWorker):
|
||
|
|
def start(self):
|
||
|
|
if self.source_id == 1:
|
||
|
|
raise RuntimeError("private path must not escape")
|
||
|
|
super().start()
|
||
|
|
|
||
|
|
_FaultyWorker.made = []
|
||
|
|
manager = WarmteLinkWorkerManager(session_factory=factory, worker_factory=_FaultyWorker)
|
||
|
|
manager.reconcile()
|
||
|
|
assert [worker.source_id for worker in _FaultyWorker.made] == [1, 2]
|
||
|
|
assert _FaultyWorker.made[1].started
|
||
|
|
assert manager.worker_count == 1
|
||
|
|
with factory() as session:
|
||
|
|
failed = session.get(MeterSource, 1)
|
||
|
|
assert failed.status == "error"
|
||
|
|
assert failed.last_error == "WarmteLink worker failed"
|
||
|
|
manager.shutdown()
|
||
|
|
engine.dispose()
|
||
|
|
|
||
|
|
|
||
|
|
def test_late_reconcile_snapshot_cannot_undo_a_newer_committed_config(tmp_path, monkeypatch):
|
||
|
|
factory, engine = _session_factory(tmp_path)
|
||
|
|
_FakeWorker.made = []
|
||
|
|
manager = WarmteLinkWorkerManager(session_factory=factory, worker_factory=_FakeWorker)
|
||
|
|
entered, release = threading.Event(), threading.Event()
|
||
|
|
original = manager._read_desired
|
||
|
|
|
||
|
|
def delayed_read():
|
||
|
|
result = original()
|
||
|
|
if not entered.is_set():
|
||
|
|
entered.set()
|
||
|
|
assert release.wait(1)
|
||
|
|
return result
|
||
|
|
|
||
|
|
monkeypatch.setattr(manager, "_read_desired", delayed_read)
|
||
|
|
old_reconcile = threading.Thread(target=manager.reconcile)
|
||
|
|
old_reconcile.start()
|
||
|
|
assert entered.wait(1)
|
||
|
|
with factory() as session:
|
||
|
|
session.query(MeterSource).filter_by(name="one").one().config = {"path": "/dev/new"}
|
||
|
|
session.commit()
|
||
|
|
new_reconcile = threading.Thread(target=manager.reconcile)
|
||
|
|
new_reconcile.start()
|
||
|
|
release.set()
|
||
|
|
old_reconcile.join(1)
|
||
|
|
new_reconcile.join(1)
|
||
|
|
assert _FakeWorker.made[-1].config == {"path": "/dev/new"}
|
||
|
|
manager.shutdown()
|
||
|
|
engine.dispose()
|
||
|
|
|
||
|
|
|
||
|
|
def test_lifespan_exception_still_shuts_serial_before_other_runtime_teardown(monkeypatch):
|
||
|
|
import app.main as main
|
||
|
|
|
||
|
|
events = []
|
||
|
|
|
||
|
|
class _Scheduler:
|
||
|
|
def __init__(self, **_kwargs): pass
|
||
|
|
def add_job(self, *_args, **_kwargs): pass
|
||
|
|
def start(self): events.append("scheduler-start")
|
||
|
|
def shutdown(self, **_kwargs): events.append("scheduler-stop")
|
||
|
|
|
||
|
|
class _Session:
|
||
|
|
def close(self): events.append("session-close")
|
||
|
|
|
||
|
|
class _Manager:
|
||
|
|
def start(self): events.append("serial-start")
|
||
|
|
def shutdown(self): events.append("serial-stop-close-join")
|
||
|
|
|
||
|
|
monkeypatch.setattr(main, "ensure_runtime_dirs", lambda: None)
|
||
|
|
monkeypatch.setattr(main, "ensure_auth_db_ready", lambda: None)
|
||
|
|
monkeypatch.setattr(main, "BackgroundScheduler", _Scheduler)
|
||
|
|
monkeypatch.setattr(main, "get_session_local", lambda: _Session)
|
||
|
|
monkeypatch.setattr(main, "build_runtime_settings", lambda *_args: object())
|
||
|
|
monkeypatch.setattr(main, "apply_dsmr_subscription", lambda: None)
|
||
|
|
monkeypatch.setattr(main.mqtt_manager, "connect", lambda _settings: events.append("mqtt-start"))
|
||
|
|
monkeypatch.setattr(main.mqtt_manager, "disconnect", lambda: events.append("mqtt-stop"))
|
||
|
|
monkeypatch.setattr(main, "warmtelink_worker_manager", _Manager())
|
||
|
|
|
||
|
|
async def exercise():
|
||
|
|
try:
|
||
|
|
async with main.lifespan(FastAPI()):
|
||
|
|
raise RuntimeError("body failed")
|
||
|
|
except RuntimeError:
|
||
|
|
pass
|
||
|
|
|
||
|
|
anyio.run(exercise)
|
||
|
|
assert events.index("serial-stop-close-join") < events.index("mqtt-stop") < events.index("scheduler-stop")
|
||
|
|
|
||
|
|
|
||
|
|
def test_production_factory_uses_fixed_115200_7n1(monkeypatch):
|
||
|
|
calls = {}
|
||
|
|
class _Serial:
|
||
|
|
def __init__(self, **kwargs): calls.update(kwargs)
|
||
|
|
def read(self, _size=1): return b""
|
||
|
|
def close(self): pass
|
||
|
|
monkeypatch.setattr("app.services.warmtelink_worker.serial.Serial", _Serial)
|
||
|
|
open_warmtelink_serial({"path": "/dev/ttyUSB0", "baudrate": 1})
|
||
|
|
assert calls["port"] == "/dev/ttyUSB0"
|
||
|
|
assert calls["baudrate"] == 115200
|