diff --git a/alembic_app/versions/20260624_12_dsmr_decouple_telegram_id.py b/alembic_app/versions/20260624_12_dsmr_decouple_telegram_id.py new file mode 100644 index 0000000..fde614c --- /dev/null +++ b/alembic_app/versions/20260624_12_dsmr_decouple_telegram_id.py @@ -0,0 +1,58 @@ +"""decouple dsmr_reading from the telegram id + +The DSMR Reader's own ``id`` field is auto-incrementing but is known to overflow +and require a manual reset to zero (a long-standing DSMR firmware quirk). If we +keep a UNIQUE constraint on ``source_id`` (the telegram id) and use it for +idempotency, an overflow/reset would make legitimately-new telegrams collide with +old ids and be silently dropped as "duplicates" — data loss. + +This migration removes that coupling: + +- Drops the UNIQUE constraint on ``source_id`` (the column is kept as a plain, + nullable reference value; it is no longer relied upon for uniqueness/dedup). +- Makes ``recorded_at`` (the telegram timestamp) UNIQUE instead — a single P1 + meter emits one telegram per timestamp, so this is a robust, telegram-id- + independent idempotency key. The table's own autoincrement ``id`` PK remains + the stable internal identity. + +Revision ID: 20260624_12_dsmr_decouple_telegram_id +Revises: 20260623_11_energy_tables +Create Date: 2026-06-24 00:00:00.000000 +""" + +from typing import Sequence, Union + +from alembic import op + + +revision: str = "20260624_12_dsmr_decouple_telegram_id" +down_revision: Union[str, None] = "20260623_11_energy_tables" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # SQLite cannot ALTER away a constraint in place, so recreate the table via + # Alembic's batch mode. The table is recreated and rows are copied; existing + # data (if any) is preserved. recorded_at must be unique for this to succeed + # — a single meter never emits two telegrams at the exact same timestamp. + with op.batch_alter_table("dsmr_reading", schema=None) as batch_op: + # Old non-unique index on recorded_at is replaced by the unique constraint. + batch_op.drop_index("ix_dsmr_reading_recorded_at") + # Telegram id is no longer a uniqueness/idempotency key. + batch_op.drop_constraint("uq_dsmr_reading_source_id", type_="unique") + # Timestamp becomes the telegram-id-independent dedup key. + batch_op.create_unique_constraint( + "uq_dsmr_reading_recorded_at", ["recorded_at"] + ) + + +def downgrade() -> None: + with op.batch_alter_table("dsmr_reading", schema=None) as batch_op: + batch_op.drop_constraint("uq_dsmr_reading_recorded_at", type_="unique") + batch_op.create_unique_constraint( + "uq_dsmr_reading_source_id", ["source_id"] + ) + batch_op.create_index( + "ix_dsmr_reading_recorded_at", ["recorded_at"], unique=False + ) diff --git a/app/api/routes/api/config.py b/app/api/routes/api/config.py index bfffd2b..ffe1120 100644 --- a/app/api/routes/api/config.py +++ b/app/api/routes/api/config.py @@ -86,6 +86,12 @@ def put_config( logger.info("MQTT settings changed — triggering reconnect.") mqtt_manager.reconnect(refreshed_settings) + # Re-apply the DSMR subscription so enabling/disabling DSMR ingest (or changing + # its topic / sample interval) takes effect immediately, without an app restart. + # Done after any MQTT reconnect so it operates on the current client. + from app.services.dsmr_ingest import apply_dsmr_subscription + apply_dsmr_subscription(refreshed_settings) + sections_raw = build_config_sections(db, refreshed_settings) return ConfigUpdateResponse(sections=_sections_from_raw(sections_raw)) diff --git a/app/integrations/mqtt.py b/app/integrations/mqtt.py index 405c5ed..825745a 100644 --- a/app/integrations/mqtt.py +++ b/app/integrations/mqtt.py @@ -175,6 +175,27 @@ class MqttManager: except Exception: logger.exception("MQTT subscribe error (topic=%s).", topic) + def unsubscribe(self, topic: str) -> None: + """Remove the handler for *topic* and unsubscribe from the broker. + + Idempotent: unknown topics are ignored. Used to apply config changes + without a restart (e.g. when DSMR ingest is turned off or its topic + changes). If the client is connected, the broker unsubscribe is sent + immediately; either way the handler is removed from the registry so it + will not be re-subscribed on the next ``_on_connect``. + """ + self._subscriptions.pop(topic, None) + with self._lock: + client = self._client + connected = self._connected + + if client is not None and connected: + try: + client.unsubscribe(topic) + logger.debug("MQTT unsubscribed from topic=%s.", topic) + except Exception: + logger.exception("MQTT unsubscribe error (topic=%s).", topic) + # ------------------------------------------------------------------ # Internal helpers — must be called with self._lock held # ------------------------------------------------------------------ diff --git a/app/main.py b/app/main.py index 9d45197..4853af6 100644 --- a/app/main.py +++ b/app/main.py @@ -29,7 +29,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.dsmr_ingest import apply_dsmr_subscription 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 @@ -246,21 +246,10 @@ 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.") + # 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) yield diff --git a/app/models/energy.py b/app/models/energy.py index 55462c4..a9ecbc7 100644 --- a/app/models/energy.py +++ b/app/models/energy.py @@ -22,27 +22,33 @@ from app.db import Base class DsmrReading(Base): """One down-sampled DSMR telegram stored as a full JSON blob. - ``recorded_at`` is a real indexed column (not inside the payload) so that - time-range queries are efficient. ``source_id`` holds the telegram's own - integer ``id`` and is used for idempotent upsert (skip if already present). - The entire telegram frame is stored verbatim in ``payload``; no field - allow-list is applied so future commodities (gas, heating, three-phase) - are accommodated without a schema change. + Identity & idempotency are **independent of the DSMR Reader's telegram id** + (that field overflows and must be manually reset to zero — a known DSMR + quirk — so relying on it for uniqueness risks silently dropping new data). + The table's own autoincrement ``id`` PK is the stable internal identity, and + ``recorded_at`` (the telegram timestamp) is the UNIQUE de-duplication key: a + single P1 meter emits exactly one telegram per timestamp. + + ``recorded_at`` is a real column (not inside the payload) so time-range + queries are efficient. The entire telegram frame is stored verbatim in + ``payload``; no field allow-list is applied so future commodities (gas, + heating, three-phase) are accommodated without a schema change. """ __tablename__ = "dsmr_reading" id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) - # UTC timestamp of the sample — real column so it can be indexed efficiently. + # UTC timestamp of the sample — real column, UNIQUE (telegram-id-independent + # idempotency key). The unique index also serves time-range queries. recorded_at: Mapped[datetime] = mapped_column( - DateTime(timezone=True), nullable=False, index=True + DateTime(timezone=True), nullable=False, unique=True ) - # Telegram's own auto-increment id (DSMR Reader assigns it). Used for - # idempotent ingestion: if the same telegram arrives twice we skip it. - # Nullable because some DSMR sources may not emit an id. - source_id: Mapped[int | None] = mapped_column(Integer, unique=True, nullable=True) + # Telegram's own id (DSMR Reader assigns it). Stored only as a reference / + # debugging aid — NOT used for uniqueness or idempotency (it overflows and + # gets reset to zero). Nullable because some DSMR sources may not emit one. + source_id: Mapped[int | None] = mapped_column(Integer, nullable=True) # Full telegram frame as a JSON object; values are typically JSON strings # (e.g. "20915.154") — callers must cast to Decimal before arithmetic. diff --git a/app/services/dsmr_ingest.py b/app/services/dsmr_ingest.py index 295d09f..27ba26f 100644 --- a/app/services/dsmr_ingest.py +++ b/app/services/dsmr_ingest.py @@ -47,6 +47,50 @@ 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 + + +def apply_dsmr_subscription(settings: "Settings") -> None: + """(Re)apply the DSMR MQTT subscription to match *settings* — restart-free. + + Call this at startup and after every config save. It makes the live MQTT + subscription reflect the current ``dsmr_ingest_enabled`` / ``dsmr_mqtt_topic`` + / ``dsmr_sample_interval_s`` settings without an app restart: + + - **Disabled** → unsubscribe any active DSMR subscription. + - **Enabled** → (re)subscribe to ``dsmr_mqtt_topic`` with a handler bound to + a *fresh* settings snapshot, so a changed sample interval also takes effect. + - **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 + + 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 + return + + 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) + + def handle_message(payload_bytes: bytes, settings: "Settings") -> None: """Parse one DSMR MQTT payload and persist it if it passes the sample filter. @@ -107,7 +151,8 @@ def _handle_message_inner(payload_bytes: bytes, settings: "Settings") -> None: # This telegram is between sample points; discard silently. return - # --- 4. Extract source_id (telegram's own id field, for idempotency) --- + # --- 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. @@ -118,19 +163,22 @@ def _handle_message_inner(payload_bytes: bytes, settings: "Settings") -> None: 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. 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) + 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(), ) - if existing is not None: - logger.debug( - "dsmr_ingest: source_id=%d already in DB, skipping.", source_id - ) - return + return reading = DsmrReading( recorded_at=ts_utc, @@ -145,12 +193,11 @@ def _handle_message_inner(payload_bytes: bytes, settings: "Settings") -> None: 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. + # Race / duplicate: another insert beat us to the same recorded_at. session.rollback() logger.debug( - "dsmr_ingest: IntegrityError for source_id=%s (duplicate, skipped).", - source_id, + "dsmr_ingest: IntegrityError for recorded_at=%s (duplicate, skipped).", + ts_utc.isoformat(), ) except Exception: session.rollback() diff --git a/scripts/app_db_adopt.py b/scripts/app_db_adopt.py index 68b9f6f..70f0cad 100644 --- a/scripts/app_db_adopt.py +++ b/scripts/app_db_adopt.py @@ -15,7 +15,7 @@ if str(PROJECT_ROOT) not in sys.path: from app.config import get_settings -APP_BASELINE_REVISION = "20260623_11_energy_tables" +APP_BASELINE_REVISION = "20260624_12_dsmr_decouple_telegram_id" class AppDatabaseAdoptionError(RuntimeError): diff --git a/tests/test_api_config.py b/tests/test_api_config.py index 18f18fa..50938ba 100644 --- a/tests/test_api_config.py +++ b/tests/test_api_config.py @@ -187,6 +187,28 @@ 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( + 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).""" + _login(client) + + payload = _full_config_payload({"DSMR_INGEST_ENABLED": "true"}) + with patch("app.services.dsmr_ingest.apply_dsmr_subscription") as spy: + response = client.put( + "/api/config", + json={"updates": payload}, + headers={"X-CSRF-Token": "any-non-empty-value"}, + ) + + assert response.status_code == 200 + spy.assert_called_once() + applied_settings = spy.call_args.args[0] + assert applied_settings.dsmr_ingest_enabled is True + + def test_put_config_blank_secret_keeps_existing_value( client: TestClient, test_database_urls ) -> None: diff --git a/tests/test_dsmr_ingest.py b/tests/test_dsmr_ingest.py index c970ba5..8afb7fd 100644 --- a/tests/test_dsmr_ingest.py +++ b/tests/test_dsmr_ingest.py @@ -214,12 +214,12 @@ def test_off_interval_not_multiple_of_10(dsmr_db): # --------------------------------------------------------------------------- -# 3. Idempotency — same source_id not re-inserted +# 3. Idempotency — keyed on recorded_at (telegram timestamp), NOT the telegram id # --------------------------------------------------------------------------- -def test_same_source_id_not_reinserted(dsmr_db): - """Feeding the same telegram twice (same id) must result in exactly one row.""" +def test_same_timestamp_not_reinserted(dsmr_db): + """Feeding the same telegram twice (same timestamp) must result in one row.""" _, SessionLocal = dsmr_db settings = _make_settings(dsmr_sample_interval_s=10) @@ -229,8 +229,8 @@ def test_same_source_id_not_reinserted(dsmr_db): assert _count_readings(SessionLocal) == 1 -def test_different_source_ids_are_independent(dsmr_db): - """Different id values from different telegrams produce separate rows.""" +def test_different_timestamps_are_independent(dsmr_db): + """Telegrams with different timestamps produce separate rows.""" _, SessionLocal = dsmr_db settings = _make_settings(dsmr_sample_interval_s=10) @@ -240,6 +240,24 @@ def test_different_source_ids_are_independent(dsmr_db): assert _count_readings(SessionLocal) == 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 + telegram id must BOTH be stored — dedup must not depend on the telegram id.""" + _, SessionLocal = dsmr_db + settings = _make_settings(dsmr_sample_interval_s=10) + + first = {**_SAMPLE_TELEGRAM, "id": 0, "timestamp": "2026-06-23T12:16:00Z"} + # Later telegram, id reset back to the same value after an overflow. + second = {**_SAMPLE_TELEGRAM, "id": 0, "timestamp": "2026-06-23T12:16:10Z"} + + _call_handle_message(first, settings, SessionLocal) + _call_handle_message(second, settings, SessionLocal) + + # Both must persist — the colliding telegram id must not cause a drop. + assert _count_readings(SessionLocal) == 2 + + # --------------------------------------------------------------------------- # 4. Missing source_id — still persisted with source_id=None # --------------------------------------------------------------------------- diff --git a/tests/test_dsmr_subscription_apply.py b/tests/test_dsmr_subscription_apply.py new file mode 100644 index 0000000..37621a9 --- /dev/null +++ b/tests/test_dsmr_subscription_apply.py @@ -0,0 +1,92 @@ +"""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. +""" + +from __future__ import annotations + +from unittest.mock import MagicMock + +import pytest + +from app.services import dsmr_ingest + + +class _FakeMqtt: + def __init__(self) -> None: + self.subscribe_calls: list[tuple[str, object]] = [] + self.unsubscribe_calls: list[str] = [] + + def subscribe(self, topic: str, handler) -> None: + self.subscribe_calls.append((topic, handler)) + + def unsubscribe(self, topic: str) -> None: + self.unsubscribe_calls.append(topic) + + +def _settings(*, enabled: bool = True, topic: str = "dsmr/json", interval: int = 10): + s = MagicMock() + s.dsmr_ingest_enabled = enabled + s.dsmr_mqtt_topic = topic + s.dsmr_sample_interval_s = interval + return s + + +@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 topic". + monkeypatch.setattr(dsmr_ingest, "_current_dsmr_topic", None) + 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_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_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_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_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 diff --git a/tests/test_energy_models.py b/tests/test_energy_models.py index 093faa8..de86793 100644 --- a/tests/test_energy_models.py +++ b/tests/test_energy_models.py @@ -117,21 +117,21 @@ def test_dsmr_reading_columns(energy_db): assert "payload" in columns and not columns["payload"]["nullable"] -def test_dsmr_reading_source_id_unique(energy_db): - """dsmr_reading.source_id must have a unique constraint.""" +def test_dsmr_reading_recorded_at_unique(energy_db): + """dsmr_reading.recorded_at is the UNIQUE de-dup key (telegram-id-independent).""" inspector = inspect(energy_db) unique_constraints = inspector.get_unique_constraints("dsmr_reading") unique_cols = [col for uc in unique_constraints for col in uc["column_names"]] - assert "source_id" in unique_cols, "source_id must have a unique constraint" + assert "recorded_at" in unique_cols, "recorded_at must have a unique constraint" -def test_dsmr_reading_recorded_at_index(energy_db): - """dsmr_reading.recorded_at must have an index.""" +def test_dsmr_reading_source_id_not_unique(energy_db): + """dsmr_reading.source_id (telegram id) must NOT be unique — it overflows/resets, + so it is kept only as a reference value and never relied on for dedup.""" inspector = inspect(energy_db) - indexes = {idx["name"]: idx for idx in inspector.get_indexes("dsmr_reading")} - assert "ix_dsmr_reading_recorded_at" in indexes, ( - "ix_dsmr_reading_recorded_at missing from dsmr_reading" - ) + unique_constraints = inspector.get_unique_constraints("dsmr_reading") + unique_cols = [col for uc in unique_constraints for col in uc["column_names"]] + assert "source_id" not in unique_cols, "source_id must NOT have a unique constraint" def test_energy_contract_columns(energy_db): @@ -268,6 +268,40 @@ def test_downgrade_removes_energy_tables(tmp_path: Path): engine.dispose() +def test_dsmr_decouple_migration_downgrade_restores_source_id_unique(tmp_path: Path): + """Downgrading the telegram-id-decouple migration to 20260623_11 must restore the + original constraints: source_id UNIQUE, recorded_at non-unique index.""" + db_path = tmp_path / "dsmr_decouple_downgrade.db" + db_url = f"sqlite:///{db_path}" + alembic_cfg = _make_app_alembic_config(db_url) + + command.upgrade(alembic_cfg, "head") + + # At head: recorded_at is unique, source_id is not. + engine = create_engine(db_url, connect_args={"check_same_thread": False}) + inspector = inspect(engine) + head_unique = [ + c for uc in inspector.get_unique_constraints("dsmr_reading") for c in uc["column_names"] + ] + assert "recorded_at" in head_unique + assert "source_id" not in head_unique + engine.dispose() + + # Downgrade just this migration. + command.downgrade(alembic_cfg, "20260623_11_energy_tables") + + engine = create_engine(db_url, connect_args={"check_same_thread": False}) + inspector = inspect(engine) + pre_unique = [ + c for uc in inspector.get_unique_constraints("dsmr_reading") for c in uc["column_names"] + ] + assert "source_id" in pre_unique, "source_id unique must be restored on downgrade" + assert "recorded_at" not in pre_unique + index_names = {idx["name"] for idx in inspector.get_indexes("dsmr_reading")} + assert "ix_dsmr_reading_recorded_at" in index_names + engine.dispose() + + # --------------------------------------------------------------------------- # 2. ORM metadata checks (Base.metadata) # --------------------------------------------------------------------------- @@ -332,11 +366,12 @@ def test_energy_cost_period_fk_ondelete_restrict(): ) -def test_dsmr_reading_source_id_unique_in_metadata(): - """DsmrReading.source_id must be declared unique in ORM metadata.""" +def test_dsmr_reading_recorded_at_unique_in_metadata(): + """DsmrReading.recorded_at must be the unique de-dup key in ORM metadata, + and source_id must NOT be unique (decoupled from the telegram id).""" table = Base.metadata.tables["dsmr_reading"] - col = table.columns["source_id"] - assert col.unique, "source_id must be declared unique" + assert table.columns["recorded_at"].unique, "recorded_at must be declared unique" + assert not table.columns["source_id"].unique, "source_id must NOT be unique" def test_tibber_price_starts_at_unique_in_metadata(): diff --git a/tests/test_mqtt_subscribe.py b/tests/test_mqtt_subscribe.py index 5ef4830..42155da 100644 --- a/tests/test_mqtt_subscribe.py +++ b/tests/test_mqtt_subscribe.py @@ -107,6 +107,36 @@ def test_subscribe_when_connected_calls_paho_subscribe_immediately() -> None: mock_client.subscribe.assert_called_once_with("dsmr/json") +def test_unsubscribe_removes_handler_from_registry() -> None: + """unsubscribe() must drop the handler so it is not re-subscribed on reconnect.""" + manager = MqttManager() + manager.subscribe("dsmr/json", MagicMock()) + + manager.unsubscribe("dsmr/json") + + assert "dsmr/json" not in manager._subscriptions + + +def test_unsubscribe_when_connected_calls_paho_unsubscribe() -> None: + """unsubscribe() while connected must call client.unsubscribe(topic).""" + manager = MqttManager() + mock_client = MagicMock() + manager._client = mock_client + manager._connected = True + manager.subscribe("dsmr/json", MagicMock()) + + manager.unsubscribe("dsmr/json") + + mock_client.unsubscribe.assert_called_once_with("dsmr/json") + + +def test_unsubscribe_unknown_topic_is_noop() -> None: + """unsubscribe() on a topic that was never registered must not raise.""" + manager = MqttManager() + manager.unsubscribe("never/registered") # must not raise + assert "never/registered" not in manager._subscriptions + + # --------------------------------------------------------------------------- # on_message dispatch # ---------------------------------------------------------------------------