895 lines
35 KiB
Python
895 lines
35 KiB
Python
"""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 datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from unittest.mock import 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
|
||
from app.models.meter_source import MeterSource
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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",
|
||
):
|
||
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.
|
||
# 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_accepted_message_marks_source_online_and_updates_last_seen(dsmr_db):
|
||
"""A valid, accepted telegram clears stale diagnostics without changing its payload."""
|
||
engine, SessionLocal = dsmr_db
|
||
settings = _make_settings(dsmr_sample_interval_s=10)
|
||
with Session(engine) as session:
|
||
source = session.get(MeterSource, 1)
|
||
assert source is not None
|
||
source.enabled = True
|
||
source.status = "error"
|
||
source.last_error = "old error"
|
||
session.commit()
|
||
|
||
_call_handle_message(_SAMPLE_TELEGRAM, settings, SessionLocal)
|
||
|
||
with Session(engine) as session:
|
||
source = session.get(MeterSource, 1)
|
||
assert source is not None
|
||
assert source.status == "online"
|
||
assert source.last_error is None
|
||
assert source.last_seen_at is not None
|
||
|
||
|
||
def test_source_state_callback_persists_only_current_generation(dsmr_db, monkeypatch):
|
||
"""Connection callbacks use a short session and stale generations leave health untouched."""
|
||
engine, SessionLocal = dsmr_db
|
||
from app.services import dsmr_ingest
|
||
|
||
snapshot = _make_settings()
|
||
current = object()
|
||
monkeypatch.setattr(dsmr_ingest, "_subscription_tokens", {snapshot.source_id: current})
|
||
with Session(engine) as session:
|
||
source = session.get(MeterSource, snapshot.source_id)
|
||
assert source is not None
|
||
source.enabled = True
|
||
session.commit()
|
||
with patch.object(dsmr_ingest, "get_session_local", return_value=SessionLocal):
|
||
dsmr_ingest.handle_captured_source_state(snapshot, current, "connecting")
|
||
dsmr_ingest.handle_captured_source_state(snapshot, current, "error")
|
||
dsmr_ingest.handle_captured_source_state(snapshot, object(), "online")
|
||
|
||
with Session(engine) as session:
|
||
source = session.get(MeterSource, snapshot.source_id)
|
||
assert source is not None
|
||
assert source.status == "error"
|
||
assert source.last_error == "MQTT connection failed."
|
||
|
||
|
||
def test_source_health_is_isolated_between_sources(dsmr_db, monkeypatch):
|
||
"""Connection and message health changes only touch their own source row."""
|
||
engine, SessionLocal = dsmr_db
|
||
from app.services import dsmr_ingest
|
||
from app.services.dsmr_ingest import DsmrSourceSnapshot
|
||
|
||
first = _make_settings()
|
||
second = DsmrSourceSnapshot(2, "second/topic", "", 10)
|
||
first_token = object()
|
||
second_token = object()
|
||
monkeypatch.setattr(
|
||
dsmr_ingest, "_subscription_tokens", {first.source_id: first_token, second.source_id: second_token}
|
||
)
|
||
with Session(engine) as session:
|
||
first_source = session.get(MeterSource, first.source_id)
|
||
assert first_source is not None
|
||
first_source.enabled = True
|
||
session.add(MeterSource(
|
||
id=2, name="Second DSMR", kind="dsmr_mqtt", enabled=True, config={},
|
||
status="online", created_at=datetime.now(timezone.utc), updated_at=datetime.now(timezone.utc),
|
||
))
|
||
session.commit()
|
||
|
||
with patch.object(dsmr_ingest, "get_session_local", return_value=SessionLocal):
|
||
dsmr_ingest.handle_captured_source_state(first, first_token, "connecting")
|
||
dsmr_ingest.handle_captured_source_state(second, second_token, "error")
|
||
dsmr_ingest.handle_message(_payload(_SAMPLE_TELEGRAM), first)
|
||
|
||
with Session(engine) as session:
|
||
first_source = session.get(MeterSource, first.source_id)
|
||
second_source = session.get(MeterSource, second.source_id)
|
||
assert first_source is not None and second_source is not None
|
||
assert first_source.status == "online"
|
||
assert first_source.last_error is None
|
||
assert first_source.last_seen_at is not None
|
||
assert second_source.status == "error"
|
||
assert second_source.last_error == "MQTT connection failed."
|
||
assert second_source.last_seen_at is None
|
||
|
||
|
||
def test_disable_reconcile_persists_unknown_and_rejects_retained_state_callback(dsmr_db, monkeypatch):
|
||
"""Disable invalidates the generation before clearing a stale online health state."""
|
||
engine, SessionLocal = dsmr_db
|
||
from app.services import dsmr_ingest
|
||
|
||
class FakeMqtt:
|
||
def __init__(self) -> None:
|
||
self.removed: list[int] = []
|
||
self.state_handler = None
|
||
self.active: set[int] = set()
|
||
|
||
def replace_source(self, source_id: int, **kwargs: object) -> bool:
|
||
self.active.add(source_id)
|
||
self.state_handler = kwargs["state_handler"]
|
||
return True
|
||
|
||
def remove_source(self, source_id: int) -> None:
|
||
self.removed.append(source_id)
|
||
self.active.discard(source_id)
|
||
|
||
def source_is_active(self, source_id: int) -> bool:
|
||
return source_id in self.active
|
||
|
||
fake = FakeMqtt()
|
||
snapshot = _make_settings()
|
||
monkeypatch.setattr(dsmr_ingest, "_subscriptions", {})
|
||
monkeypatch.setattr(dsmr_ingest, "_subscription_client_ids", {})
|
||
monkeypatch.setattr(dsmr_ingest, "_subscription_tokens", {})
|
||
monkeypatch.setattr(dsmr_ingest, "_tariffs", {})
|
||
monkeypatch.setattr(dsmr_ingest, "get_session_local", lambda: SessionLocal)
|
||
monkeypatch.setattr("app.integrations.mqtt.mqtt_manager", fake)
|
||
monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [snapshot])
|
||
with Session(engine) as session:
|
||
source = session.get(MeterSource, snapshot.source_id)
|
||
assert source is not None
|
||
source.enabled = True
|
||
source.status = "online"
|
||
session.commit()
|
||
dsmr_ingest.apply_dsmr_subscription()
|
||
assert fake.state_handler is not None
|
||
|
||
with Session(engine) as session:
|
||
source = session.get(MeterSource, snapshot.source_id)
|
||
assert source is not None
|
||
source.enabled = False
|
||
session.commit()
|
||
monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [])
|
||
dsmr_ingest.apply_dsmr_subscription()
|
||
fake.state_handler("online")
|
||
|
||
with Session(engine) as session:
|
||
source = session.get(MeterSource, snapshot.source_id)
|
||
assert source is not None
|
||
assert source.status == "unknown"
|
||
assert source.last_error is None
|
||
assert fake.removed == [snapshot.source_id]
|
||
|
||
|
||
def test_topic_collision_persists_error_when_reconcile_removes_active_client(dsmr_db, monkeypatch):
|
||
"""A rejected enabled source cannot retain health from its removed client."""
|
||
engine, SessionLocal = dsmr_db
|
||
from app.services import dsmr_ingest
|
||
|
||
class FakeMqtt:
|
||
def __init__(self) -> None:
|
||
self.removed: list[int] = []
|
||
self.active: set[int] = set()
|
||
|
||
def replace_source(self, source_id: int, **kwargs: object) -> bool:
|
||
self.active.add(source_id)
|
||
return True
|
||
|
||
def remove_source(self, source_id: int) -> None:
|
||
self.removed.append(source_id)
|
||
self.active.discard(source_id)
|
||
|
||
def source_is_active(self, source_id: int) -> bool:
|
||
return source_id in self.active
|
||
|
||
fake = FakeMqtt()
|
||
valid = _make_settings(dsmr_mqtt_topic="dsmr/telegram")
|
||
collision = _make_settings(dsmr_mqtt_topic="dsmr/telegram")
|
||
collision = dsmr_ingest.DsmrSourceSnapshot(
|
||
collision.source_id,
|
||
collision.topic,
|
||
collision.topic,
|
||
collision.sample_interval_s,
|
||
)
|
||
monkeypatch.setattr(dsmr_ingest, "_subscriptions", {})
|
||
monkeypatch.setattr(dsmr_ingest, "_subscription_client_ids", {})
|
||
monkeypatch.setattr(dsmr_ingest, "_subscription_tokens", {})
|
||
monkeypatch.setattr(dsmr_ingest, "_tariffs", {})
|
||
monkeypatch.setattr(dsmr_ingest, "get_session_local", lambda: SessionLocal)
|
||
monkeypatch.setattr("app.integrations.mqtt.mqtt_manager", fake)
|
||
with Session(engine) as session:
|
||
source = session.get(MeterSource, valid.source_id)
|
||
assert source is not None
|
||
source.enabled = True
|
||
source.status = "online"
|
||
session.commit()
|
||
|
||
monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [valid])
|
||
dsmr_ingest.apply_dsmr_subscription()
|
||
monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [collision])
|
||
dsmr_ingest.apply_dsmr_subscription()
|
||
|
||
with Session(engine) as session:
|
||
source = session.get(MeterSource, valid.source_id)
|
||
assert source is not None
|
||
assert source.enabled is True
|
||
assert source.status == "error"
|
||
assert source.last_error == "DSMR source configuration invalid."
|
||
assert fake.removed == [valid.source_id]
|
||
assert fake.active == set()
|
||
|
||
|
||
def test_topic_collision_persists_error_on_startup_without_runtime_client(dsmr_db, monkeypatch):
|
||
"""A collision corrects stale persisted online health without an installed client."""
|
||
engine, SessionLocal = dsmr_db
|
||
from app.services import dsmr_ingest
|
||
|
||
collision = dsmr_ingest.DsmrSourceSnapshot(1, "dsmr/telegram", "dsmr/telegram", 10)
|
||
monkeypatch.setattr(dsmr_ingest, "_subscriptions", {})
|
||
monkeypatch.setattr(dsmr_ingest, "_subscription_client_ids", {})
|
||
monkeypatch.setattr(dsmr_ingest, "_subscription_tokens", {})
|
||
monkeypatch.setattr(dsmr_ingest, "get_session_local", lambda: SessionLocal)
|
||
monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [collision])
|
||
with Session(engine) as session:
|
||
source = session.get(MeterSource, collision.source_id)
|
||
assert source is not None
|
||
source.enabled = True
|
||
source.status = "online"
|
||
session.commit()
|
||
|
||
dsmr_ingest.apply_dsmr_subscription()
|
||
|
||
with Session(engine) as session:
|
||
source = session.get(MeterSource, collision.source_id)
|
||
assert source is not None
|
||
assert source.status == "error"
|
||
assert source.last_error == "DSMR source configuration invalid."
|
||
|
||
|
||
def test_topic_collision_does_not_change_a_disabled_source(dsmr_db, monkeypatch):
|
||
"""A concurrent disable is not overwritten by collision error handling."""
|
||
engine, SessionLocal = dsmr_db
|
||
from app.services import dsmr_ingest
|
||
|
||
collision = dsmr_ingest.DsmrSourceSnapshot(1, "dsmr/telegram", "dsmr/telegram", 10)
|
||
monkeypatch.setattr(dsmr_ingest, "_subscriptions", {})
|
||
monkeypatch.setattr(dsmr_ingest, "_subscription_client_ids", {})
|
||
monkeypatch.setattr(dsmr_ingest, "_subscription_tokens", {})
|
||
monkeypatch.setattr(dsmr_ingest, "get_session_local", lambda: SessionLocal)
|
||
monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [collision])
|
||
with Session(engine) as session:
|
||
source = session.get(MeterSource, collision.source_id)
|
||
assert source is not None
|
||
source.enabled = False
|
||
source.status = "unknown"
|
||
source.last_error = None
|
||
session.commit()
|
||
|
||
dsmr_ingest.apply_dsmr_subscription()
|
||
|
||
with Session(engine) as session:
|
||
source = session.get(MeterSource, collision.source_id)
|
||
assert source is not None
|
||
assert source.status == "unknown"
|
||
assert source.last_error 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_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
|
||
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 telegram id — still persisted with telegram_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)
|
||
|
||
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].telegram_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):
|
||
from app.services import dsmr_ingest as _di
|
||
|
||
monkeypatch.setattr(_di, "_tariffs", {})
|
||
monkeypatch.setattr(_di, "_current_tariff", None)
|
||
|
||
|
||
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", 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", 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", 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", 1)
|
||
assert get_current_tariff(1) == 2
|
||
|
||
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(1, 2)
|
||
handle_tariff_message(b"x", 1)
|
||
# Must NOT raise and must NOT change the tariff.
|
||
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, 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(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", 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):
|
||
"""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, 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
|