From ef48032adbcb520d82b5a3291faade76b8d6cc7d Mon Sep 17 00:00:00 2001 From: Tianyu Liu Date: Tue, 23 Jun 2026 21:52:57 +0200 Subject: [PATCH] M6-T06: extend MqttManager with subscribe + DSMR ingest - mqtt.py: subscription registry, on_message dispatch (swallows handler errors), re-subscribe on (re)connect; publish/connect/disconnect/reconnect unchanged. - app/services/dsmr_ingest.py: handle_message stores the full telegram frame (no allowlist; gas/phases/null preserved), 10s downsample by timestamp.second, source_id idempotency (select + IntegrityError rollback). Net-thread safe. - main.py registers the DSMR subscription only when dsmr_ingest_enabled. --- app/integrations/mqtt.py | 63 ++++ app/main.py | 17 + app/services/dsmr_ingest.py | 159 ++++++++++ docs/design/m6-tibber-dynamic-energy.md | 2 +- tests/test_dsmr_ingest.py | 394 ++++++++++++++++++++++++ tests/test_mqtt_subscribe.py | 286 +++++++++++++++++ 6 files changed, 920 insertions(+), 1 deletion(-) create mode 100644 app/services/dsmr_ingest.py create mode 100644 tests/test_dsmr_ingest.py create mode 100644 tests/test_mqtt_subscribe.py diff --git a/app/integrations/mqtt.py b/app/integrations/mqtt.py index e58d3ab..405c5ed 100644 --- a/app/integrations/mqtt.py +++ b/app/integrations/mqtt.py @@ -18,12 +18,18 @@ Key design decisions - **Reconnect**: ``reconnect(settings)`` tears down the old client and establishes a fresh connection with the new settings. Callers (e.g. PUT /api/config) call this after saving MQTT-related config values. +- **Subscribe**: ``subscribe(topic, handler)`` registers a topic → handler + mapping. Subscriptions are re-established automatically on reconnect. + The ``on_message`` callback dispatches incoming payloads to the registered + handler; handler exceptions are caught and logged so they never crash the + paho loop thread or the network connection. """ from __future__ import annotations import logging import threading +from collections.abc import Callable from typing import TYPE_CHECKING import paho.mqtt.client as mqtt @@ -68,6 +74,9 @@ class MqttManager: self._client: mqtt.Client | None = None self._lock = threading.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]] = {} # ------------------------------------------------------------------ # Public properties @@ -142,6 +151,30 @@ class MqttManager: except Exception: logger.exception("MQTT publish error (topic=%s).", topic) + def subscribe(self, topic: str, handler: Callable[[bytes], None]) -> None: + """Register *handler* to be called when a message arrives on *topic*. + + If the client is already connected, the subscription is sent to the + broker immediately. Otherwise it is queued and will be established + the next time ``_on_connect`` fires (including after a reconnect). + + Handler exceptions are swallowed by the ``on_message`` dispatcher so + that a buggy handler can never crash the paho loop thread. + """ + self._subscriptions[topic] = handler + # Grab the client reference outside the lock to avoid holding the lock + # while calling back into paho (which may itself acquire internal locks). + with self._lock: + client = self._client + connected = self._connected + + if client is not None and connected: + try: + client.subscribe(topic) + logger.debug("MQTT subscribed to topic=%s (immediate).", topic) + except Exception: + logger.exception("MQTT subscribe error (topic=%s).", topic) + # ------------------------------------------------------------------ # Internal helpers — must be called with self._lock held # ------------------------------------------------------------------ @@ -168,6 +201,15 @@ class MqttManager: else: logger.info("MQTT connected (reason_code=%s).", reason_code) self._connected = True + # Re-establish all registered subscriptions after (re-)connect. + for sub_topic in self._subscriptions: + try: + _client.subscribe(sub_topic) + logger.debug("MQTT re-subscribed to topic=%s.", sub_topic) + except Exception: + logger.exception( + "MQTT re-subscribe failed (topic=%s).", sub_topic + ) # VERSION2 on_disconnect signature: # (client, userdata, disconnect_flags, reason_code, properties) @@ -184,8 +226,29 @@ class MqttManager: else: logger.info("MQTT disconnected cleanly.") + # VERSION2 on_message signature: (client, userdata, message) + def _on_message( + _client: mqtt.Client, + _userdata: object, + message: mqtt.MQTTMessage, + ) -> None: + handler = self._subscriptions.get(message.topic) + if handler is None: + logger.debug( + "MQTT on_message: no handler for topic=%s.", message.topic + ) + return + try: + handler(message.payload) + except Exception: + logger.exception( + "MQTT message handler raised for topic=%s (swallowed).", + message.topic, + ) + client.on_connect = _on_connect client.on_disconnect = _on_disconnect + client.on_message = _on_message # TLS if settings.mqtt_tls_enabled: diff --git a/app/main.py b/app/main.py index 82aa9ce..a00a5c0 100644 --- a/app/main.py +++ b/app/main.py @@ -28,6 +28,7 @@ from app.config import get_settings from app.integrations.mqtt import mqtt_manager from app.services.auth import AuthBootstrapError, initialize_auth_schema from app.services.config_page import build_runtime_settings, seed_missing_config_from_bootstrap, sync_app_hostname_from_bootstrap +from app.services.dsmr_ingest import handle_message as dsmr_handle_message from app.services.public_ip import check_public_ipv4_and_notify from app.services.modbus_poll import poll_all_enabled_devices, BASE_POLL_TICK_SECONDS from app.services.ha_discovery import publish_discovery, publish_states @@ -201,6 +202,22 @@ async def lifespan(_: FastAPI): _startup_session.close() mqtt_manager.connect(_startup_runtime_settings) + # DSMR ingest: subscribe to the configured MQTT topic when enabled. + # The handler captures a snapshot of the current runtime settings (for the + # dsmr_sample_interval_s value). Runtime setting changes take effect after + # a server restart or an explicit reconnect — consistent with other MQTT + # configuration in this project. + if _startup_runtime_settings.dsmr_ingest_enabled: + _dsmr_topic = _startup_runtime_settings.dsmr_mqtt_topic + _dsmr_settings_snapshot = _startup_runtime_settings + mqtt_manager.subscribe( + _dsmr_topic, + lambda payload: dsmr_handle_message(payload, _dsmr_settings_snapshot), + ) + logger.info("DSMR ingest enabled — subscribed to topic=%s.", _dsmr_topic) + else: + logger.debug("DSMR ingest disabled — not subscribing to DSMR topic.") + yield # MQTT: clean shutdown before the process exits. diff --git a/app/services/dsmr_ingest.py b/app/services/dsmr_ingest.py new file mode 100644 index 0000000..295d09f --- /dev/null +++ b/app/services/dsmr_ingest.py @@ -0,0 +1,159 @@ +"""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. +""" + +from __future__ import annotations + +import json +import logging +from datetime import datetime, timezone +from typing import TYPE_CHECKING + +import sqlalchemy.exc +from sqlalchemy import select + +from app.db import get_session_local +from app.models.energy import DsmrReading + +if TYPE_CHECKING: + from app.config import Settings + +logger = logging.getLogger(__name__) + + +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 field, for idempotency) --- + 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 --- + session_local = get_session_local() + session = session_local() + try: + # Idempotency check: if source_id is known and already exists, skip. + if source_id is not None: + existing = session.scalar( + select(DsmrReading).where(DsmrReading.source_id == source_id) + ) + if existing is not None: + logger.debug( + "dsmr_ingest: source_id=%d already in DB, skipping.", source_id + ) + return + + reading = DsmrReading( + recorded_at=ts_utc, + source_id=source_id, + payload=data, # full frame, verbatim + ) + session.add(reading) + session.commit() + logger.debug( + "dsmr_ingest: persisted reading recorded_at=%s source_id=%s.", + ts_utc.isoformat(), + source_id, + ) + except sqlalchemy.exc.IntegrityError: + # Race condition: another handler inserted the same source_id between our + # check and our insert. Roll back and ignore. + session.rollback() + logger.debug( + "dsmr_ingest: IntegrityError for source_id=%s (duplicate, skipped).", + source_id, + ) + except Exception: + session.rollback() + logger.exception("dsmr_ingest: DB error (swallowed).") + finally: + session.close() diff --git a/docs/design/m6-tibber-dynamic-energy.md b/docs/design/m6-tibber-dynamic-energy.md index 2616a17..c5d9d0f 100644 --- a/docs/design/m6-tibber-dynamic-energy.md +++ b/docs/design/m6-tibber-dynamic-energy.md @@ -362,7 +362,7 @@ Phase D（API + 前端） - **Reviewer checklist**: token 不进日志；httpx 超时；upsert 幂等；时间存 UTC；无对 `tibber_price` 的破坏性批删。 ### M6-T06 — MqttManager 扩订阅 + DSMR ingest -- **Status**: `todo` · **Depends**: M6-T01, M6-T02 +- **Status**: `done` · **Depends**: M6-T01, M6-T02 - **Context**: 给只发不收的 `MqttManager` 扩订阅；DSMR ingest 整帧 blob + 10s 降采样落库。 - **Files**: `modify app/integrations/mqtt.py`（订阅注册 + on_message 分发）；`create app/services/dsmr_ingest.py`；`modify app/main.py`（启用时注册订阅）；`create tests/test_mqtt_subscribe.py`、`tests/test_dsmr_ingest.py` - **Steps**: diff --git a/tests/test_dsmr_ingest.py b/tests/test_dsmr_ingest.py new file mode 100644 index 0000000..c970ba5 --- /dev/null +++ b/tests/test_dsmr_ingest.py @@ -0,0 +1,394 @@ +"""Tests for M6-T06: DSMR ingest service (handle_message). + +All tests use a temporary SQLite database (upgraded to the full Alembic head) +so that the real DsmrReading model/constraints are exercised. + +Covers: +- Sample telegram with second=00 (on interval) is persisted as a full-frame blob. +- Sample telegram with second=48 (off interval, 10s default) is discarded. +- Telegram with second=00 → all fields including gas (extra_device_*) and null + phases are stored verbatim. +- Duplicate source_id → not re-inserted (idempotency). +- source_id absent → row still inserted with source_id=None. +- dsmr_sample_interval_s=0 → no ZeroDivisionError; every telegram is persisted. +- Invalid JSON → silently discarded (no exception propagated). +- Missing 'timestamp' field → silently discarded. +- Unparseable 'timestamp' → silently discarded. +- Handler exception (e.g. bad DB) → does not propagate out of handle_message. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest +from alembic import command +from alembic.config import Config +from sqlalchemy import create_engine, select +from sqlalchemy.orm import Session + +from app.models.energy import DsmrReading + + +# --------------------------------------------------------------------------- +# Helpers / fixtures +# --------------------------------------------------------------------------- + + +def _alembic_config(database_url: str) -> Config: + cfg = Config("alembic_app.ini") + cfg.set_main_option("sqlalchemy.url", database_url) + return cfg + + +@pytest.fixture() +def dsmr_db(tmp_path: Path): + """Temporary SQLite DB upgraded to the Alembic head; yields (engine, session_factory).""" + db_path = tmp_path / "dsmr_ingest_test.db" + db_url = f"sqlite:///{db_path}" + command.upgrade(_alembic_config(db_url), "head") + engine = create_engine(db_url, connect_args={"check_same_thread": False}) + from sqlalchemy.orm import sessionmaker + + SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, class_=Session) + yield engine, SessionLocal + engine.dispose() + + +def _make_settings( + *, + dsmr_sample_interval_s: int = 10, + 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 + + +# The reference telegram sample from §6.3 of the design doc. +# Numeric values are JSON strings; missing phases are null. +_SAMPLE_TELEGRAM = { + "id": 200086230, + "timestamp": "2026-06-23T12:16:00Z", # second=00, on 10s boundary + "electricity_delivered_1": "20915.154", + "electricity_returned_1": "2979.905", + "electricity_delivered_2": "15212.090", + "electricity_returned_2": "6786.406", + "electricity_currently_delivered": "0.000", + "phase_currently_delivered_l1": "0.000", + "phase_currently_delivered_l2": None, # absent phase + "extra_device_timestamp": "2026-06-23T12:15:00Z", + "extra_device_delivered": "6208.234", + "phase_voltage_l1": "237.0", + "phase_voltage_l2": None, # absent phase +} + +# Telegram with second=48 (not divisible by 10) — should be discarded. +_SAMPLE_TELEGRAM_SECOND_48 = { + **_SAMPLE_TELEGRAM, + "id": 200086248, + "timestamp": "2026-06-23T12:16:48Z", # second=48 +} + +# Telegram with second=10 — another valid boundary. +_SAMPLE_TELEGRAM_SECOND_10 = { + **_SAMPLE_TELEGRAM, + "id": 200086210, + "timestamp": "2026-06-23T12:16:10Z", # second=10 +} + + +def _payload(data: dict) -> bytes: + return json.dumps(data).encode() + + +# --------------------------------------------------------------------------- +# Helper: invoke handle_message with the test DB injected via patch +# --------------------------------------------------------------------------- + + +def _call_handle_message( + data: dict, + settings, + SessionLocal, +) -> None: + """Call dsmr_ingest.handle_message with the test SessionLocal patched in.""" + from app.services import dsmr_ingest + + with patch.object(dsmr_ingest, "get_session_local", return_value=SessionLocal): + dsmr_ingest.handle_message(_payload(data), settings) + + +def _count_readings(SessionLocal) -> int: + with SessionLocal() as session: + return session.scalar( + __import__("sqlalchemy", fromlist=["func"]).func.count(DsmrReading.id) + ) + + +def _get_readings(SessionLocal) -> list[DsmrReading]: + with SessionLocal() as session: + return session.scalars(select(DsmrReading)).all() + + +# --------------------------------------------------------------------------- +# 1. On-interval telegram is persisted (whole frame) +# --------------------------------------------------------------------------- + + +def test_second_00_persists_full_frame(dsmr_db): + """A telegram with second=00 (10s boundary) must be persisted with the full payload.""" + engine, SessionLocal = dsmr_db + settings = _make_settings(dsmr_sample_interval_s=10) + + _call_handle_message(_SAMPLE_TELEGRAM, settings, SessionLocal) + + with Session(engine) as session: + readings = session.scalars(select(DsmrReading)).all() + + assert len(readings) == 1 + row = readings[0] + + # Recorded_at must be parsed from the telegram timestamp. + assert row.recorded_at is not None + assert row.recorded_at.second == 0 + + # source_id must equal the telegram id. + assert row.source_id == _SAMPLE_TELEGRAM["id"] + + # Full frame preserved — spot-check several fields. + payload = row.payload + assert payload["electricity_delivered_1"] == "20915.154" + assert payload["electricity_returned_2"] == "6786.406" + assert payload["extra_device_delivered"] == "6208.234" + assert payload["extra_device_timestamp"] == "2026-06-23T12:15:00Z" + + # Null phases stored verbatim (not omitted, not converted). + assert "phase_currently_delivered_l2" in payload + assert payload["phase_currently_delivered_l2"] is None + assert "phase_voltage_l2" in payload + assert payload["phase_voltage_l2"] is None + + +def test_second_10_persists(dsmr_db): + """A telegram with second=10 (another 10s boundary) must also be persisted.""" + _, SessionLocal = dsmr_db + settings = _make_settings(dsmr_sample_interval_s=10) + + _call_handle_message(_SAMPLE_TELEGRAM_SECOND_10, settings, SessionLocal) + + assert _count_readings(SessionLocal) == 1 + + +# --------------------------------------------------------------------------- +# 2. Off-interval telegram is discarded +# --------------------------------------------------------------------------- + + +def test_second_48_discarded(dsmr_db): + """A telegram with second=48 (not on 10s boundary) must be silently discarded.""" + _, SessionLocal = dsmr_db + settings = _make_settings(dsmr_sample_interval_s=10) + + _call_handle_message(_SAMPLE_TELEGRAM_SECOND_48, settings, SessionLocal) + + assert _count_readings(SessionLocal) == 0 + + +def test_off_interval_not_multiple_of_10(dsmr_db): + """Seconds 1–9, 11–19, etc. are all off-interval and must be discarded.""" + _, SessionLocal = dsmr_db + settings = _make_settings(dsmr_sample_interval_s=10) + + for second in (1, 3, 7, 9, 11, 23, 47, 59): + ts = f"2026-06-23T12:16:{second:02d}Z" + data = {**_SAMPLE_TELEGRAM, "id": 200000000 + second, "timestamp": ts} + _call_handle_message(data, settings, SessionLocal) + + assert _count_readings(SessionLocal) == 0 + + +# --------------------------------------------------------------------------- +# 3. Idempotency — same source_id not re-inserted +# --------------------------------------------------------------------------- + + +def test_same_source_id_not_reinserted(dsmr_db): + """Feeding the same telegram twice (same id) must result in exactly one row.""" + _, SessionLocal = dsmr_db + settings = _make_settings(dsmr_sample_interval_s=10) + + _call_handle_message(_SAMPLE_TELEGRAM, settings, SessionLocal) + _call_handle_message(_SAMPLE_TELEGRAM, settings, SessionLocal) # duplicate + + assert _count_readings(SessionLocal) == 1 + + +def test_different_source_ids_are_independent(dsmr_db): + """Different id values from different telegrams produce separate rows.""" + _, SessionLocal = dsmr_db + settings = _make_settings(dsmr_sample_interval_s=10) + + _call_handle_message(_SAMPLE_TELEGRAM, settings, SessionLocal) + _call_handle_message(_SAMPLE_TELEGRAM_SECOND_10, settings, SessionLocal) + + assert _count_readings(SessionLocal) == 2 + + +# --------------------------------------------------------------------------- +# 4. Missing source_id — still persisted with source_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.""" + engine, SessionLocal = dsmr_db + settings = _make_settings(dsmr_sample_interval_s=10) + + data = {k: v for k, v in _SAMPLE_TELEGRAM.items() if k != "id"} + + _call_handle_message(data, settings, SessionLocal) + + with Session(engine) as session: + readings = session.scalars(select(DsmrReading)).all() + + assert len(readings) == 1 + assert readings[0].source_id is None + + +# --------------------------------------------------------------------------- +# 5. interval=0 → no ZeroDivisionError; all telegrams persisted +# --------------------------------------------------------------------------- + + +def test_interval_zero_does_not_crash(dsmr_db): + """When dsmr_sample_interval_s=0, no ZeroDivisionError; all telegrams kept.""" + _, SessionLocal = dsmr_db + settings = _make_settings(dsmr_sample_interval_s=0) + + for second in (0, 7, 13, 48, 59): + ts = f"2026-06-23T12:16:{second:02d}Z" + data = {**_SAMPLE_TELEGRAM, "id": 300000000 + second, "timestamp": ts} + _call_handle_message(data, settings, SessionLocal) + + # All 5 should have been persisted (no sampling applied). + assert _count_readings(SessionLocal) == 5 + + +# --------------------------------------------------------------------------- +# 6. Invalid / malformed payloads — silently discarded, no exception propagated +# --------------------------------------------------------------------------- + + +def test_invalid_json_discarded(dsmr_db): + """Invalid JSON payload must be silently discarded without raising.""" + _, SessionLocal = dsmr_db + settings = _make_settings() + from app.services import dsmr_ingest + + with patch.object(dsmr_ingest, "get_session_local", return_value=SessionLocal): + # Must not raise + dsmr_ingest.handle_message(b"not-valid-json", settings) + + assert _count_readings(SessionLocal) == 0 + + +def test_missing_timestamp_discarded(dsmr_db): + """Telegram without a 'timestamp' field must be discarded.""" + _, SessionLocal = dsmr_db + settings = _make_settings() + data = {k: v for k, v in _SAMPLE_TELEGRAM.items() if k != "timestamp"} + + _call_handle_message(data, settings, SessionLocal) + assert _count_readings(SessionLocal) == 0 + + +def test_unparseable_timestamp_discarded(dsmr_db): + """Telegram with a garbage 'timestamp' field must be discarded.""" + _, SessionLocal = dsmr_db + settings = _make_settings() + data = {**_SAMPLE_TELEGRAM, "timestamp": "not-a-date"} + + _call_handle_message(data, settings, SessionLocal) + assert _count_readings(SessionLocal) == 0 + + +def test_handle_message_does_not_propagate_any_exception() -> None: + """handle_message must never propagate any exception to the caller.""" + from app.services import dsmr_ingest + + settings = _make_settings() + + def _explode(): + raise RuntimeError("DB is on fire") + + with patch.object(dsmr_ingest, "get_session_local", side_effect=_explode): + # Must not raise + dsmr_ingest.handle_message(_payload(_SAMPLE_TELEGRAM), settings) + + +# --------------------------------------------------------------------------- +# 7. Null phase values stored verbatim +# --------------------------------------------------------------------------- + + +def test_null_phases_stored_as_none(dsmr_db): + """Null phase values must be preserved as JSON null (Python None) in payload.""" + engine, SessionLocal = dsmr_db + settings = _make_settings(dsmr_sample_interval_s=10) + + _call_handle_message(_SAMPLE_TELEGRAM, settings, SessionLocal) + + with Session(engine) as session: + reading = session.scalars(select(DsmrReading)).first() + + assert reading is not None + assert reading.payload.get("phase_currently_delivered_l2") is None + assert reading.payload.get("phase_voltage_l2") is None + + +# --------------------------------------------------------------------------- +# 8. Gas / extra_device fields stored +# --------------------------------------------------------------------------- + + +def test_gas_fields_stored(dsmr_db): + """extra_device_* (gas) fields must be present in the stored payload.""" + engine, SessionLocal = dsmr_db + settings = _make_settings(dsmr_sample_interval_s=10) + + _call_handle_message(_SAMPLE_TELEGRAM, settings, SessionLocal) + + with Session(engine) as session: + reading = session.scalars(select(DsmrReading)).first() + + assert reading is not None + payload = reading.payload + assert "extra_device_delivered" in payload + assert payload["extra_device_delivered"] == "6208.234" + assert "extra_device_timestamp" in payload + + +# --------------------------------------------------------------------------- +# 9. Timestamp with +00:00 suffix (no Z) also works +# --------------------------------------------------------------------------- + + +def test_timestamp_with_utc_offset_suffix(dsmr_db): + """Timestamps using '+00:00' instead of 'Z' must also be parsed correctly.""" + _, SessionLocal = dsmr_db + settings = _make_settings(dsmr_sample_interval_s=10) + + data = { + **_SAMPLE_TELEGRAM, + "id": 999999, + "timestamp": "2026-06-23T12:16:00+00:00", # no Z, uses +00:00 + } + _call_handle_message(data, settings, SessionLocal) + + assert _count_readings(SessionLocal) == 1 diff --git a/tests/test_mqtt_subscribe.py b/tests/test_mqtt_subscribe.py new file mode 100644 index 0000000..5ef4830 --- /dev/null +++ b/tests/test_mqtt_subscribe.py @@ -0,0 +1,286 @@ +"""Tests for M6-T06: MqttManager subscribe + on_message dispatch. + +All paho interaction is mocked — no real broker is required. + +Covers: +- subscribe() registers topic → handler in the internal registry. +- on_message dispatches the payload to the correct handler. +- A handler that raises does NOT propagate the exception out of on_message. +- Existing publish() behaviour is not regressed. +- When already connected, subscribe() calls client.subscribe() immediately. +- on_connect re-subscribes to all registered topics after (re-)connect. +- subscribe() before connect: topic is subscribed when connect fires. +""" + +from __future__ import annotations + +from unittest.mock import MagicMock, patch + +from app.integrations.mqtt import MqttManager + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_settings( + *, + mqtt_enabled: bool = True, + mqtt_broker_host: str = "broker.test", + mqtt_broker_port: int = 1883, + mqtt_username: str = "", + mqtt_password: str = "", + mqtt_tls_enabled: bool = False, +): + s = MagicMock() + s.mqtt_enabled = mqtt_enabled + s.mqtt_broker_host = mqtt_broker_host + s.mqtt_broker_port = mqtt_broker_port + s.mqtt_username = mqtt_username + s.mqtt_password = mqtt_password + s.mqtt_tls_enabled = mqtt_tls_enabled + return s + + +def _make_mqtt_message(topic: str, payload: bytes) -> MagicMock: + """Construct a minimal mock of a paho MQTTMessage.""" + msg = MagicMock() + msg.topic = topic + msg.payload = payload + return msg + + +# --------------------------------------------------------------------------- +# subscribe() — registry +# --------------------------------------------------------------------------- + + +def test_subscribe_registers_handler() -> None: + """subscribe() must store the handler in the internal _subscriptions dict.""" + manager = MqttManager() + handler = MagicMock() + + manager.subscribe("test/topic", handler) + + assert "test/topic" in manager._subscriptions + assert manager._subscriptions["test/topic"] is handler + + +def test_subscribe_overwrites_previous_handler() -> None: + """Subscribing to the same topic twice replaces the handler.""" + manager = MqttManager() + h1 = MagicMock() + h2 = MagicMock() + + manager.subscribe("test/topic", h1) + manager.subscribe("test/topic", h2) + + assert manager._subscriptions["test/topic"] is h2 + + +def test_subscribe_when_not_connected_does_not_call_paho_subscribe() -> None: + """subscribe() before connect must not attempt to call client.subscribe().""" + real_manager = MqttManager() + handler = MagicMock() + + mock_client = MagicMock() + real_manager._client = None + real_manager._connected = False + + real_manager.subscribe("dsmr/json", handler) + + # No client → subscribe should not have been called on paho + mock_client.subscribe.assert_not_called() + + +def test_subscribe_when_connected_calls_paho_subscribe_immediately() -> None: + """subscribe() while connected must call client.subscribe(topic) right away.""" + manager = MqttManager() + mock_client = MagicMock() + manager._client = mock_client + manager._connected = True + + handler = MagicMock() + manager.subscribe("dsmr/json", handler) + + mock_client.subscribe.assert_called_once_with("dsmr/json") + + +# --------------------------------------------------------------------------- +# on_message dispatch +# --------------------------------------------------------------------------- + + +def _extract_on_message(manager: MqttManager, settings) -> object: + """Connect with a mocked paho client and return the on_message callback that + was registered on the client mock.""" + mock_client = MagicMock() + mock_client.connect.return_value = None + + with patch("app.integrations.mqtt.mqtt.Client", return_value=mock_client): + manager.connect(settings) + + # The callback was assigned as: client.on_message = _on_message + return mock_client.on_message + + +def test_on_message_dispatches_to_registered_handler() -> None: + """on_message must invoke the handler registered for the message's topic.""" + manager = MqttManager() + settings = _make_settings() + + received_payloads: list[bytes] = [] + + def handler(payload: bytes) -> None: + received_payloads.append(payload) + + manager.subscribe("dsmr/json", handler) + + on_message = _extract_on_message(manager, settings) + + msg = _make_mqtt_message("dsmr/json", b'{"id": 1}') + on_message(MagicMock(), None, msg) # simulate paho calling the callback + + assert received_payloads == [b'{"id": 1}'] + + +def test_on_message_ignores_unknown_topic() -> None: + """on_message for an unregistered topic must not raise and must not call any handler.""" + manager = MqttManager() + settings = _make_settings() + handler = MagicMock() + manager.subscribe("other/topic", handler) + + on_message = _extract_on_message(manager, settings) + + msg = _make_mqtt_message("unknown/topic", b"data") + # Must not raise + on_message(MagicMock(), None, msg) + + handler.assert_not_called() + + +def test_on_message_swallows_handler_exception() -> None: + """If the handler raises, on_message must NOT propagate the exception.""" + manager = MqttManager() + settings = _make_settings() + + def bad_handler(payload: bytes) -> None: + raise RuntimeError("handler exploded") + + manager.subscribe("dsmr/json", bad_handler) + on_message = _extract_on_message(manager, settings) + + msg = _make_mqtt_message("dsmr/json", b"{}") + # This must not raise + on_message(MagicMock(), None, msg) + + +def test_on_message_does_not_crash_on_handler_exception_multiple_calls() -> None: + """After a handler exception the manager remains functional for subsequent messages.""" + manager = MqttManager() + settings = _make_settings() + + call_count = [0] + + def flaky_handler(payload: bytes) -> None: + call_count[0] += 1 + if call_count[0] == 1: + raise ValueError("first call fails") + + manager.subscribe("dsmr/json", flaky_handler) + on_message = _extract_on_message(manager, settings) + + msg1 = _make_mqtt_message("dsmr/json", b"first") + msg2 = _make_mqtt_message("dsmr/json", b"second") + + on_message(MagicMock(), None, msg1) # should not raise despite exception in handler + on_message(MagicMock(), None, msg2) # handler called again + + assert call_count[0] == 2 + + +# --------------------------------------------------------------------------- +# on_connect re-subscribes registered topics +# --------------------------------------------------------------------------- + + +def test_on_connect_subscribes_all_registered_topics() -> None: + """When connect fires successfully, _on_connect must subscribe every registered topic.""" + manager = MqttManager() + settings = _make_settings() + + manager.subscribe("topic/a", MagicMock()) + manager.subscribe("topic/b", MagicMock()) + + mock_client = MagicMock() + mock_client.connect.return_value = None + + # Capture the on_connect callback that was registered on the mock client + with patch("app.integrations.mqtt.mqtt.Client", return_value=mock_client): + manager.connect(settings) + + on_connect_cb = mock_client.on_connect + + # Simulate broker accepting the connection (reason_code.is_failure == False) + reason_code = MagicMock() + reason_code.is_failure = False + on_connect_cb(mock_client, None, MagicMock(), reason_code, None) + + # client.subscribe should have been called once per topic during on_connect + subscribe_calls = [c[0][0] for c in mock_client.subscribe.call_args_list] + assert "topic/a" in subscribe_calls + assert "topic/b" in subscribe_calls + + +def test_on_connect_does_not_subscribe_on_failure() -> None: + """When the broker rejects the connection, _on_connect must NOT subscribe.""" + manager = MqttManager() + settings = _make_settings() + manager.subscribe("topic/a", MagicMock()) + + mock_client = MagicMock() + mock_client.connect.return_value = None + + with patch("app.integrations.mqtt.mqtt.Client", return_value=mock_client): + manager.connect(settings) + + on_connect_cb = mock_client.on_connect + reason_code = MagicMock() + reason_code.is_failure = True + + mock_client.subscribe.reset_mock() + on_connect_cb(mock_client, None, MagicMock(), reason_code, None) + + mock_client.subscribe.assert_not_called() + + +# --------------------------------------------------------------------------- +# Existing publish() is not regressed +# --------------------------------------------------------------------------- + + +def test_publish_still_works_after_subscribe_registered() -> None: + """Adding a subscription must not break publish() behaviour.""" + manager = MqttManager() + mock_client = MagicMock() + mock_client.connect.return_value = None + manager._client = mock_client + manager._connected = True + + manager.subscribe("dsmr/json", MagicMock()) + manager.publish("home/sensor", b"data", retain=True, qos=1) + + mock_client.publish.assert_called_once_with( + "home/sensor", payload=b"data", qos=1, retain=True + ) + + +def test_publish_skipped_when_not_connected_regression() -> None: + """publish() must remain a no-op when not connected, even with subscriptions.""" + manager = MqttManager() + manager.subscribe("dsmr/json", MagicMock()) + + # No client set up — publish should be silent + manager.publish("topic", b"payload") # must not raise