diff --git a/app/api/routes/api/meter_sources.py b/app/api/routes/api/meter_sources.py index 9f2fea9..746cd81 100644 --- a/app/api/routes/api/meter_sources.py +++ b/app/api/routes/api/meter_sources.py @@ -27,10 +27,21 @@ from app.services.meter_sources import ( create_source, delete_source, list_bindings, list_sources, update_binding, update_source, ) from app.services import timezone as _tz_mod +from app.services.warmtelink_worker import warmtelink_worker_manager router = APIRouter(prefix="/api/energy", tags=["api-energy-meter-sources"]) +def _reconcile_warmtelink_after_commit() -> None: + """Runtime convergence is best-effort; the already committed API result wins.""" + try: + warmtelink_worker_manager.reconcile() + except Exception: + # The manager records individual source failures itself. Do not turn a + # successful durable create/update/delete into a misleading HTTP 500. + return + + def _as_utc(value: datetime) -> datetime: if value.tzinfo is None: return value.replace(tzinfo=_tz_mod.local_tz()).astimezone(UTC) @@ -118,6 +129,7 @@ def post_source(body: MeterSourceCreate, db: Session = Depends(get_db), source = create_source(db, name=body.name, kind=body.kind, config=body.config, enabled=body.enabled) db.commit() db.refresh(source) + _reconcile_warmtelink_after_commit() return _source_response(source) except (SourceProfileError, MeterSourceError) as exc: db.rollback() @@ -138,6 +150,7 @@ def patch_source(source_uuid: str, body: MeterSourcePatch, db: Session = Depends updated = update_source(db, source.id, name=body.name, enabled=body.enabled, config_patch=body.config) db.commit() db.refresh(updated) + _reconcile_warmtelink_after_commit() return _source_response(updated) except (SourceProfileError, MeterSourceError) as exc: db.rollback() @@ -156,6 +169,7 @@ def remove_source(source_uuid: str, db: Session = Depends(get_db), try: delete_source(db, source.id) db.commit() + _reconcile_warmtelink_after_commit() return Response(status_code=status.HTTP_204_NO_CONTENT) except SourceDeleteRestrictedError as exc: db.rollback() diff --git a/app/main.py b/app/main.py index 8549fd5..f4e2b3c 100644 --- a/app/main.py +++ b/app/main.py @@ -38,6 +38,7 @@ from app.services.modbus_poll import poll_all_enabled_devices, BASE_POLL_TICK_SE from app.services.ha_discovery import publish_discovery, publish_states from app.services.tibber_prices import refresh_prices from app.services.energy_cost import compute_closed_periods +from app.services.warmtelink_worker import warmtelink_worker_manager from app.services.timezone import local_tz from scripts.app_db_adopt import AppDatabaseAdoptionError, validate_app_runtime_db @@ -273,18 +274,28 @@ async def lifespan(_: FastAPI): _startup_runtime_settings = build_runtime_settings(_startup_session, get_settings()) finally: _startup_session.close() - mqtt_manager.connect(_startup_runtime_settings) + serial_started = False + try: + mqtt_manager.connect(_startup_runtime_settings) - # DSMR sources carry their own runtime configuration and are reconciled - # after the MQTT manager is connected. - apply_dsmr_subscription() + # DSMR sources carry their own runtime configuration and are reconciled + # after the MQTT manager is connected. + apply_dsmr_subscription() + # Mark it before reconcile: a partial reconcile can already own a fd or + # a non-daemon thread and must receive the same orderly shutdown. + serial_started = True + warmtelink_worker_manager.start() - yield - - # MQTT: clean shutdown before the process exits. - mqtt_manager.disconnect() - - scheduler.shutdown(wait=False) + yield + finally: + # Serial descriptors/workers must be handled first on every exit path. + if serial_started: + try: + warmtelink_worker_manager.shutdown() + except Exception: + logger.exception("WarmteLink shutdown failed") + mqtt_manager.disconnect() + scheduler.shutdown(wait=False) def create_app() -> FastAPI: diff --git a/app/services/warmtelink_worker.py b/app/services/warmtelink_worker.py new file mode 100644 index 0000000..62cb488 --- /dev/null +++ b/app/services/warmtelink_worker.py @@ -0,0 +1,339 @@ +"""Read-only WarmteLink serial workers and their lifecycle manager. + +The worker deliberately owns no long-lived SQLAlchemy session and never +retains telegram bytes after handing a complete frame to the ingestor. +""" + +from __future__ import annotations + +from collections.abc import Callable +from contextlib import suppress +from dataclasses import dataclass +from datetime import UTC, datetime +import logging +import threading +from typing import Protocol + +import serial +from sqlalchemy import select +from sqlalchemy.orm import Session + +from app.db import get_session_local +from app.integrations.p1 import TelegramFramer +from app.models.meter_source import MeterSource +from app.services.warmtelink_ingest import WarmteLinkIngestor + +logger = logging.getLogger(__name__) + +_BACKOFF_SECONDS = (1, 2, 4, 8, 16, 32, 60) +_JOIN_TIMEOUT_SECONDS = 5 + + +class ReadOnlySerial(Protocol): + def read(self, size: int = 1) -> bytes: ... + + def close(self) -> None: ... + + +SerialFactory = Callable[[dict], ReadOnlySerial] +SessionFactory = Callable[[], Session] + + +def _default_session_factory() -> Session: + """Resolve the cached sessionmaker at call time, then open one session.""" + return get_session_local()() + + +class WorkerClock(Protocol): + """Injectable interruptible clock, keeping retry tests deterministic.""" + + def wait(self, stop_event: threading.Event, seconds: float) -> bool: ... + + +class _EventClock: + def wait(self, stop_event: threading.Event, seconds: float) -> bool: + return stop_event.wait(seconds) + + +def open_warmtelink_serial(config: dict) -> ReadOnlySerial: + """Open the fixed WarmteLink P1 profile; no write-capable API is exposed.""" + return serial.Serial( + port=config["path"], baudrate=115200, bytesize=serial.SEVENBITS, + parity=serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, timeout=1, + ) + + +@dataclass(frozen=True) +class _WorkerConfig: + source_id: int + config: dict + + +class WarmteLinkWorker: + """One interruptible, read-only serial loop for one meter source.""" + + def __init__( + self, source_id: int, config: dict, *, session_factory: SessionFactory = _default_session_factory, + serial_factory: SerialFactory = open_warmtelink_serial, + stop_event: threading.Event | None = None, + ingestor: WarmteLinkIngestor | None = None, + clock: WorkerClock | None = None, + ) -> None: + self.source_id = source_id + self.config = dict(config) + self._session_factory = session_factory + self._serial_factory = serial_factory + self._stop_event = stop_event or threading.Event() + self._ingestor = ingestor or WarmteLinkIngestor() + self._clock = clock or _EventClock() + self._serial: ReadOnlySerial | None = None + self._serial_lock = threading.Lock() + # Never inherit a daemon flag from a caller's background thread: a serial + # descriptor and its orderly shutdown must remain visible to the process. + self._thread = threading.Thread( + target=self._run, name=f"warmtelink-{source_id}", daemon=False + ) + + @property + def thread(self) -> threading.Thread: + return self._thread + + def start(self) -> None: + self._thread.start() + + def stop(self) -> None: + self._stop_event.set() + self._close_serial() + + def join(self, timeout: float = _JOIN_TIMEOUT_SECONDS) -> bool: + self._thread.join(timeout) + return not self._thread.is_alive() + + def _close_serial(self) -> None: + with self._serial_lock: + device, self._serial = self._serial, None + if device is not None: + with suppress(Exception): + device.close() + + def _record_error(self, message: str) -> None: + try: + with self._session_factory() as session: + source = session.get(MeterSource, self.source_id) + if source is not None: + source.status = "error" + source.last_error = message + source.updated_at = datetime.now(UTC) + session.commit() + except Exception: + # A source-status failure must not end another source's worker. + return + + def _run(self) -> None: + backoff_index = 0 + framer = TelegramFramer() + while not self._stop_event.is_set(): + try: + device = self._serial_factory(self.config) + with self._serial_lock: + if self._stop_event.is_set(): + with suppress(Exception): + device.close() + return + self._serial = device + while not self._stop_event.is_set(): + chunk = device.read(1024) + if not chunk: + # ``timeout`` reads are normal, but still yield so a bad + # fake/device cannot turn an empty read into a busy spin. + self._clock.wait(self._stop_event, 0.05) + continue + frames = framer.feed(chunk) + for frame in frames: + if self._stop_event.is_set(): + break + self._ingestor.handle_frame( + self.source_id, frame, session_factory=self._session_factory + ) + # A complete frame proves transport recovery even if its + # contents are rejected by the privacy/admission layer. + backoff_index = 0 + except Exception: + self._record_error("WarmteLink serial connection failed") + delay = _BACKOFF_SECONDS[min(backoff_index, len(_BACKOFF_SECONDS) - 1)] + backoff_index += 1 + self._clock.wait(self._stop_event, delay) + finally: + self._close_serial() + + +class WarmteLinkWorkerManager: + """Reconcile enabled serial sources into exactly one worker each.""" + + def __init__( + self, *, session_factory: SessionFactory = _default_session_factory, + serial_factory: SerialFactory = open_warmtelink_serial, + worker_factory: Callable[..., WarmteLinkWorker] = WarmteLinkWorker, + ) -> None: + self._session_factory = session_factory + self._serial_factory = serial_factory + self._worker_factory = worker_factory + self._workers: dict[int, tuple[_WorkerConfig, WarmteLinkWorker]] = {} + self._lock = threading.Lock() + self._reapers: set[int] = set() + self._shutting_down = False + + @property + def worker_count(self) -> int: + with self._lock: + return len(self._workers) + + def reconcile(self) -> None: + # Reading desired state under the same lock which applies it prevents a + # delayed pre-commit snapshot from rolling a newer commit backwards. + with self._lock: + if self._shutting_down: + return + desired = self._read_desired() + self._reconcile_locked(desired) + + def start(self) -> None: + """Enable reconciliation for a newly entered application lifespan.""" + with self._lock: + self._shutting_down = False + self.reconcile() + + def _read_desired(self) -> dict[int, _WorkerConfig]: + with self._session_factory() as session: + return { + source.id: _WorkerConfig(source.id, dict(source.config)) + for source in session.execute( + select(MeterSource).where( + MeterSource.kind == "warmtelink_serial", MeterSource.enabled.is_(True) + ) + ).scalars() + } + + def _record_manager_error(self, source_id: int) -> None: + """Best-effort, deliberately non-sensitive lifecycle failure status.""" + try: + with self._session_factory() as session: + source = session.get(MeterSource, source_id) + if source is not None: + source.status = "error" + source.last_error = "WarmteLink worker failed" + source.updated_at = datetime.now(UTC) + session.commit() + except Exception: + return + + def _reconcile_locked(self, desired: dict[int, _WorkerConfig]) -> None: + stale = [ + source_id for source_id, (config, _) in self._workers.items() + if source_id not in desired or desired[source_id] != config + ] + blocked: set[int] = set() + for source_id in stale: + _, worker = self._workers[source_id] + try: + worker.stop() + stopped = worker.join() + except Exception: + self._record_manager_error(source_id) + blocked.add(source_id) + continue + if stopped: + self._workers.pop(source_id, None) + else: + logger.error("WarmteLink worker did not stop for source %s", source_id) + blocked.add(source_id) + self._schedule_reaper_locked(source_id, worker) + for source_id, config in desired.items(): + if source_id in self._workers or source_id in blocked: + continue + worker: WarmteLinkWorker | None = None + try: + worker = self._worker_factory( + source_id, config.config, session_factory=self._session_factory, + serial_factory=self._serial_factory, + ) + self._workers[source_id] = (config, worker) + worker.start() + except Exception: + self._record_manager_error(source_id) + # A failed start normally has no thread. If an unusual worker + # did start before raising, keep it tracked until it is reaped. + if not self._worker_is_alive(worker): + self._workers.pop(source_id, None) + else: + self._schedule_reaper_locked(source_id, worker) + + @staticmethod + def _worker_is_alive(worker: object | None) -> bool: + thread = getattr(worker, "thread", None) + return bool(thread is not None and thread.is_alive()) + + def _schedule_reaper_locked(self, source_id: int, worker: WarmteLinkWorker) -> None: + if source_id in self._reapers: + return + self._reapers.add(source_id) + threading.Thread( + target=self._reap_worker, args=(source_id, worker), + # This bookkeeping watcher must not turn a deliberately bounded + # application shutdown into an unbounded process wait. The actual + # serial worker itself is explicitly non-daemon. + name=f"warmtelink-reaper-{source_id}", daemon=True, + ).start() + + def _reap_worker(self, source_id: int, worker: WarmteLinkWorker) -> None: + """Wait for one timed-out worker, then converge without another API call.""" + try: + while True: + try: + if worker.join(): + break + except Exception: + self._record_manager_error(source_id) + return + # A custom worker can report a bounded join timeout immediately; + # yield before asking again so its reaper cannot busy-spin. + threading.Event().wait(0.05) + with self._lock: + current = self._workers.get(source_id) + if current is not None and current[1] is worker: + self._workers.pop(source_id) + self._reapers.discard(source_id) + should_reconcile = not self._shutting_down + if should_reconcile: + self.reconcile() + finally: + with self._lock: + self._reapers.discard(source_id) + + def shutdown(self) -> None: + with self._lock: + self._shutting_down = True + workers = list(self._workers.items()) + for source_id, (_, worker) in workers: + try: + worker.stop() + except Exception: + self._record_manager_error(source_id) + for source_id, (_, worker) in workers: + try: + stopped = worker.join() + except Exception: + self._record_manager_error(source_id) + continue + if not stopped: + logger.error("WarmteLink worker did not stop during shutdown for source %s", source_id) + with self._lock: + self._schedule_reaper_locked(source_id, worker) + else: + with self._lock: + current = self._workers.get(source_id) + if current is not None and current[1] is worker: + self._workers.pop(source_id) + + +warmtelink_worker_manager = WarmteLinkWorkerManager() diff --git a/docs/design/m8-warmtelink-energy.md b/docs/design/m8-warmtelink-energy.md index 411f100..a4046fc 100644 --- a/docs/design/m8-warmtelink-energy.md +++ b/docs/design/m8-warmtelink-energy.md @@ -651,7 +651,7 @@ T01～T06 先把现有 DSMR 安全迁到统一 source/binding；T07～T11 再接 ### M8-T10 — Serial worker manager 与应用生命周期 [structural] -- **Status**: `todo` +- **Status**: `done` - **Depends**: M8-T09, M8-T06 - **Context**: 以可测试的 manager 为每个 enabled WarmteLink source 持有一个只读 worker。 diff --git a/tests/test_meter_source_api.py b/tests/test_meter_source_api.py index 2cd2148..d1159d0 100644 --- a/tests/test_meter_source_api.py +++ b/tests/test_meter_source_api.py @@ -79,6 +79,44 @@ def test_source_profiles_and_crud_mask_secrets(auth_database): engine.dispose() +def test_serial_source_reconciles_only_after_commit(auth_database, monkeypatch): + """Creation and mutation trigger the worker manager after a successful commit.""" + from app.api.routes.api import meter_sources + + calls = [] + monkeypatch.setattr(meter_sources.warmtelink_worker_manager, "reconcile", lambda: calls.append(True)) + client, engine = _client(auth_database) + with client: + _login(client) + baseline = len(calls) + response = client.post("/api/energy/sources", headers={"X-CSRF-Token": _CSRF}, json={ + "name": "Serial", "kind": "warmtelink_serial", "config": {"path": "/dev/ttyUSB0"}, + }) + assert response.status_code == 201 + source_uuid = response.json()["uuid"] + assert len(calls) == baseline + 1 + assert client.patch(f"/api/energy/sources/{source_uuid}", headers={"X-CSRF-Token": _CSRF}, json={"enabled": False}).status_code == 200 + assert len(calls) == baseline + 2 + engine.dispose() + + +def test_serial_source_commit_is_not_reported_as_failed_when_reconcile_raises(auth_database, monkeypatch): + from app.api.routes.api import meter_sources + + client, engine = _client(auth_database) + with client: + _login(client) + monkeypatch.setattr( + meter_sources.warmtelink_worker_manager, "reconcile", + lambda: (_ for _ in ()).throw(RuntimeError()), + ) + response = client.post("/api/energy/sources", headers={"X-CSRF-Token": _CSRF}, json={ + "name": "Serial", "kind": "warmtelink_serial", "config": {"path": "/dev/ttyUSB0"}, + }) + assert response.status_code == 201 + engine.dispose() + + def test_binding_routes_and_atomic_meter_declaration(auth_database): client, engine = _client(auth_database) with client: diff --git a/tests/test_warmtelink_worker.py b/tests/test_warmtelink_worker.py new file mode 100644 index 0000000..296918e --- /dev/null +++ b/tests/test_warmtelink_worker.py @@ -0,0 +1,308 @@ +"""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