home-automation/tests/test_dsmr_subscription_apply.py

"""DSMR source-driven MQTT subscription reconciliation tests."""

from __future__ import annotations

from dataclasses import replace
from types import SimpleNamespace

import pytest

from app.services import dsmr_ingest
from app.services.dsmr_ingest import DsmrSourceSnapshot


class _FakeMqtt:
    def __init__(self) -> None:
        self.replace_calls: list[tuple[int, dict[str, object]]] = []
        self.remove_calls: list[int] = []
        self.handlers: dict[int, dict[str, object]] = {}

    def replace_source(self, source_id: int, **kwargs: object) -> bool:
        self.replace_calls.append((source_id, kwargs))
        self.handlers[source_id] = kwargs["subscriptions"]  # type: ignore[assignment]
        return True

    def remove_source(self, source_id: int) -> None:
        self.remove_calls.append(source_id)
        self.handlers.pop(source_id, None)

    def source_is_active(self, source_id: int) -> bool:
        return source_id in self.handlers


def _source(
    source_id: int,
    topic: str,
    tariff_topic: str = "",
    interval: int = 10,
    **connection: object,
) -> DsmrSourceSnapshot:
    return DsmrSourceSnapshot(source_id, topic, tariff_topic, interval, **connection)


@pytest.fixture()
def fake_mqtt(monkeypatch):
    fake = _FakeMqtt()
    monkeypatch.setattr("app.integrations.mqtt.mqtt_manager", fake)
    monkeypatch.setattr(dsmr_ingest, "_subscriptions", {})
    monkeypatch.setattr(dsmr_ingest, "_subscription_client_ids", {})
    monkeypatch.setattr(dsmr_ingest, "_subscription_tokens", {})
    monkeypatch.setattr(dsmr_ingest, "_tariffs", {})
    return fake


def test_reconcile_subscribes_each_enabled_source(fake_mqtt, monkeypatch):
    monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [_source(1, "a"), _source(2, "b")])
    dsmr_ingest.apply_dsmr_subscription()
    assert [source_id for source_id, _ in fake_mqtt.replace_calls] == [1, 2]


def test_changed_source_replaces_only_its_subscription(fake_mqtt, monkeypatch):
    monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [_source(1, "a"), _source(2, "b")])
    dsmr_ingest.apply_dsmr_subscription()
    monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [_source(1, "changed"), _source(2, "b")])
    dsmr_ingest.apply_dsmr_subscription()
    assert fake_mqtt.remove_calls == [1]
    assert [source_id for source_id, _ in fake_mqtt.replace_calls] == [1, 2, 1]


def test_disable_unsubscribes_and_clears_source_tariff(fake_mqtt, monkeypatch):
    monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [_source(1, "a", "tariff/a")])
    dsmr_ingest.apply_dsmr_subscription()
    dsmr_ingest.set_current_tariff(1, 2)
    monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [])
    dsmr_ingest.apply_dsmr_subscription()
    assert fake_mqtt.remove_calls == [1]
    assert dsmr_ingest.get_current_tariff(1) is None


def test_same_snapshot_has_no_subscription_churn(fake_mqtt, monkeypatch):
    monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [_source(1, "a")])
    dsmr_ingest.apply_dsmr_subscription()
    dsmr_ingest.apply_dsmr_subscription()
    assert len(fake_mqtt.replace_calls) == 1
    assert fake_mqtt.remove_calls == []


def test_changed_base_client_id_replaces_all_enabled_sources(fake_mqtt, monkeypatch):
    snapshot = _source(1, "topic", broker_host="broker.test")
    monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [snapshot])

    dsmr_ingest.apply_dsmr_subscription(SimpleNamespace(mqtt_client_id="home-automation"))
    dsmr_ingest.apply_dsmr_subscription(SimpleNamespace(mqtt_client_id="home-automation-dev"))

    assert fake_mqtt.remove_calls == [1]
    assert [source_id for source_id, _ in fake_mqtt.replace_calls] == [1, 1]
    assert fake_mqtt.replace_calls[-1][1]["base_client_id"] == "home-automation-dev"


def test_same_topic_on_different_brokers_is_allowed(fake_mqtt, monkeypatch):
    monkeypatch.setattr(
        dsmr_ingest,
        "_enabled_snapshots",
        lambda: [_source(1, "same", broker_host="one.test"), _source(2, "same", broker_host="two.test")],
    )
    dsmr_ingest.apply_dsmr_subscription()
    assert [source_id for source_id, _ in fake_mqtt.replace_calls] == [1, 2]


