From 2e125dbd53b5d8cc2a431f77a2ee6569b76c11fa Mon Sep 17 00:00:00 2001 From: Tianyu Liu Date: Sun, 23 Aug 2026 00:25:11 +0200 Subject: [PATCH] M8-T04: reconcile DSMR ingest from meter sources --- app/integrations/mqtt.py | 200 +++++++++- app/main.py | 7 +- app/services/config_page.py | 25 -- app/services/dsmr_ingest.py | 523 ++++++++++++++------------ docs/design/m8-warmtelink-energy.md | 2 +- tests/test_api_config.py | 146 ++----- tests/test_dsmr_ingest.py | 205 ++++++++-- tests/test_dsmr_subscription_apply.py | 271 ++++++------- tests/test_mqtt_subscribe.py | 255 +++++++++++++ 9 files changed, 1074 insertions(+), 560 deletions(-) diff --git a/app/integrations/mqtt.py b/app/integrations/mqtt.py index 825745a..6db1316 100644 --- a/app/integrations/mqtt.py +++ b/app/integrations/mqtt.py @@ -30,6 +30,7 @@ from __future__ import annotations import logging import threading from collections.abc import Callable +from dataclasses import dataclass from typing import TYPE_CHECKING import paho.mqtt.client as mqtt @@ -50,6 +51,15 @@ MQTT_SETTINGS_KEYS = { } +@dataclass +class _SourceClientState: + """One installed source generation and its in-flight callback count.""" + + client: mqtt.Client + generation: int + in_flight: int = 0 + + def _is_configured(settings: Settings) -> bool: """Return True if MQTT is enabled *and* the broker host is set.""" return bool(settings.mqtt_enabled and settings.mqtt_broker_host) @@ -72,11 +82,26 @@ class MqttManager: def __init__(self) -> None: self._client: mqtt.Client | None = None - self._lock = threading.Lock() + self._lock = threading.RLock() + # Source replacement/removal is serialized independently from callback + # bookkeeping. In particular, paho's loop_stop() joins its network + # thread, whose callback completion also needs ``_lock``. + self._source_lifecycle_lock = threading.Lock() + self._source_idle = threading.Condition(self._lock) self._connected = False # topic → handler registry; persists across reconnects so subscriptions # are automatically re-established when the client reconnects. self._subscriptions: dict[str, Callable[[bytes], None]] = {} + # DSMR sources are independent connections: their credentials and TLS + # configuration belong to MeterSource.config, not app_config. + self._source_clients: dict[int, mqtt.Client] = {} + self._source_subscriptions: dict[int, dict[str, Callable[[bytes], None]]] = {} + self._source_connected: set[int] = set() + # Each replacement gets a distinct identity. A paho callback can run + # after its client was stopped, so source id alone is not sufficient. + self._source_generations: dict[int, int] = {} + self._source_states: dict[int, _SourceClientState] = {} + self._next_source_generation = 0 # ------------------------------------------------------------------ # Public properties @@ -115,6 +140,11 @@ class MqttManager: """ with self._lock: self._stop_client() + with self._source_lifecycle_lock: + with self._lock: + source_ids = list(self._source_clients) + for source_id in source_ids: + self._stop_source_client(source_id) def reconnect(self, settings: Settings) -> None: """Disconnect the current client (if any) and reconnect with *settings*. @@ -196,8 +226,134 @@ class MqttManager: except Exception: logger.exception("MQTT unsubscribe error (topic=%s).", topic) + def replace_source( + self, + source_id: int, + *, + host: str, + port: int, + username: str, + password: str, + tls_enabled: bool, + subscriptions: dict[str, Callable[[bytes], None]], + ) -> bool: + """Replace one source-owned client and its handlers. + + This intentionally does not touch the legacy app-wide client or any + other source client. It is also safe for a source to be temporarily + unconfigured: handlers are retained in the source registry but no + connection is attempted until a host is supplied. + """ + with self._source_lifecycle_lock: + self._stop_source_client(source_id) + if not host: + return False + self._next_source_generation += 1 + generation = self._next_source_generation + captured_subscriptions = dict(subscriptions) + client = mqtt.Client( + callback_api_version=mqtt.CallbackAPIVersion.VERSION2, + client_id=f"home-automation-dsmr-{source_id}", + ) + + def _on_connect( + connected_client: mqtt.Client, + _userdata: object, + _flags: mqtt.ConnectFlags, + reason_code: mqtt.ReasonCode, + _properties: mqtt.Properties | None, + ) -> None: + with self._lock: + if not self._is_current_source_client(source_id, generation, connected_client): + return + if reason_code.is_failure: + logger.warning("DSMR MQTT connection refused for source_id=%s", source_id) + return + self._source_connected.add(source_id) + for topic in captured_subscriptions: + try: + connected_client.subscribe(topic) + except Exception: + logger.exception("DSMR MQTT re-subscribe failed for source_id=%s", source_id) + + def _on_disconnect( + disconnected_client: mqtt.Client, + _userdata: object, + _flags: mqtt.DisconnectFlags, + _reason_code: mqtt.ReasonCode, + _properties: mqtt.Properties | None, + ) -> None: + with self._lock: + if not self._is_current_source_client(source_id, generation, disconnected_client): + return + self._source_connected.discard(source_id) + + def _on_message( + message_client: mqtt.Client, + _userdata: object, + message: mqtt.MQTTMessage, + ) -> None: + with self._lock: + if not self._is_current_source_client(source_id, generation, message_client): + return + handler = captured_subscriptions.get(message.topic) + state = self._source_states.get(source_id) + if handler is None or state is None: + return + # This permit covers the entire handler call. Teardown first + # invalidates the state and then waits for all permits, so an + # old callback cannot run after teardown returns. + state.in_flight += 1 + try: + handler(message.payload) + except Exception: + logger.exception("DSMR source handler raised (source_id=%s)", source_id) + finally: + with self._lock: + state.in_flight -= 1 + if state.in_flight == 0: + self._source_idle.notify_all() + + client.on_connect = _on_connect + client.on_disconnect = _on_disconnect + client.on_message = _on_message + if tls_enabled: + try: + client.tls_set() + except Exception: + logger.exception("DSMR MQTT TLS setup failed for source_id=%s", source_id) + return False + if username: + client.username_pw_set(username=username, password=password or None) + # Register ownership before network processing begins. A broker + # may deliver CONNACK synchronously from connect(), or on the loop + # thread before connect() returns. + with self._lock: + self._source_clients[source_id] = client + self._source_subscriptions[source_id] = captured_subscriptions + self._source_generations[source_id] = generation + self._source_states[source_id] = _SourceClientState(client, generation) + client.loop_start() + try: + client.connect(host=host, port=port, keepalive=60) + except Exception: + logger.exception("DSMR MQTT connect failed (source_id=%s, host=%s)", source_id, host) + self._stop_source_client(source_id) + return False + return True + + def remove_source(self, source_id: int) -> None: + """Drop one source client and its handlers, including queued callbacks.""" + with self._source_lifecycle_lock: + self._stop_source_client(source_id) + + def source_is_active(self, source_id: int) -> bool: + """Whether a source-owned client is currently installed for callbacks.""" + with self._lock: + return source_id in self._source_clients + # ------------------------------------------------------------------ - # Internal helpers — must be called with self._lock held + # Internal helpers # ------------------------------------------------------------------ def _start_client(self, settings: Settings) -> None: @@ -333,6 +489,46 @@ class MqttManager: logger.debug("MQTT loop_stop raised (ignoring).", exc_info=True) logger.info("MQTT client stopped.") + def _stop_source_client(self, source_id: int) -> None: + """Detach then stop a source client without blocking callback bookkeeping. + + Callers hold ``_source_lifecycle_lock``. The first phase makes the + generation unreachable while holding ``_lock``. Paho operations and + the in-flight wait are deliberately outside that lock: loop_stop() + joins paho's network thread, and an active callback needs ``_lock`` to + release its permit in ``_on_message``'s finally block. + """ + with self._lock: + state = self._source_states.pop(source_id, None) + client = self._source_clients.pop(source_id, None) + self._source_subscriptions.pop(source_id, None) + self._source_connected.discard(source_id) + # Invalidate callbacks even when there was no successfully + # installed client (for example after a failed replacement). + self._source_generations.pop(source_id, None) + if client is not None: + try: + client.disconnect() + except Exception: + logger.debug("DSMR MQTT disconnect raised (source_id=%s)", source_id, exc_info=True) + try: + client.loop_stop() + except Exception: + logger.debug("DSMR MQTT loop_stop raised (source_id=%s)", source_id, exc_info=True) + if state is not None: + with self._lock: + while state.in_flight: + self._source_idle.wait() + + def _is_current_source_client( + self, source_id: int, generation: int, client: mqtt.Client + ) -> bool: + """Check callback ownership while ``_lock`` is held.""" + return ( + self._source_generations.get(source_id) == generation + and self._source_clients.get(source_id) is client + ) + # --------------------------------------------------------------------------- # Module-level singleton — shared across lifespan and route handlers diff --git a/app/main.py b/app/main.py index 923d23d..e502d5b 100644 --- a/app/main.py +++ b/app/main.py @@ -274,10 +274,9 @@ async def lifespan(_: FastAPI): _startup_session.close() mqtt_manager.connect(_startup_runtime_settings) - # DSMR ingest: subscribe to the configured MQTT topic when enabled. The same - # applier is called from PUT /api/config, so toggling DSMR via the UI takes - # effect without an app restart. - apply_dsmr_subscription(_startup_runtime_settings) + # DSMR sources carry their own runtime configuration and are reconciled + # after the MQTT manager is connected. + apply_dsmr_subscription() yield diff --git a/app/services/config_page.py b/app/services/config_page.py index d05128b..9fd434a 100644 --- a/app/services/config_page.py +++ b/app/services/config_page.py @@ -129,27 +129,6 @@ CONFIG_FIELDS: tuple[ConfigField, ...] = ( "HA State Topic Prefix", ), ConfigField("Modbus", "MODBUS_POLLING_ENABLED", "modbus_polling_enabled", "Modbus Polling Enabled", input_type="checkbox"), - ConfigField( - "DSMR", - "DSMR_INGEST_ENABLED", - "dsmr_ingest_enabled", - "DSMR Ingest Enabled", - input_type="checkbox", - ), - ConfigField("DSMR", "DSMR_MQTT_TOPIC", "dsmr_mqtt_topic", "DSMR MQTT Topic"), - ConfigField( - "DSMR", - "DSMR_SAMPLE_INTERVAL_S", - "dsmr_sample_interval_s", - "DSMR Sample Interval (s)", - input_type="number", - ), - ConfigField( - "DSMR", - "DSMR_TARIFF_TOPIC", - "dsmr_tariff_topic", - "DSMR Tariff Topic", - ), ConfigField( "Tibber", "TIBBER_API_TOKEN", @@ -358,10 +337,6 @@ def _settings_payload(settings: Settings) -> dict[str, Any]: "ha_discovery_enabled": settings.ha_discovery_enabled, "ha_discovery_prefix": settings.ha_discovery_prefix, "ha_state_topic_prefix": settings.ha_state_topic_prefix, - "dsmr_ingest_enabled": settings.dsmr_ingest_enabled, - "dsmr_mqtt_topic": settings.dsmr_mqtt_topic, - "dsmr_sample_interval_s": settings.dsmr_sample_interval_s, - "dsmr_tariff_topic": settings.dsmr_tariff_topic, "tibber_api_token": settings.tibber_api_token, "tibber_home_id": settings.tibber_home_id, } diff --git a/app/services/dsmr_ingest.py b/app/services/dsmr_ingest.py index 63389ae..fd2489c 100644 --- a/app/services/dsmr_ingest.py +++ b/app/services/dsmr_ingest.py @@ -1,46 +1,21 @@ -"""DSMR telegram ingest service. - -Subscribes to the DSMR Reader MQTT topic (``dsmr/json``) and persists -down-sampled DSMR telegram frames to the ``dsmr_reading`` table. - -Design decisions ----------------- -- **Whole-frame storage**: the entire parsed telegram dict is stored as a JSON - blob in ``DsmrReading.payload``; no field allow-list is applied. This lets - future commodities (gas, heating, three-phase) be accommodated without a - table-schema change. -- **10-second down-sampling** (configurable via ``dsmr_sample_interval_s``): - only telegrams whose ``timestamp`` second falls on an exact multiple of the - interval are persisted. This reduces write volume from ~60 rows/min to ~6 - rows/min while guaranteeing that every 15-minute boundary (second=00) is - captured. -- **Idempotency**: the telegram's own ``id`` field is stored as ``source_id`` - with a UNIQUE constraint. A second delivery of the same telegram (e.g. after - a broker reconnect) is silently skipped. -- **Network-thread safety**: ``handle_message`` is called from paho's background - loop thread. It opens and closes its own short-lived SQLAlchemy session and - swallows all exceptions so that a buggy payload or transient DB error never - crashes the paho loop or drops the MQTT connection. -- **Numeric values kept as strings**: the DSMR Reader emits all numeric readings - as JSON strings (e.g. ``"20915.154"``). They are stored verbatim; conversion - to ``Decimal`` is deferred to the billing engine (T07) where precision matters. -- **Null phases**: some telegrams omit certain phase readings (``null`` in JSON); - these are stored as-is without special handling. -""" +"""DSMR MQTT ingest, keyed by durable meter-source identity.""" from __future__ import annotations import json import logging import threading +from dataclasses import dataclass from datetime import datetime, timezone from typing import TYPE_CHECKING import sqlalchemy.exc from sqlalchemy import select +from sqlalchemy.orm import Session from app.db import get_session_local -from app.models.energy import DsmrReading +from app.models.energy import DsmrReading, Meter +from app.models.meter_source import MeterSource, MeterSourceBinding, MeterSourceChannel if TYPE_CHECKING: from app.config import Settings @@ -48,250 +23,304 @@ if TYPE_CHECKING: logger = logging.getLogger(__name__) -# Tracks the DSMR topic currently subscribed via the MQTT manager, so a config -# change can unsubscribe the old topic before subscribing the new one. -_current_dsmr_topic: str | None = None +@dataclass(frozen=True, slots=True) +class DsmrSourceSnapshot: + """The only DSMR runtime configuration a network handler may use.""" -# Tracks the DSMR tariff topic currently subscribed via the MQTT manager. -_current_tariff_topic: str | None = None + source_id: int + topic: str + tariff_topic: str + sample_interval_s: int + broker_host: str = "" + broker_port: int = 1883 + username: str = "" + password: str = "" + tls_enabled: bool = False -# Current electricity tariff: 1 = dal/off-peak, 2 = normal/peak. -# Written by the paho network thread, read by the publish job — guarded by a lock. -_current_tariff: int | None = None + +_subscriptions: dict[int, DsmrSourceSnapshot] = {} +_subscription_lock = threading.RLock() +# A configuration value is not an ownership identity: disable and re-enable +# can produce an equal snapshot. Each installed handler therefore captures a +# fresh token and verifies object identity before it can write or update tariff. +_subscription_tokens: dict[int, object] = {} +_reconcile_lock = threading.RLock() +_tariffs: dict[int, int] = {} _tariff_lock = threading.Lock() +# Kept only for legacy direct test callers of set_current_tariff(value). Runtime +# handlers never write this value; production callers resolve a binding first. +_current_tariff: int | None = None +_UNSET = object() -def get_current_tariff() -> int | None: - """Return the most recently received electricity tariff (1 or 2), or None.""" - with _tariff_lock: +def get_current_tariff(meter_source_id: int | None = None) -> int | None: + """Return a source tariff, or resolve the active electricity binding. + + The no-argument form is retained for the pre-M8 expose integration. It + opens a short session to select the current electricity binding, so a + source's MQTT callback can never make another source's tariff current. + ``_current_tariff`` is solely a test-era fallback when no binding database + is available; runtime MQTT handlers do not update it. + """ + if meter_source_id is None: + session_local = get_session_local() + session = session_local() + try: + source_id = _current_electricity_source_id(session, datetime.now(timezone.utc)) + if source_id is not None: + return get_current_tariff(source_id) + except Exception: + logger.debug("DSMR legacy tariff lookup could not resolve a binding", exc_info=True) + finally: + session.close() return _current_tariff - - -def set_current_tariff(value: int | None) -> None: - """Set the current electricity tariff (1 or 2), or clear it with None.""" with _tariff_lock: - global _current_tariff - _current_tariff = value + return _tariffs.get(meter_source_id) -def handle_tariff_message(payload_bytes: bytes) -> None: - """Parse one DSMR tariff MQTT payload and update the in-memory tariff state. - - Called from the paho network thread; *must* swallow all exceptions so that - a bad payload never crashes the loop or drops the broker connection. - - Accepts payload as bytes or str (paho can deliver either). Ignores - whitespace. Only accepts integer values 1 or 2; anything else is discarded - and the previous known tariff is preserved. - - Parameters - ---------- - payload_bytes: - Raw bytes from the MQTT message (may also be a str in some paho versions). - """ - try: - # Decode bytes → str if needed; strip surrounding whitespace. - if isinstance(payload_bytes, (bytes, bytearray)): - raw = payload_bytes.decode("utf-8", errors="replace").strip() +def set_current_tariff(meter_source_id: int, value: int | None | object = _UNSET) -> None: + """Set or clear an individual source's tariff state.""" + global _current_tariff + if value is _UNSET: + # Compatibility with older direct callers. Do not route runtime source + # updates through this global fallback. + _current_tariff = meter_source_id if meter_source_id in (1, 2) else None + return + with _tariff_lock: + if value is None: + _tariffs.pop(meter_source_id, None) else: - raw = str(payload_bytes).strip() + _tariffs[meter_source_id] = value + +def _current_electricity_source_id(session: Session, at: datetime) -> int | None: + """Return the DSMR source bound to electricity at ``at``, if any.""" + return session.scalar( + select(MeterSource.id) + .join(MeterSourceChannel, MeterSourceChannel.source_id == MeterSource.id) + .join(MeterSourceBinding, MeterSourceBinding.channel_id == MeterSourceChannel.id) + .join(Meter, Meter.id == MeterSourceBinding.meter_id) + .where( + Meter.commodity == "electricity", + MeterSourceBinding.started_at <= at, + (MeterSourceBinding.ended_at.is_(None)) | (MeterSourceBinding.ended_at > at), + ) + .order_by(MeterSourceBinding.started_at.desc()) + .limit(1) + ) + + +def get_current_electricity_tariff(session: Session, at: datetime | None = None) -> int | None: + """Resolve tariff through the current electricity binding, never globally.""" + source_id = _current_electricity_source_id(session, at or datetime.now(timezone.utc)) + if source_id is None: + return None + return get_current_tariff(source_id) + + +def handle_tariff_message(payload_bytes: bytes, meter_source_id: int) -> None: + """Parse one source's tariff payload without raising in paho's thread.""" + try: + raw = ( + payload_bytes.decode("utf-8", errors="replace").strip() + if isinstance(payload_bytes, (bytes, bytearray)) + else str(payload_bytes).strip() + ) value = int(raw) - if value not in (1, 2): - logger.debug( - "dsmr_ingest.handle_tariff_message: unexpected tariff value %d (expected 1 or 2, ignored).", - value, - ) - return - - set_current_tariff(value) - logger.debug("dsmr_ingest.handle_tariff_message: tariff updated to %d.", value) - + if value in (1, 2): + set_current_tariff(meter_source_id, value) except Exception: - # Malformed payload (e.g. non-numeric); swallow silently to protect network thread. - logger.debug( - "dsmr_ingest.handle_tariff_message: could not parse payload %r (ignored).", - payload_bytes, - ) + logger.debug("DSMR tariff payload ignored for source_id=%s", meter_source_id) -def apply_dsmr_subscription(settings: "Settings") -> None: - """(Re)apply the DSMR MQTT subscriptions to match *settings* — restart-free. - - Call this at startup and after every config save. It makes the live MQTT - subscriptions reflect the current ``dsmr_ingest_enabled`` / ``dsmr_mqtt_topic`` - / ``dsmr_sample_interval_s`` / ``dsmr_tariff_topic`` settings without an app - restart: - - - **Disabled** → unsubscribe any active DSMR and tariff subscriptions. - - **Enabled** → (re)subscribe to ``dsmr_mqtt_topic`` with a handler bound to - a *fresh* settings snapshot, so a changed sample interval also takes effect. - Also subscribe to ``dsmr_tariff_topic`` when non-empty. - - **Topic changed** → unsubscribe the old topic before subscribing the new one. - - Idempotent and safe to call when MQTT is not connected (the subscription is - queued in the manager and established on the next connect). - """ - # Imported here (not at module top) to avoid a circular import at app start. - from app.integrations.mqtt import mqtt_manager - - global _current_dsmr_topic, _current_tariff_topic - - if not settings.dsmr_ingest_enabled: - if _current_dsmr_topic is not None: - mqtt_manager.unsubscribe(_current_dsmr_topic) - logger.info("DSMR ingest disabled — unsubscribed from topic=%s.", _current_dsmr_topic) - _current_dsmr_topic = None - if _current_tariff_topic is not None: - mqtt_manager.unsubscribe(_current_tariff_topic) - logger.info( - "DSMR ingest disabled — unsubscribed from tariff topic=%s.", - _current_tariff_topic, - ) - _current_tariff_topic = None - return - - # --- Main DSMR telegram topic --- - topic = settings.dsmr_mqtt_topic - if _current_dsmr_topic is not None and _current_dsmr_topic != topic: - mqtt_manager.unsubscribe(_current_dsmr_topic) - - # Re-subscribe (overwrites any existing handler for this topic) with a fresh - # settings snapshot so dsmr_sample_interval_s changes take effect too. - snapshot = settings - mqtt_manager.subscribe(topic, lambda payload: handle_message(payload, snapshot)) - _current_dsmr_topic = topic - logger.info("DSMR ingest enabled — subscribed to topic=%s.", topic) - - # --- DSMR tariff topic (dual-tariff slot indicator) --- - tariff_topic = settings.dsmr_tariff_topic if settings.dsmr_tariff_topic else "" - if tariff_topic: - if _current_tariff_topic is not None and _current_tariff_topic != tariff_topic: - mqtt_manager.unsubscribe(_current_tariff_topic) - mqtt_manager.subscribe(tariff_topic, lambda payload: handle_tariff_message(payload)) - _current_tariff_topic = tariff_topic - logger.info("DSMR tariff topic — subscribed to topic=%s.", tariff_topic) - else: - # tariff_topic is empty → unsubscribe any existing tariff subscription. - if _current_tariff_topic is not None: - mqtt_manager.unsubscribe(_current_tariff_topic) - logger.info( - "DSMR tariff topic cleared — unsubscribed from topic=%s.", - _current_tariff_topic, - ) - _current_tariff_topic = None +def _snapshot(source: MeterSource) -> DsmrSourceSnapshot: + config = source.config + return DsmrSourceSnapshot( + source_id=source.id, + topic=str(config.get("topic", "dsmr/json")), + tariff_topic=str(config.get("tariff_topic", "")), + sample_interval_s=int(config.get("sample_interval_s", 10)), + broker_host=str(config.get("broker_host", "")), + broker_port=int(config.get("broker_port", 1883)), + username=str(config.get("username", "")), + password=str(config.get("password", "")), + tls_enabled=bool(config.get("tls_enabled", False)), + ) -def handle_message(payload_bytes: bytes, settings: "Settings") -> None: - """Parse one DSMR MQTT payload and persist it if it passes the sample filter. - - Called from the paho network thread; *must* swallow all exceptions so that - a bad payload or transient error does not crash the loop or drop the broker - connection. - - Parameters - ---------- - payload_bytes: - Raw bytes from the MQTT message. - settings: - Runtime settings snapshot (captured at subscription time). Used for - ``dsmr_sample_interval_s``. - """ - try: - _handle_message_inner(payload_bytes, settings) - except Exception: - logger.exception("dsmr_ingest.handle_message: unexpected error (swallowed).") - - -def _handle_message_inner(payload_bytes: bytes, settings: "Settings") -> None: - """Inner implementation — may raise; caller wraps in try/except.""" - # --- 1. Parse JSON --- - try: - data: dict = json.loads(payload_bytes) - except (json.JSONDecodeError, ValueError): - logger.debug("dsmr_ingest: invalid JSON payload (skipped).") - return - - if not isinstance(data, dict): - logger.debug("dsmr_ingest: payload is not a JSON object (skipped).") - return - - # --- 2. Parse timestamp --- - raw_ts = data.get("timestamp") - if raw_ts is None: - logger.debug("dsmr_ingest: missing 'timestamp' field (skipped).") - return - - try: - # Python 3.11+ accepts the trailing 'Z' directly; for 3.10 compat we - # replace 'Z' with '+00:00' before parsing. - ts_str = raw_ts if not isinstance(raw_ts, str) else raw_ts.replace("Z", "+00:00") - ts_utc: datetime = datetime.fromisoformat(ts_str) - # Ensure it is timezone-aware UTC. - if ts_utc.tzinfo is None: - ts_utc = ts_utc.replace(tzinfo=timezone.utc) - except (ValueError, TypeError, AttributeError): - logger.debug( - "dsmr_ingest: cannot parse 'timestamp' value %r (skipped).", raw_ts - ) - return - - # --- 3. Down-sample: only persist if second falls on interval boundary --- - interval = settings.dsmr_sample_interval_s - if interval > 0 and (ts_utc.second % interval) != 0: - # This telegram is between sample points; discard silently. - return - - # --- 4. Extract source_id (telegram's own id) — stored only as a reference, - # NOT used for uniqueness/idempotency (it overflows and gets reset). --- - source_id: int | None = data.get("id") - if source_id is not None and not isinstance(source_id, int): - # Unexpected type — treat as missing rather than raising. - logger.debug( - "dsmr_ingest: 'id' field has unexpected type %s (ignoring).", - type(source_id).__name__, - ) - source_id = None - - # --- 5. Persist to database --- - # Idempotency is keyed on recorded_at (the telegram timestamp), which is - # telegram-id-independent: a single P1 meter emits one telegram per second, - # and down-sampling keeps at most one per interval-aligned second. The - # UNIQUE(recorded_at) constraint is the backstop for the IntegrityError race. +def _enabled_snapshots() -> list[DsmrSourceSnapshot]: session_local = get_session_local() session = session_local() try: - existing = session.scalar( - select(DsmrReading).where(DsmrReading.recorded_at == ts_utc) - ) - if existing is not None: - logger.debug( - "dsmr_ingest: recorded_at=%s already in DB, skipping.", - ts_utc.isoformat(), + sources = session.scalars( + select(MeterSource).where(MeterSource.kind == "dsmr_mqtt", MeterSource.enabled.is_(True)) + ).all() + return [_snapshot(source) for source in sources] + finally: + session.close() + + +def apply_dsmr_subscription(_settings: "Settings | None" = None) -> None: + """Reconcile enabled DSMR source subscriptions from the database. + + The optional, ignored settings parameter preserves the old config-route + call shape while ensuring flat DSMR settings are no longer read. + """ + from app.integrations.mqtt import mqtt_manager + + try: + desired = {snapshot.source_id: snapshot for snapshot in _enabled_snapshots()} + except Exception: + logger.exception("DSMR subscription reconcile failed while reading sources") + return + + # Do not retain _subscription_lock while stopping MQTT clients: a callback + # may currently hold it through its complete dispatch, and MqttManager + # waits for that callback before teardown returns. + with _reconcile_lock: + # Every source owns a distinct MQTT client, so equal topics from + # different sources/brokers are dispatchable. A telegram and tariff + # topic on the *same* client would overwrite one handler, however. + for source_id, snapshot in list(desired.items()): + if snapshot.tariff_topic and snapshot.topic == snapshot.tariff_topic: + logger.error("DSMR source_id=%s rejected: telegram/tariff topic collision", source_id) + desired.pop(source_id) + + with _subscription_lock: + stale_source_ids = [ + source_id + for source_id, current in _subscriptions.items() + if desired.get(source_id) != current + ] + for source_id in stale_source_ids: + _subscriptions.pop(source_id, None) + _subscription_tokens.pop(source_id, None) + set_current_tariff(source_id, None) + + for source_id in stale_source_ids: + mqtt_manager.remove_source(source_id) + + for source_id, snapshot in desired.items(): + with _subscription_lock: + current = _subscriptions.get(source_id) + if current == snapshot and mqtt_manager.source_is_active(source_id): + continue + if current is not None: + # The client went inactive outside reconcile. Invalidate its + # old token before rebuilding the same snapshot. + with _subscription_lock: + if _subscriptions.get(source_id) == current: + _subscriptions.pop(source_id, None) + _subscription_tokens.pop(source_id, None) + mqtt_manager.remove_source(source_id) + token = object() + handlers = { + snapshot.topic: lambda payload, captured=snapshot, captured_token=token: ( + handle_captured_message(payload, captured, captured_token) + ) + } + if snapshot.tariff_topic: + handlers[snapshot.tariff_topic] = ( + lambda payload, captured=snapshot, captured_token=token: ( + handle_captured_tariff_message(payload, captured, captured_token) + ) + ) + with _subscription_lock: + _subscriptions[source_id] = snapshot + _subscription_tokens[source_id] = token + applied = mqtt_manager.replace_source( + source_id, + host=snapshot.broker_host, + port=snapshot.broker_port, + username=snapshot.username, + password=snapshot.password, + tls_enabled=snapshot.tls_enabled, + subscriptions=handlers, ) + if not applied: + with _subscription_lock: + if _subscription_tokens.get(source_id) is token: + _subscriptions.pop(source_id, None) + _subscription_tokens.pop(source_id, None) + + +def handle_message(payload_bytes: bytes, snapshot: DsmrSourceSnapshot) -> None: + """Persist one down-sampled frame under its captured source identity.""" + try: + _handle_message_inner(payload_bytes, snapshot) + except Exception: + logger.exception("DSMR ingest handler failed for source_id=%s (swallowed)", snapshot.source_id) + + +def handle_captured_message( + payload_bytes: bytes, snapshot: DsmrSourceSnapshot, token: object | None = None +) -> None: + """Run a broker callback only while its exact source generation is active.""" + with _subscription_lock: + if token is not None: + if _subscription_tokens.get(snapshot.source_id) is not token: + return + elif _subscriptions.get(snapshot.source_id) != snapshot: + return + handle_message(payload_bytes, snapshot) + + +def handle_captured_tariff_message( + payload_bytes: bytes, snapshot: DsmrSourceSnapshot, token: object | None = None +) -> None: + """Ignore tariff callbacks retained from a removed/replaced source.""" + with _subscription_lock: + if token is not None: + if _subscription_tokens.get(snapshot.source_id) is not token: + return + elif _subscriptions.get(snapshot.source_id) != snapshot: return + handle_tariff_message(payload_bytes, snapshot.source_id) - reading = DsmrReading( - recorded_at=ts_utc, - source_id=source_id, - payload=data, # full frame, verbatim + +def _handle_message_inner(payload_bytes: bytes, snapshot: DsmrSourceSnapshot) -> None: + try: + data = json.loads(payload_bytes) + except (json.JSONDecodeError, ValueError): + return + if not isinstance(data, dict): + return + try: + raw_ts = data["timestamp"] + ts_utc = datetime.fromisoformat(raw_ts.replace("Z", "+00:00")) + if ts_utc.tzinfo is None: + ts_utc = ts_utc.replace(tzinfo=timezone.utc) + except (KeyError, ValueError, TypeError, AttributeError): + return + if snapshot.sample_interval_s > 0 and ts_utc.second % snapshot.sample_interval_s: + return + telegram_id = data.get("id") + if telegram_id is not None and not isinstance(telegram_id, int): + telegram_id = None + + session_local = get_session_local() + session = session_local() + try: + exists = session.scalar( + select(DsmrReading.id).where( + DsmrReading.meter_source_id == snapshot.source_id, + DsmrReading.recorded_at == ts_utc, + ) ) - session.add(reading) - session.commit() - logger.debug( - "dsmr_ingest: persisted reading recorded_at=%s source_id=%s.", - ts_utc.isoformat(), - source_id, - ) + if exists is None: + session.add( + DsmrReading( + meter_source_id=snapshot.source_id, + recorded_at=ts_utc, + telegram_id=telegram_id, + payload=data, + ) + ) + session.commit() except sqlalchemy.exc.IntegrityError: - # Race / duplicate: another insert beat us to the same recorded_at. session.rollback() - logger.debug( - "dsmr_ingest: IntegrityError for recorded_at=%s (duplicate, skipped).", - ts_utc.isoformat(), - ) except Exception: session.rollback() - logger.exception("dsmr_ingest: DB error (swallowed).") + logger.exception("DSMR database write failed for source_id=%s", snapshot.source_id) finally: session.close() diff --git a/docs/design/m8-warmtelink-energy.md b/docs/design/m8-warmtelink-energy.md index 641e619..3c95da7 100644 --- a/docs/design/m8-warmtelink-energy.md +++ b/docs/design/m8-warmtelink-energy.md @@ -409,7 +409,7 @@ T01～T06 先把现有 DSMR 安全迁到统一 source/binding；T07～T11 再接 ### M8-T04 — DSMR runtime 改为多 Source 配置 [structural] -- **Status**: `todo` +- **Status**: `done` - **Depends**: M8-T03 - **Context**: schema 回填后，DSMR subscription/ingest 应以数据库 source 为单一运行时配置来源。 diff --git a/tests/test_api_config.py b/tests/test_api_config.py index a3aed49..0ae25ee 100644 --- a/tests/test_api_config.py +++ b/tests/test_api_config.py @@ -187,15 +187,13 @@ def test_put_config_with_csrf_header_updates_app_name( assert app_name_field["value"] == "Updated via API" -def test_put_config_reapplies_dsmr_subscription( +def test_put_config_reapplies_dsmr_source_subscription( client: TestClient, test_database_urls ) -> None: - """Saving config must re-apply the DSMR subscription so enabling DSMR ingest - takes effect without an app restart (the route calls apply_dsmr_subscription - with the refreshed settings).""" + """A config save triggers source subscription reconciliation.""" _login(client) - payload = _full_config_payload({"DSMR_INGEST_ENABLED": "true"}) + payload = _full_config_payload() with patch("app.services.dsmr_ingest.apply_dsmr_subscription") as spy: response = client.put( "/api/config", @@ -205,8 +203,7 @@ def test_put_config_reapplies_dsmr_subscription( assert response.status_code == 200 spy.assert_called_once() - applied_settings = spy.call_args.args[0] - assert applied_settings.dsmr_ingest_enabled is True + assert spy.call_args.args def test_put_config_blank_secret_keeps_existing_value( @@ -594,7 +591,6 @@ EXPECTED_CHECKBOX_FIELDS = { "MQTT_TLS_ENABLED", "HA_DISCOVERY_ENABLED", "MODBUS_POLLING_ENABLED", - "DSMR_INGEST_ENABLED", } @@ -731,24 +727,15 @@ def test_put_config_mqtt_reconnect_uses_db_merged_settings( # --------------------------------------------------------------------------- -def test_get_config_includes_dsmr_section(client: TestClient) -> None: - """GET /api/config must include a DSMR section with expected fields including DSMR_TARIFF_TOPIC.""" +def test_get_config_excludes_legacy_dsmr_section(client: TestClient) -> None: + """DSMR is configured through MeterSource, never the legacy config form.""" _login(client) response = client.get("/api/config") body = response.json() section_names = {s["name"] for s in body["sections"]} - assert "DSMR" in section_names, f"DSMR section missing; got {section_names}" - - dsmr_section = next(s for s in body["sections"] if s["name"] == "DSMR") - env_names = {f["env_name"] for f in dsmr_section["fields"]} - assert "DSMR_INGEST_ENABLED" in env_names - assert "DSMR_MQTT_TOPIC" in env_names - assert "DSMR_SAMPLE_INTERVAL_S" in env_names - assert "DSMR_TARIFF_TOPIC" in env_names, ( - f"DSMR_TARIFF_TOPIC must be present in DSMR section; got {env_names}" - ) + assert "DSMR" not in section_names def test_get_config_includes_tibber_section(client: TestClient) -> None: @@ -783,36 +770,46 @@ def test_get_config_tibber_api_token_is_secret(client: TestClient) -> None: ) -def test_get_config_dsmr_ingest_enabled_is_checkbox(client: TestClient) -> None: - """DSMR_INGEST_ENABLED must have input_type='checkbox' for correct frontend rendering.""" +def test_get_config_excludes_all_legacy_dsmr_fields(client: TestClient) -> None: + """Old DSMR KV values stay in DB but are not returned by the config API.""" _login(client) response = client.get("/api/config") body = response.json() - dsmr_section = next(s for s in body["sections"] if s["name"] == "DSMR") - enabled_field = next(f for f in dsmr_section["fields"] if f["env_name"] == "DSMR_INGEST_ENABLED") - - assert enabled_field["input_type"] == "checkbox", ( - f"DSMR_INGEST_ENABLED input_type should be 'checkbox', got {enabled_field['input_type']!r}" - ) + fields = {field["env_name"] for section in body["sections"] for field in section["fields"]} + assert not fields.intersection({"DSMR_INGEST_ENABLED", "DSMR_MQTT_TOPIC", "DSMR_SAMPLE_INTERVAL_S", "DSMR_TARIFF_TOPIC"}) -def test_get_config_dsmr_sample_interval_input_type_is_number(client: TestClient) -> None: - """DSMR_SAMPLE_INTERVAL_S must have input_type='number'.""" +def test_config_save_preserves_legacy_dsmr_kv_rows(client: TestClient, test_database_urls) -> None: + """A config-only save neither reads nor deletes retired DSMR configuration.""" _login(client) + conn = sqlite3.connect(test_database_urls["app_path"]) + try: + conn.execute( + "INSERT INTO app_config (key, value, updated_at) VALUES (?, ?, CURRENT_TIMESTAMP)", + ("DSMR_MQTT_TOPIC", "legacy/topic"), + ) + conn.execute( + "INSERT INTO app_config (key, value, updated_at) VALUES (?, ?, CURRENT_TIMESTAMP)", + ("DSMR_SAMPLE_INTERVAL_S", "37"), + ) + conn.commit() + finally: + conn.close() - response = client.get("/api/config") - body = response.json() - - dsmr_section = next(s for s in body["sections"] if s["name"] == "DSMR") - interval_field = next( - f for f in dsmr_section["fields"] if f["env_name"] == "DSMR_SAMPLE_INTERVAL_S" - ) - - assert interval_field["input_type"] == "number", ( - f"DSMR_SAMPLE_INTERVAL_S input_type should be 'number', got {interval_field['input_type']!r}" + response = client.put( + "/api/config", + json={"updates": _full_config_payload({"APP_NAME": "config-only save"})}, + headers={"X-CSRF-Token": "any-non-empty-value"}, ) + assert response.status_code == 200 + conn = sqlite3.connect(test_database_urls["app_path"]) + try: + rows = dict(conn.execute("SELECT key, value FROM app_config WHERE key LIKE 'DSMR_%'")) + finally: + conn.close() + assert rows == {"DSMR_MQTT_TOPIC": "legacy/topic", "DSMR_SAMPLE_INTERVAL_S": "37"} def test_put_config_blank_tibber_api_token_keeps_existing( @@ -882,75 +879,6 @@ def test_put_config_new_tibber_api_token_overwrites_existing( assert rows.get("TIBBER_API_TOKEN") == "new-tibber-token" -def test_put_config_invalid_dsmr_sample_interval_returns_422_and_does_not_write( - client: TestClient, test_database_urls -) -> None: - """Non-integer DSMR_SAMPLE_INTERVAL_S must return 422 and not persist the bad value.""" - _login(client) - - payload = _full_config_payload({"DSMR_SAMPLE_INTERVAL_S": "not-a-number"}) - response = client.put( - "/api/config", - json={"updates": payload}, - headers={"X-CSRF-Token": "token"}, - ) - - assert response.status_code == 422 - - conn = sqlite3.connect(test_database_urls["app_path"]) - try: - rows = dict(conn.execute("SELECT key, value FROM app_config").fetchall()) - finally: - conn.close() - - assert rows.get("DSMR_SAMPLE_INTERVAL_S") != "not-a-number" - - -def test_put_config_dsmr_tariff_topic_persists_and_reflects_in_get( - client: TestClient, test_database_urls -) -> None: - """DSMR_TARIFF_TOPIC must persist via PUT and be readable via GET /api/config.""" - _login(client) - - new_topic = "meter/tariff/slot" - payload = _full_config_payload({"DSMR_TARIFF_TOPIC": new_topic}) - with patch("app.services.dsmr_ingest.apply_dsmr_subscription"): - response = client.put( - "/api/config", - json={"updates": payload}, - headers={"X-CSRF-Token": "token"}, - ) - assert response.status_code == 200 - - # The updated value must appear in the GET response. - get_resp = client.get("/api/config") - body = get_resp.json() - dsmr_section = next(s for s in body["sections"] if s["name"] == "DSMR") - tariff_field = next(f for f in dsmr_section["fields"] if f["env_name"] == "DSMR_TARIFF_TOPIC") - assert tariff_field["value"] == new_topic, ( - f"Expected DSMR_TARIFF_TOPIC to be {new_topic!r}, got {tariff_field['value']!r}" - ) - - -def test_put_config_dsmr_tariff_topic_in_settings_payload(client: TestClient) -> None: - """DSMR_TARIFF_TOPIC must appear in _settings_payload (GET /api/config returns it).""" - _login(client) - - # The default value from Settings must appear in the DSMR section. - response = client.get("/api/config") - body = response.json() - - dsmr_section = next(s for s in body["sections"] if s["name"] == "DSMR") - tariff_field = next( - (f for f in dsmr_section["fields"] if f["env_name"] == "DSMR_TARIFF_TOPIC"), None - ) - assert tariff_field is not None, "DSMR_TARIFF_TOPIC must appear in DSMR config section" - # Default value should be the DSMR reader meter-stats topic. - assert tariff_field["value"] == "dsmr/meter-stats/electricity_tariff", ( - f"Unexpected default for DSMR_TARIFF_TOPIC: {tariff_field['value']!r}" - ) - - def test_get_config_tibber_api_token_value_masked_after_save( client: TestClient, test_database_urls ) -> None: diff --git a/tests/test_dsmr_ingest.py b/tests/test_dsmr_ingest.py index 488c4a4..11fceed 100644 --- a/tests/test_dsmr_ingest.py +++ b/tests/test_dsmr_ingest.py @@ -20,8 +20,9 @@ Covers: from __future__ import annotations import json +from datetime import datetime, timezone from pathlib import Path -from unittest.mock import MagicMock, patch +from unittest.mock import patch import pytest from alembic import command @@ -63,11 +64,9 @@ def _make_settings( dsmr_ingest_enabled: bool = True, dsmr_mqtt_topic: str = "dsmr/json", ): - s = MagicMock() - s.dsmr_sample_interval_s = dsmr_sample_interval_s - s.dsmr_ingest_enabled = dsmr_ingest_enabled - s.dsmr_mqtt_topic = dsmr_mqtt_topic - return s + del dsmr_ingest_enabled + from app.services.dsmr_ingest import DsmrSourceSnapshot + return DsmrSourceSnapshot(1, dsmr_mqtt_topic, "", dsmr_sample_interval_s) # The reference telegram sample from §6.3 of the design doc. @@ -240,6 +239,20 @@ def test_different_timestamps_are_independent(dsmr_db): assert _count_readings(SessionLocal) == 2 +def test_two_sources_can_store_the_same_timestamp_independently(dsmr_db): + """The source identity, not timestamp alone, defines DSMR idempotency.""" + _, SessionLocal = dsmr_db + first = _make_settings(dsmr_sample_interval_s=10) + from app.services.dsmr_ingest import DsmrSourceSnapshot + + second = DsmrSourceSnapshot(2, "second/topic", "", 10) + _call_handle_message(_SAMPLE_TELEGRAM, first, SessionLocal) + _call_handle_message(_SAMPLE_TELEGRAM, second, SessionLocal) + + rows = _get_readings(SessionLocal) + assert {row.meter_source_id for row in rows} == {1, 2} + + def test_telegram_id_collision_does_not_drop_new_data(dsmr_db): """Regression: the telegram id overflows / gets reset to zero in DSMR firmware. Two DISTINCT telegrams (different timestamps) that happen to share the SAME @@ -259,12 +272,12 @@ def test_telegram_id_collision_does_not_drop_new_data(dsmr_db): # --------------------------------------------------------------------------- -# 4. Missing source_id — still persisted with source_id=None +# 4. Missing telegram id — still persisted with telegram_id=None # --------------------------------------------------------------------------- -def test_missing_id_persisted_with_source_id_none(dsmr_db): - """Telegram without an 'id' field must be stored with source_id=None.""" +def test_missing_id_persisted_with_telegram_id_none(dsmr_db): + """Telegram without an 'id' field must be stored with telegram_id=None.""" engine, SessionLocal = dsmr_db settings = _make_settings(dsmr_sample_interval_s=10) @@ -276,7 +289,7 @@ def test_missing_id_persisted_with_source_id_none(dsmr_db): readings = session.scalars(select(DsmrReading)).all() assert len(readings) == 1 - assert readings[0].source_id is None + assert readings[0].telegram_id is None # --------------------------------------------------------------------------- @@ -419,80 +432,87 @@ def test_timestamp_with_utc_offset_suffix(dsmr_db): @pytest.fixture() def reset_tariff(monkeypatch): - """Reset _current_tariff to None before and after each tariff test.""" from app.services import dsmr_ingest as _di + monkeypatch.setattr(_di, "_tariffs", {}) monkeypatch.setattr(_di, "_current_tariff", None) - yield - # monkeypatch auto-restores on teardown def test_tariff_message_value_2_sets_tariff(reset_tariff): """Payload b'2' must set the current tariff to 2 (normal/peak).""" from app.services.dsmr_ingest import handle_tariff_message, get_current_tariff - handle_tariff_message(b"2") - assert get_current_tariff() == 2 + handle_tariff_message(b"2", 1) + assert get_current_tariff(1) == 2 def test_tariff_message_value_1_sets_tariff(reset_tariff): """Payload b'1' must set the current tariff to 1 (dal/off-peak).""" from app.services.dsmr_ingest import handle_tariff_message, get_current_tariff - handle_tariff_message(b"1") - assert get_current_tariff() == 1 + handle_tariff_message(b"1", 1) + assert get_current_tariff(1) == 1 def test_tariff_message_updates_from_2_to_1(reset_tariff): """Subsequent payloads must overwrite the previous tariff value.""" from app.services.dsmr_ingest import handle_tariff_message, get_current_tariff - handle_tariff_message(b"2") - assert get_current_tariff() == 2 - handle_tariff_message(b"1") - assert get_current_tariff() == 1 + handle_tariff_message(b"2", 1) + assert get_current_tariff(1) == 2 + handle_tariff_message(b"1", 1) + assert get_current_tariff(1) == 1 + + +def test_tariffs_are_isolated_by_source(reset_tariff): + from app.services.dsmr_ingest import handle_tariff_message, get_current_tariff + + handle_tariff_message(b"1", 1) + handle_tariff_message(b"2", 2) + assert get_current_tariff(1) == 1 + assert get_current_tariff(2) == 2 def test_tariff_message_strips_whitespace(reset_tariff): """Payloads with surrounding whitespace (e.g. b'2\\n') must be accepted.""" from app.services.dsmr_ingest import handle_tariff_message, get_current_tariff - handle_tariff_message(b"2\n") - assert get_current_tariff() == 2 + handle_tariff_message(b"2\n", 1) + assert get_current_tariff(1) == 2 - handle_tariff_message(b" 1 ") - assert get_current_tariff() == 1 + handle_tariff_message(b" 1 ", 1) + assert get_current_tariff(1) == 1 def test_tariff_message_invalid_non_numeric_does_not_update(reset_tariff): """Non-numeric payload must not update the tariff; previous value is preserved.""" from app.services.dsmr_ingest import handle_tariff_message, get_current_tariff, set_current_tariff - set_current_tariff(2) - handle_tariff_message(b"x") + set_current_tariff(1, 2) + handle_tariff_message(b"x", 1) # Must NOT raise and must NOT change the tariff. - assert get_current_tariff() == 2 + assert get_current_tariff(1) == 2 def test_tariff_message_invalid_empty_does_not_update(reset_tariff): """Empty payload must not update the tariff; previous value is preserved.""" from app.services.dsmr_ingest import handle_tariff_message, get_current_tariff, set_current_tariff - set_current_tariff(1) - handle_tariff_message(b"") - assert get_current_tariff() == 1 + set_current_tariff(1, 1) + handle_tariff_message(b"", 1) + assert get_current_tariff(1) == 1 def test_tariff_message_out_of_range_value_does_not_update(reset_tariff): """Payload with out-of-range integer (not 1 or 2) must not update the tariff.""" from app.services.dsmr_ingest import handle_tariff_message, get_current_tariff, set_current_tariff - set_current_tariff(2) - handle_tariff_message(b"3") # 3 is not a valid tariff - assert get_current_tariff() == 2 + set_current_tariff(1, 2) + handle_tariff_message(b"3", 1) # 3 is not a valid tariff + assert get_current_tariff(1) == 2 - handle_tariff_message(b"0") # 0 is not a valid tariff - assert get_current_tariff() == 2 + handle_tariff_message(b"0", 1) # 0 is not a valid tariff + assert get_current_tariff(1) == 2 def test_tariff_message_does_not_raise_on_any_input(reset_tariff): @@ -501,4 +521,115 @@ def test_tariff_message_does_not_raise_on_any_input(reset_tariff): # All of these must complete without raising. for payload in (b"", b"x", b"99", b"\xff\xfe", b"None", b"2.0"): - handle_tariff_message(payload) # must not raise + handle_tariff_message(payload, 1) # must not raise + + +def test_electricity_tariff_resolves_current_binding_and_handoff(dsmr_db, reset_tariff): + """Runtime tariffs remain isolated and are selected through the active binding.""" + from datetime import timedelta + + from app.models.energy import Meter + from app.models.meter_source import MeterSource, MeterSourceBinding, MeterSourceChannel + from app.services.dsmr_ingest import get_current_electricity_tariff, handle_tariff_message + + _, SessionLocal = dsmr_db + now = datetime.now(timezone.utc).replace(microsecond=0) + with SessionLocal() as session: + meter = Meter( + label="electricity", + commodity="electricity", + started_at=now - timedelta(days=2), + ended_at=None, + reason="initial", + note=None, + created_at=now, + ) + first = MeterSource( + name="first", kind="dsmr_mqtt", enabled=True, config={}, created_at=now, updated_at=now + ) + second = MeterSource( + name="second", kind="dsmr_mqtt", enabled=True, config={}, created_at=now, updated_at=now + ) + session.add_all([meter, first, second]) + session.flush() + first_channel = MeterSourceChannel( + source_id=first.id, channel_key="electricity", label="first", unit="kWh", + created_at=now, updated_at=now + ) + second_channel = MeterSourceChannel( + source_id=second.id, channel_key="electricity", label="second", unit="kWh", + created_at=now, updated_at=now + ) + session.add_all([first_channel, second_channel]) + session.flush() + handoff = now - timedelta(hours=1) + session.add_all([ + MeterSourceBinding(meter_id=meter.id, channel_id=first_channel.id, started_at=now - timedelta(days=2), ended_at=handoff, created_at=now, updated_at=now), + MeterSourceBinding(meter_id=meter.id, channel_id=second_channel.id, started_at=handoff, ended_at=None, created_at=now, updated_at=now), + ]) + session.commit() + handle_tariff_message(b"1", first.id) + handle_tariff_message(b"2", second.id) + assert get_current_electricity_tariff(session, now - timedelta(days=3)) is None + assert get_current_electricity_tariff(session, handoff - timedelta(seconds=1)) == 1 + assert get_current_electricity_tariff(session, handoff) == 2 + assert get_current_electricity_tariff(session, now + timedelta(days=3)) == 2 + + +def test_legacy_getter_resolves_runtime_tariff_through_current_binding(dsmr_db, reset_tariff): + """The unchanged no-argument caller selects the bound source, not a global tariff.""" + from datetime import timedelta + + from app.models.energy import Meter + from app.models.meter_source import MeterSource, MeterSourceBinding, MeterSourceChannel + from app.services import dsmr_ingest + + _, SessionLocal = dsmr_db + now = datetime.now(timezone.utc).replace(microsecond=0) + with SessionLocal() as session: + meter = Meter( + label="electricity", commodity="electricity", started_at=now - timedelta(days=1), + ended_at=None, reason="initial", note=None, created_at=now, + ) + bound = MeterSource( + name="bound", kind="dsmr_mqtt", enabled=True, config={}, created_at=now, updated_at=now + ) + other = MeterSource( + name="other", kind="dsmr_mqtt", enabled=True, config={}, created_at=now, updated_at=now + ) + session.add_all([meter, bound, other]) + session.flush() + channel = MeterSourceChannel( + source_id=bound.id, channel_key="electricity", label="bound", unit="kWh", + created_at=now, updated_at=now, + ) + session.add(channel) + session.flush() + session.add(MeterSourceBinding( + meter_id=meter.id, channel_id=channel.id, started_at=now - timedelta(days=1), ended_at=None, + created_at=now, updated_at=now, + )) + session.commit() + dsmr_ingest.handle_tariff_message(b"1", bound.id) + dsmr_ingest.handle_tariff_message(b"2", other.id) + + with patch.object(dsmr_ingest, "get_session_local", return_value=SessionLocal): + assert dsmr_ingest.get_current_tariff() == 1 + + +@pytest.mark.parametrize("replacement", ["disable", "delete", "config-change"]) +def test_retained_source_handler_cannot_write_after_reconcile(dsmr_db, monkeypatch, replacement): + """A callback fetched before disable/delete/reconfigure is rejected before DB access.""" + from app.services import dsmr_ingest + from app.services.dsmr_ingest import DsmrSourceSnapshot + + _, SessionLocal = dsmr_db + old = DsmrSourceSnapshot(1, "old", "", 10, broker_host="one.test") + if replacement in {"disable", "delete"}: + active = {} + else: + active = {1: DsmrSourceSnapshot(1, "new", "", 10, broker_host="changed.test")} + monkeypatch.setattr(dsmr_ingest, "_subscriptions", active) + with patch.object(dsmr_ingest, "get_session_local", return_value=SessionLocal): + dsmr_ingest.handle_captured_message(_payload(_SAMPLE_TELEGRAM), old) + assert _count_readings(SessionLocal) == 0 diff --git a/tests/test_dsmr_subscription_apply.py b/tests/test_dsmr_subscription_apply.py index 6aa3323..b3defe6 100644 --- a/tests/test_dsmr_subscription_apply.py +++ b/tests/test_dsmr_subscription_apply.py @@ -1,173 +1,174 @@ -"""Tests for restart-free DSMR subscription management (apply_dsmr_subscription). - -These verify that toggling DSMR ingest / changing its topic / changing its sample -interval via the config UI is reflected in the live MQTT subscription without an -app restart. A fake MQTT manager is injected so no real broker is touched. -""" +"""DSMR source-driven MQTT subscription reconciliation tests.""" from __future__ import annotations -from unittest.mock import MagicMock +from dataclasses import replace import pytest from app.services import dsmr_ingest +from app.services.dsmr_ingest import DsmrSourceSnapshot class _FakeMqtt: def __init__(self) -> None: - self.subscribe_calls: list[tuple[str, object]] = [] - self.unsubscribe_calls: list[str] = [] + self.replace_calls: list[tuple[int, dict[str, object]]] = [] + self.remove_calls: list[int] = [] + self.handlers: dict[int, dict[str, object]] = {} - def subscribe(self, topic: str, handler) -> None: - self.subscribe_calls.append((topic, handler)) + def replace_source(self, source_id: int, **kwargs: object) -> bool: + self.replace_calls.append((source_id, kwargs)) + self.handlers[source_id] = kwargs["subscriptions"] # type: ignore[assignment] + return True - def unsubscribe(self, topic: str) -> None: - self.unsubscribe_calls.append(topic) + def remove_source(self, source_id: int) -> None: + self.remove_calls.append(source_id) + self.handlers.pop(source_id, None) + + def source_is_active(self, source_id: int) -> bool: + return source_id in self.handlers -def _settings( - *, - enabled: bool = True, - topic: str = "dsmr/json", - interval: int = 10, +def _source( + source_id: int, + topic: str, tariff_topic: str = "", -): - s = MagicMock() - s.dsmr_ingest_enabled = enabled - s.dsmr_mqtt_topic = topic - s.dsmr_sample_interval_s = interval - s.dsmr_tariff_topic = tariff_topic - return s + interval: int = 10, + **connection: object, +) -> DsmrSourceSnapshot: + return DsmrSourceSnapshot(source_id, topic, tariff_topic, interval, **connection) @pytest.fixture() def fake_mqtt(monkeypatch): fake = _FakeMqtt() - # apply_dsmr_subscription does `from app.integrations.mqtt import mqtt_manager` - # at call time, so patching the module attribute is picked up. monkeypatch.setattr("app.integrations.mqtt.mqtt_manager", fake) - # Reset (and auto-restore) the module-level "currently subscribed topics". - monkeypatch.setattr(dsmr_ingest, "_current_dsmr_topic", None) - monkeypatch.setattr(dsmr_ingest, "_current_tariff_topic", None) + monkeypatch.setattr(dsmr_ingest, "_subscriptions", {}) + monkeypatch.setattr(dsmr_ingest, "_tariffs", {}) return fake -def test_enabled_subscribes_to_topic(fake_mqtt): - dsmr_ingest.apply_dsmr_subscription(_settings(enabled=True, topic="dsmr/json")) - - assert len(fake_mqtt.subscribe_calls) == 1 - assert fake_mqtt.subscribe_calls[0][0] == "dsmr/json" - assert fake_mqtt.unsubscribe_calls == [] - assert dsmr_ingest._current_dsmr_topic == "dsmr/json" +def test_reconcile_subscribes_each_enabled_source(fake_mqtt, monkeypatch): + monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [_source(1, "a"), _source(2, "b")]) + dsmr_ingest.apply_dsmr_subscription() + assert [source_id for source_id, _ in fake_mqtt.replace_calls] == [1, 2] -def test_disabled_after_enabled_unsubscribes(fake_mqtt): - dsmr_ingest.apply_dsmr_subscription(_settings(enabled=True, topic="dsmr/json")) - dsmr_ingest.apply_dsmr_subscription(_settings(enabled=False)) - - assert fake_mqtt.unsubscribe_calls == ["dsmr/json"] - assert dsmr_ingest._current_dsmr_topic is None +def test_changed_source_replaces_only_its_subscription(fake_mqtt, monkeypatch): + monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [_source(1, "a"), _source(2, "b")]) + dsmr_ingest.apply_dsmr_subscription() + monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [_source(1, "changed"), _source(2, "b")]) + dsmr_ingest.apply_dsmr_subscription() + assert fake_mqtt.remove_calls == [1] + assert [source_id for source_id, _ in fake_mqtt.replace_calls] == [1, 2, 1] -def test_topic_change_unsubscribes_old_subscribes_new(fake_mqtt): - dsmr_ingest.apply_dsmr_subscription(_settings(topic="dsmr/json")) - dsmr_ingest.apply_dsmr_subscription(_settings(topic="meter/dsmr")) - - assert fake_mqtt.unsubscribe_calls == ["dsmr/json"] - assert [t for t, _ in fake_mqtt.subscribe_calls] == ["dsmr/json", "meter/dsmr"] - assert dsmr_ingest._current_dsmr_topic == "meter/dsmr" +def test_disable_unsubscribes_and_clears_source_tariff(fake_mqtt, monkeypatch): + monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [_source(1, "a", "tariff/a")]) + dsmr_ingest.apply_dsmr_subscription() + dsmr_ingest.set_current_tariff(1, 2) + monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: []) + dsmr_ingest.apply_dsmr_subscription() + assert fake_mqtt.remove_calls == [1] + assert dsmr_ingest.get_current_tariff(1) is None -def test_reapply_same_topic_resubscribes_fresh_handler(fake_mqtt): - # A changed sample interval must take effect — the handler is re-bound to a - # fresh settings snapshot, so re-applying the same topic re-subscribes. - dsmr_ingest.apply_dsmr_subscription(_settings(topic="dsmr/json", interval=10)) - dsmr_ingest.apply_dsmr_subscription(_settings(topic="dsmr/json", interval=20)) - - assert len(fake_mqtt.subscribe_calls) == 2 - assert fake_mqtt.unsubscribe_calls == [] # same topic, no churn - handler1 = fake_mqtt.subscribe_calls[0][1] - handler2 = fake_mqtt.subscribe_calls[1][1] - assert handler1 is not handler2 # fresh closure carrying the new settings +def test_same_snapshot_has_no_subscription_churn(fake_mqtt, monkeypatch): + monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [_source(1, "a")]) + dsmr_ingest.apply_dsmr_subscription() + dsmr_ingest.apply_dsmr_subscription() + assert len(fake_mqtt.replace_calls) == 1 + assert fake_mqtt.remove_calls == [] -def test_disabled_when_never_enabled_is_noop(fake_mqtt): - dsmr_ingest.apply_dsmr_subscription(_settings(enabled=False)) - - assert fake_mqtt.subscribe_calls == [] - assert fake_mqtt.unsubscribe_calls == [] - assert dsmr_ingest._current_dsmr_topic is None - - -# --------------------------------------------------------------------------- -# Tariff topic subscription management -# --------------------------------------------------------------------------- - - -def test_enabled_with_tariff_topic_subscribes_both(fake_mqtt): - """When enabled and tariff_topic is non-empty, both topics must be subscribed.""" - dsmr_ingest.apply_dsmr_subscription( - _settings(enabled=True, topic="dsmr/json", tariff_topic="dsmr/meter-stats/electricity_tariff") +def test_same_topic_on_different_brokers_is_allowed(fake_mqtt, monkeypatch): + monkeypatch.setattr( + dsmr_ingest, + "_enabled_snapshots", + lambda: [_source(1, "same", broker_host="one.test"), _source(2, "same", broker_host="two.test")], ) - - subscribed_topics = [t for t, _ in fake_mqtt.subscribe_calls] - assert "dsmr/json" in subscribed_topics - assert "dsmr/meter-stats/electricity_tariff" in subscribed_topics - assert len(fake_mqtt.subscribe_calls) == 2 - assert dsmr_ingest._current_tariff_topic == "dsmr/meter-stats/electricity_tariff" + dsmr_ingest.apply_dsmr_subscription() + assert [source_id for source_id, _ in fake_mqtt.replace_calls] == [1, 2] -def test_enabled_with_empty_tariff_topic_subscribes_only_main(fake_mqtt): - """When tariff_topic is empty, only the main DSMR topic is subscribed.""" - dsmr_ingest.apply_dsmr_subscription( - _settings(enabled=True, topic="dsmr/json", tariff_topic="") +@pytest.mark.parametrize( + "sources", + [ + [_source(1, "shared", "shared")], + ], +) +def test_duplicate_telegram_or_tariff_topic_is_rejected(fake_mqtt, monkeypatch, sources): + """One MQTT topic cannot safely dispatch to more than one source handler.""" + monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: sources) + dsmr_ingest.apply_dsmr_subscription() + assert fake_mqtt.replace_calls == [] + + +@pytest.mark.parametrize( + "field,value", + [ + ("broker_host", "changed.test"), + ("broker_port", 2883), + ("username", "different-user"), + ("password", "different-password"), + ("tls_enabled", True), + ("sample_interval_s", 30), + ("tariff_topic", "tariff/changed"), + ], +) +def test_each_source_config_change_replaces_only_that_source(fake_mqtt, monkeypatch, field, value): + first = _source(1, "a", "tariff/a", broker_host="one.test", username="one") + second = _source(2, "b", "tariff/b", broker_host="two.test", username="two") + monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [first, second]) + dsmr_ingest.apply_dsmr_subscription() + changed = _source(1, "a", "tariff/a", broker_host="one.test", username="one") + changed = replace(changed, **{field: value}) + monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [changed, second]) + dsmr_ingest.apply_dsmr_subscription() + assert fake_mqtt.remove_calls == [1] + assert [source_id for source_id, _ in fake_mqtt.replace_calls] == [1, 2, 1] + + +def test_failed_replace_is_not_marked_applied_and_is_retried(fake_mqtt, monkeypatch): + snapshot = _source(1, "a", broker_host="one.test") + monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [snapshot]) + original_replace = fake_mqtt.replace_source + outcomes = iter([False, True]) + + def replace_once_fails(source_id: int, **kwargs: object) -> bool: + original_replace(source_id, **kwargs) + return next(outcomes) + + fake_mqtt.replace_source = replace_once_fails # type: ignore[method-assign] + dsmr_ingest.apply_dsmr_subscription() + dsmr_ingest.apply_dsmr_subscription() + assert [source_id for source_id, _ in fake_mqtt.replace_calls] == [1, 1] + + +def test_inactive_manager_source_is_rebuilt_on_next_reconcile(fake_mqtt, monkeypatch): + snapshot = _source(1, "a", broker_host="one.test") + monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [snapshot]) + dsmr_ingest.apply_dsmr_subscription() + fake_mqtt.handlers.clear() # models MqttManager.disconnect() tearing down clients + dsmr_ingest.apply_dsmr_subscription() + assert [source_id for source_id, _ in fake_mqtt.replace_calls] == [1, 1] + + +def test_same_snapshot_aba_rejects_retained_handler(fake_mqtt, monkeypatch): + """An equal re-enabled snapshot has a fresh callback identity token.""" + snapshot = _source(1, "same", broker_host="one.test") + received: list[tuple[bytes, DsmrSourceSnapshot]] = [] + monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [snapshot]) + monkeypatch.setattr( + dsmr_ingest, "handle_message", lambda payload, captured: received.append((payload, captured)) ) + dsmr_ingest.apply_dsmr_subscription() + old_handler = fake_mqtt.handlers[1]["same"] + monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: []) + dsmr_ingest.apply_dsmr_subscription() + monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [snapshot]) + dsmr_ingest.apply_dsmr_subscription() - assert len(fake_mqtt.subscribe_calls) == 1 - assert fake_mqtt.subscribe_calls[0][0] == "dsmr/json" - assert dsmr_ingest._current_tariff_topic is None - - -def test_disabled_after_tariff_subscription_unsubscribes_both(fake_mqtt): - """Disabling ingest must also unsubscribe the tariff topic.""" - dsmr_ingest.apply_dsmr_subscription( - _settings(enabled=True, topic="dsmr/json", tariff_topic="dsmr/meter-stats/electricity_tariff") - ) - dsmr_ingest.apply_dsmr_subscription(_settings(enabled=False)) - - assert "dsmr/json" in fake_mqtt.unsubscribe_calls - assert "dsmr/meter-stats/electricity_tariff" in fake_mqtt.unsubscribe_calls - assert dsmr_ingest._current_dsmr_topic is None - assert dsmr_ingest._current_tariff_topic is None - - -def test_tariff_topic_change_resubscribes(fake_mqtt): - """Changing the tariff topic must unsubscribe the old one and subscribe the new one.""" - old_tariff = "dsmr/meter-stats/electricity_tariff" - new_tariff = "meter/tariff" - - dsmr_ingest.apply_dsmr_subscription( - _settings(enabled=True, topic="dsmr/json", tariff_topic=old_tariff) - ) - dsmr_ingest.apply_dsmr_subscription( - _settings(enabled=True, topic="dsmr/json", tariff_topic=new_tariff) - ) - - assert old_tariff in fake_mqtt.unsubscribe_calls - subscribed_topics = [t for t, _ in fake_mqtt.subscribe_calls] - assert new_tariff in subscribed_topics - assert dsmr_ingest._current_tariff_topic == new_tariff - - -def test_tariff_topic_cleared_unsubscribes(fake_mqtt): - """Setting tariff_topic to empty after it was subscribed must unsubscribe it.""" - dsmr_ingest.apply_dsmr_subscription( - _settings(enabled=True, topic="dsmr/json", tariff_topic="dsmr/meter-stats/electricity_tariff") - ) - dsmr_ingest.apply_dsmr_subscription( - _settings(enabled=True, topic="dsmr/json", tariff_topic="") - ) - - assert "dsmr/meter-stats/electricity_tariff" in fake_mqtt.unsubscribe_calls - assert dsmr_ingest._current_tariff_topic is None + old_handler(b"stale") # type: ignore[operator] + fake_mqtt.handlers[1]["same"](b"fresh") # type: ignore[operator] + assert received == [(b"fresh", snapshot)] diff --git a/tests/test_mqtt_subscribe.py b/tests/test_mqtt_subscribe.py index 42155da..b9e4afa 100644 --- a/tests/test_mqtt_subscribe.py +++ b/tests/test_mqtt_subscribe.py @@ -14,6 +14,7 @@ Covers: from __future__ import annotations +import threading from unittest.mock import MagicMock, patch from app.integrations.mqtt import MqttManager @@ -231,6 +232,260 @@ def test_on_message_does_not_crash_on_handler_exception_multiple_calls() -> None assert call_count[0] == 2 +def test_replace_source_uses_isolated_client_and_source_credentials() -> None: + """A DSMR source has its own client; replacing it leaves peers untouched.""" + manager = MqttManager() + first_client = MagicMock() + second_client = MagicMock() + third_client = MagicMock() + received: list[tuple[str, bytes]] = [] + + with patch( + "app.integrations.mqtt.mqtt.Client", side_effect=[first_client, second_client, third_client] + ): + manager.replace_source( + 1, + host="one.test", + port=1884, + username="one-user", + password="one-secret", + tls_enabled=True, + subscriptions={"one/topic": lambda payload: received.append(("one", payload))}, + ) + manager.replace_source( + 2, + host="two.test", + port=2884, + username="two-user", + password="two-secret", + tls_enabled=False, + subscriptions={"two/topic": lambda payload: received.append(("two", payload))}, + ) + manager.replace_source( + 1, + host="changed.test", + port=1885, + username="changed-user", + password="changed-secret", + tls_enabled=False, + subscriptions={"changed/topic": lambda payload: received.append(("changed", payload))}, + ) + + first_client.tls_set.assert_called_once_with() + first_client.username_pw_set.assert_called_once_with(username="one-user", password="one-secret") + first_client.connect.assert_called_once_with(host="one.test", port=1884, keepalive=60) + first_client.disconnect.assert_called_once_with() + second_client.disconnect.assert_not_called() + second_client.connect.assert_called_once_with(host="two.test", port=2884, keepalive=60) + third_client.connect.assert_called_once_with(host="changed.test", port=1885, keepalive=60) + + second_client.on_message(second_client, None, _make_mqtt_message("two/topic", b"two")) + third_client.on_message(third_client, None, _make_mqtt_message("changed/topic", b"changed")) + assert received == [("two", b"two"), ("changed", b"changed")] + + +def test_replaced_source_client_callbacks_cannot_reach_new_generation() -> None: + """A retained old paho client cannot subscribe, mutate state, or dispatch new handlers.""" + manager = MqttManager() + old_client = MagicMock() + new_client = MagicMock() + received: list[tuple[str, bytes]] = [] + with patch("app.integrations.mqtt.mqtt.Client", side_effect=[old_client, new_client]): + assert manager.replace_source( + 7, + host="old.test", + port=1883, + username="", + password="", + tls_enabled=False, + subscriptions={"same/topic": lambda payload: received.append(("old", payload))}, + ) + assert manager.replace_source( + 7, + host="new.test", + port=1883, + username="", + password="", + tls_enabled=False, + subscriptions={"same/topic": lambda payload: received.append(("new", payload))}, + ) + + accepted = MagicMock() + accepted.is_failure = False + old_client.on_connect(old_client, None, MagicMock(), accepted, None) + old_client.on_message(old_client, None, _make_mqtt_message("same/topic", b"stale")) + old_client.on_disconnect(old_client, None, MagicMock(), MagicMock(), None) + old_client.subscribe.assert_not_called() + assert received == [] + assert 7 not in manager._source_connected + + new_client.on_message(new_client, None, _make_mqtt_message("same/topic", b"fresh")) + assert received == [("new", b"fresh")] + + +def test_removed_then_reenabled_identical_source_rejects_old_callback() -> None: + manager = MqttManager() + old_client = MagicMock() + reenabled_client = MagicMock() + received: list[bytes] = [] + kwargs = { + "host": "broker.test", + "port": 1883, + "username": "", + "password": "", + "tls_enabled": False, + "subscriptions": {"same/topic": lambda payload: received.append(payload)}, + } + with patch("app.integrations.mqtt.mqtt.Client", side_effect=[old_client, reenabled_client]): + assert manager.replace_source(7, **kwargs) + manager.remove_source(7) + assert manager.replace_source(7, **kwargs) + old_client.on_message(old_client, None, _make_mqtt_message("same/topic", b"stale")) + reenabled_client.on_message(reenabled_client, None, _make_mqtt_message("same/topic", b"fresh")) + assert received == [b"fresh"] + + +def test_source_connect_failure_is_not_active_and_can_be_retried() -> None: + manager = MqttManager() + failed_client = MagicMock() + failed_client.connect.side_effect = OSError("broker down") + recovered_client = MagicMock() + with patch("app.integrations.mqtt.mqtt.Client", side_effect=[failed_client, recovered_client]): + assert not manager.replace_source( + 1, host="broker.test", port=1883, username="", password="", tls_enabled=False, + subscriptions={"topic": lambda _payload: None}, + ) + assert not manager.source_is_active(1) + assert manager.replace_source( + 1, host="broker.test", port=1883, username="", password="", tls_enabled=False, + subscriptions={"topic": lambda _payload: None}, + ) + assert manager.source_is_active(1) + failed_client.loop_stop.assert_called_once_with() + + +def test_source_tls_failure_is_not_active() -> None: + manager = MqttManager() + failed_client = MagicMock() + failed_client.tls_set.side_effect = OSError("bad TLS") + with patch("app.integrations.mqtt.mqtt.Client", return_value=failed_client): + assert not manager.replace_source( + 1, host="broker.test", port=1883, username="", password="", tls_enabled=True, + subscriptions={"topic": lambda _payload: None}, + ) + assert not manager.source_is_active(1) + + +def test_source_sync_connack_before_connect_returns_subscribes_all_topics() -> None: + """Ownership is installed before a synchronous CONNACK callback can run.""" + manager = MqttManager() + + class SyncConnackClient: + def __init__(self) -> None: + self.subscribed: list[str] = [] + + def loop_start(self) -> None: + pass + + def connect(self, **_kwargs: object) -> None: + accepted = MagicMock() + accepted.is_failure = False + self.on_connect(self, None, MagicMock(), accepted, None) + + def subscribe(self, topic: str) -> None: + self.subscribed.append(topic) + + def disconnect(self) -> None: + pass + + def loop_stop(self) -> None: + pass + + client = SyncConnackClient() + with patch("app.integrations.mqtt.mqtt.Client", return_value=client): + assert manager.replace_source( + 9, + host="broker.test", + port=1883, + username="", + password="", + tls_enabled=False, + subscriptions={"telegram/topic": lambda _payload: None, "tariff/topic": lambda _payload: None}, + ) + + assert manager.source_is_active(9) + assert client.subscribed == ["telegram/topic", "tariff/topic"] + + +def test_source_teardown_with_joining_loop_stop_waits_for_callback_before_aba() -> None: + """loop_stop may join a callback that needs the manager lock to finish.""" + manager = MqttManager() + + class JoiningClient: + def loop_start(self) -> None: + pass + + def connect(self, **_kwargs: object) -> None: + pass + + def disconnect(self) -> None: + pass + + def loop_stop(self) -> None: + self.callback_thread.join() + + old_client = JoiningClient() + new_client = MagicMock() + started = threading.Event() + release = threading.Event() + removed = threading.Event() + received: list[bytes] = [] + + def old_handler(payload: bytes) -> None: + started.set() + release.wait(timeout=2) + received.append(payload) + + kwargs = { + "host": "broker.test", + "port": 1883, + "username": "", + "password": "", + "tls_enabled": False, + "subscriptions": {"same/topic": old_handler}, + } + with patch("app.integrations.mqtt.mqtt.Client", side_effect=[old_client, new_client]): + assert manager.replace_source(7, **kwargs) + callback_thread = threading.Thread( + target=old_client.on_message, + args=(old_client, None, _make_mqtt_message("same/topic", b"old")), + daemon=True, + ) + old_client.callback_thread = callback_thread + callback_thread.start() + assert started.wait(timeout=1) + + def remove_source() -> None: + manager.remove_source(7) + removed.set() + + teardown_thread = threading.Thread(target=remove_source, daemon=True) + teardown_thread.start() + assert not removed.wait(timeout=0.05) + release.set() + assert removed.wait(timeout=1) + callback_thread.join(timeout=1) + teardown_thread.join(timeout=1) + + assert not callback_thread.is_alive() + assert not teardown_thread.is_alive() + assert removed.is_set() + with patch("app.integrations.mqtt.mqtt.Client", return_value=new_client): + assert manager.replace_source(7, **kwargs) + old_client.on_message(old_client, None, _make_mqtt_message("same/topic", b"stale")) + assert received == [b"old"] + + # --------------------------------------------------------------------------- # on_connect re-subscribes registered topics # ---------------------------------------------------------------------------