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