"""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 — keyed on recorded_at (telegram timestamp), NOT the telegram id # --------------------------------------------------------------------------- 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) _call_handle_message(_SAMPLE_TELEGRAM, settings, SessionLocal) _call_handle_message(_SAMPLE_TELEGRAM, settings, SessionLocal) # duplicate assert _count_readings(SessionLocal) == 1 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) _call_handle_message(_SAMPLE_TELEGRAM, settings, SessionLocal) _call_handle_message(_SAMPLE_TELEGRAM_SECOND_10, settings, SessionLocal) 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 # --------------------------------------------------------------------------- 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 # --------------------------------------------------------------------------- # 10. handle_tariff_message: parse, validate, update, reject invalid payloads # --------------------------------------------------------------------------- @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, "_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 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 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 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" 1 ") assert get_current_tariff() == 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") # Must NOT raise and must NOT change the tariff. assert get_current_tariff() == 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 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 handle_tariff_message(b"0") # 0 is not a valid tariff assert get_current_tariff() == 2 def test_tariff_message_does_not_raise_on_any_input(reset_tariff): """handle_tariff_message must never propagate any exception to the caller.""" from app.services.dsmr_ingest import handle_tariff_message # 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