home-automation/tests/test_dsmr_ingest.py

"""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


# ---------------------------------------------------------------------------
# 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_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