@pytest.mark.parametrize(
    "sources",
    [
        [_source(1, "shared", "shared")],
    ],
)
def test_duplicate_telegram_or_tariff_topic_is_rejected(fake_mqtt, monkeypatch, sources):
    """One MQTT topic cannot safely dispatch to more than one source handler."""
    monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: sources)
    dsmr_ingest.apply_dsmr_subscription()
    assert fake_mqtt.replace_calls == []


@pytest.mark.parametrize(
    "field,value",
    [
        ("broker_host", "changed.test"),
        ("broker_port", 2883),
        ("username", "different-user"),
        ("password", "different-password"),
        ("tls_enabled", True),
        ("sample_interval_s", 30),
        ("tariff_topic", "tariff/changed"),
    ],
)
def test_each_source_config_change_replaces_only_that_source(fake_mqtt, monkeypatch, field, value):
    first = _source(1, "a", "tariff/a", broker_host="one.test", username="one")
    second = _source(2, "b", "tariff/b", broker_host="two.test", username="two")
    monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [first, second])
    dsmr_ingest.apply_dsmr_subscription()
    changed = _source(1, "a", "tariff/a", broker_host="one.test", username="one")
    changed = replace(changed, **{field: value})
    monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [changed, second])
    dsmr_ingest.apply_dsmr_subscription()
    assert fake_mqtt.remove_calls == [1]
    assert [source_id for source_id, _ in fake_mqtt.replace_calls] == [1, 2, 1]


def test_failed_replace_is_not_marked_applied_and_is_retried(fake_mqtt, monkeypatch):
    snapshot = _source(1, "a", broker_host="one.test")
    monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [snapshot])
    original_replace = fake_mqtt.replace_source
    outcomes = iter([False, True])

    def replace_once_fails(source_id: int, **kwargs: object) -> bool:
        original_replace(source_id, **kwargs)
        return next(outcomes)

    fake_mqtt.replace_source = replace_once_fails  # type: ignore[method-assign]
    dsmr_ingest.apply_dsmr_subscription()
    dsmr_ingest.apply_dsmr_subscription()
    assert [source_id for source_id, _ in fake_mqtt.replace_calls] == [1, 1]


def test_inactive_manager_source_is_rebuilt_on_next_reconcile(fake_mqtt, monkeypatch):
    snapshot = _source(1, "a", broker_host="one.test")
    monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [snapshot])
    dsmr_ingest.apply_dsmr_subscription()
    fake_mqtt.handlers.clear()  # models MqttManager.disconnect() tearing down clients
    dsmr_ingest.apply_dsmr_subscription()
    assert [source_id for source_id, _ in fake_mqtt.replace_calls] == [1, 1]


def test_same_snapshot_aba_rejects_retained_handler(fake_mqtt, monkeypatch):
    """An equal re-enabled snapshot has a fresh callback identity token."""
    snapshot = _source(1, "same", broker_host="one.test")
    received: list[tuple[bytes, DsmrSourceSnapshot]] = []
    monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [snapshot])
    monkeypatch.setattr(
        dsmr_ingest, "handle_message", lambda payload, captured: received.append((payload, captured))
    )
    dsmr_ingest.apply_dsmr_subscription()
    old_handler = fake_mqtt.handlers[1]["same"]
    monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [])
    dsmr_ingest.apply_dsmr_subscription()
    monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [snapshot])
    dsmr_ingest.apply_dsmr_subscription()

    old_handler(b"stale")  # type: ignore[operator]
    fake_mqtt.handlers[1]["same"](b"fresh")  # type: ignore[operator]
    assert received == [(b"fresh", snapshot)]


def test_source_state_handler_is_generation_scoped(fake_mqtt, monkeypatch):
    snapshot = _source(1, "same", broker_host="one.test")
    monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [snapshot])
    dsmr_ingest.apply_dsmr_subscription()
    stale = fake_mqtt.replace_calls[-1][1]["state_handler"]
    monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [])
    dsmr_ingest.apply_dsmr_subscription()
    monkeypatch.setattr(dsmr_ingest, "_enabled_snapshots", lambda: [snapshot])
    dsmr_ingest.apply_dsmr_subscription()

    monkeypatch.setattr(dsmr_ingest, "get_session_local", lambda: (_ for _ in ()).throw(AssertionError()))
    stale("error")  # type: ignore[operator]