home-automation/tests/test_dsmr_source_migration.py

"""Isolated revision-14/15 fixtures for the DSMR source-adoption migration."""

from __future__ import annotations

import json
from datetime import datetime, timedelta, timezone
from pathlib import Path

import pytest
import sqlalchemy.exc
from alembic import command
from alembic.config import Config
from sqlalchemy import create_engine, event, inspect, text

from app.integrations.meter_sources import sanitize_source_config, validate_source_config


def _config(database_url: str) -> Config:
    config = Config("alembic_app.ini")
    config.set_main_option("sqlalchemy.url", database_url)
    return config


def _engine(database_url: str):
    engine = create_engine(database_url, connect_args={"check_same_thread": False})

    @event.listens_for(engine, "connect")
    def _foreign_keys(connection, _record) -> None:
        connection.execute("PRAGMA foreign_keys=ON")

    return engine


def _insert_meter(connection, label: str, started: datetime, ended: datetime | None) -> int:
    connection.execute(
        text(
            "INSERT INTO meter (uuid, label, commodity, started_at, ended_at, reason, note, created_at) "
            "VALUES (:uuid, :label, 'electricity', :started, :ended, 'initial', NULL, :started)"
        ),
        {"uuid": f"{label:0<8}-0000-4000-8000-000000000000", "label": label,
         "started": started, "ended": ended},
    )
    return int(connection.execute(text("SELECT last_insert_rowid()")).scalar_one())


def _insert_contract_version(connection, start: datetime) -> int:
    connection.execute(
        text(
            "INSERT INTO energy_contract (name, kind, active, currency, created_at, updated_at) "
            "VALUES ('Historic contract', 'manual', 1, 'EUR', :at, :at)"
        ),
        {"at": start},
    )
    contract_id = int(connection.execute(text("SELECT last_insert_rowid()")).scalar_one())
    connection.execute(
        text(
            "INSERT INTO energy_contract_version "
            "(contract_id, effective_from, effective_to, \"values\", created_at) "
            "VALUES (:contract_id, :at, NULL, :values, :at)"
        ),
        {"contract_id": contract_id, "at": start, "values": json.dumps({"historic": True})},
    )
    return int(connection.execute(text("SELECT last_insert_rowid()")).scalar_one())


def _insert_cost(
    connection,
    period_start: datetime,
    meter_id: int,
    contract_version_id: int,
    sequence: int,
) -> None:
    amount = 2.5 + sequence
    connection.execute(
        text(
            "INSERT INTO energy_cost_period "
            "(period_start, d1_kwh, d2_kwh, r1_kwh, r2_kwh, import_cost, export_revenue, net_cost, "
            "currency, pricing, contract_version_id, meter_id, degraded, computed_at) "
            "VALUES (:start, :d1, :d2, :r1, :r2, :import_cost, :export_revenue, :net_cost, "
            "'EUR', :pricing, :contract_version_id, :meter, 0, :computed_at)"
        ),
        {
            "start": period_start,
            "d1": 1.0 + sequence,
            "d2": 2.0 + sequence,
            "r1": 3.0 + sequence,
            "r2": 4.0 + sequence,
            "import_cost": amount,
            "export_revenue": 0.25 + sequence,
            "net_cost": amount - (0.25 + sequence),
            "pricing": json.dumps({"historic": True, "sequence": sequence}),
            "contract_version_id": contract_version_id,
            "meter": meter_id,
            "computed_at": period_start + timedelta(seconds=sequence),
        },
    )


def test_populated_revision_14_adopts_dsmr_history_at_revision_16(tmp_path: Path):
    database_url = f"sqlite:///{tmp_path / 'revision_14_history.db'}"
    config = _config(database_url)
    command.upgrade(config, "20260625_14_meter_uuid")
    start = datetime(2026, 8, 1, tzinfo=timezone.utc)
    engine = _engine(database_url)
    try:
        with engine.begin() as connection:
            connection.execute(
                text("INSERT INTO app_config (key, value, updated_at) VALUES (:key, :value, :at)"),
                [
                    {"key": "DSMR_INGEST_ENABLED", "value": "true", "at": start},
                    {"key": "DSMR_MQTT_TOPIC", "value": "historic/dsmr", "at": start},
                    {"key": "DSMR_TARIFF_TOPIC", "value": "historic/tariff", "at": start},
                    {"key": "DSMR_SAMPLE_INTERVAL_S", "value": "0", "at": start},
                    {"key": "MQTT_BROKER_HOST", "value": "mqtt.example.invalid", "at": start},
                    {"key": "MQTT_BROKER_PORT", "value": "1884", "at": start},
                    {"key": "MQTT_USERNAME", "value": "historic-user", "at": start},
                    {"key": "MQTT_PASSWORD", "value": "historic-password", "at": start},
                    {"key": "MQTT_TLS_ENABLED", "value": "true", "at": start},
                    {"key": "UNRELATED_CONFIG", "value": "untouched", "at": start},
                ],
            )
            config_before = dict(connection.execute(text("SELECT key, value FROM app_config")).all())
            for offset, telegram_id in ((0, 77), (20, 78), (40, 77)):
                connection.execute(
                    text("INSERT INTO dsmr_reading (recorded_at, source_id, payload) VALUES (:at, :id, :payload)"),
                    {"at": start + timedelta(minutes=offset), "id": telegram_id,
                     "payload": json.dumps({"id": telegram_id, "keep": f"payload-{offset}"})},
                )
            first = _insert_meter(connection, "meterone", start - timedelta(hours=1), start + timedelta(minutes=20))
            second = _insert_meter(connection, "metertwo", start + timedelta(minutes=20), start + timedelta(minutes=40))
            third = _insert_meter(connection, "meterthree", start + timedelta(minutes=40), None)
            _insert_meter(connection, "nodata", start + timedelta(days=1), None)
            contract_version = _insert_contract_version(connection, start - timedelta(days=1))
            # One normal period per epoch plus a period ending exactly at each
            # replacement boundary.  Meter/binding intervals are half-open, so
            # the latter must remain unbound/degraded.
            _insert_cost(connection, start, first, contract_version, 0)
            _insert_cost(connection, start + timedelta(minutes=5), first, contract_version, 1)
            _insert_cost(connection, start + timedelta(minutes=21), second, contract_version, 2)
            _insert_cost(connection, start + timedelta(minutes=25), second, contract_version, 3)
            _insert_cost(connection, start + timedelta(minutes=41), third, contract_version, 4)
            cost_before = connection.execute(
                text(
                    "SELECT id, period_start, d1_kwh, d2_kwh, r1_kwh, r2_kwh, import_cost, "
                    "export_revenue, net_cost, currency, pricing, contract_version_id, meter_id, "
                    "degraded, computed_at FROM energy_cost_period ORDER BY period_start"
                )
            ).mappings().all()
    finally:
        engine.dispose()

    command.upgrade(config, "20260822_16_dsmr_source_adoption")
    engine = _engine(database_url)
    try:
        with engine.connect() as connection:
            assert connection.execute(text("SELECT version_num FROM alembic_version")).scalar_one() == (
                "20260822_16_dsmr_source_adoption"
            )
            assert connection.execute(text("SELECT COUNT(*) FROM dsmr_reading")).scalar_one() == 3
            source = connection.execute(
                text("SELECT id, enabled, config FROM meter_source WHERE kind = 'dsmr_mqtt'")
            ).one()
            assert source.enabled == 1
            source_config = json.loads(source.config)
            assert source_config == {
                "broker_host": "mqtt.example.invalid", "broker_port": 1884,
                "username": "historic-user", "password": "historic-password", "tls_enabled": True,
                "topic": "historic/dsmr", "tariff_topic": "historic/tariff", "sample_interval_s": 0,
            }
            assert validate_source_config("dsmr_mqtt", source_config) == source_config
            assert sanitize_source_config("dsmr_mqtt", source_config)["sample_interval_s"] == 0
            assert connection.execute(text("SELECT value FROM app_config WHERE key = 'DSMR_MQTT_TOPIC'")).scalar_one() == "historic/dsmr"
            assert dict(connection.execute(text("SELECT key, value FROM app_config")).all()) == config_before
            assert connection.execute(text("SELECT group_concat(telegram_id) FROM dsmr_reading")).scalar_one() == "77,78,77"
            assert connection.execute(text("SELECT payload FROM dsmr_reading ORDER BY recorded_at")).scalars().all() == [
                json.dumps({"id": 77, "keep": "payload-0"}),
                json.dumps({"id": 78, "keep": "payload-20"}),
                json.dumps({"id": 77, "keep": "payload-40"}),
            ]
            assert connection.execute(text("SELECT COUNT(*) FROM meter_source_binding")).scalar_one() == 3
            assert connection.execute(
                text("SELECT COUNT(*) FROM meter_source_binding WHERE meter_id = (SELECT id FROM meter WHERE label = 'nodata')")
            ).scalar_one() == 0
            periods = connection.execute(
                text(
                    "SELECT id, period_start, d1_kwh, d2_kwh, r1_kwh, r2_kwh, import_cost, "
                    "export_revenue, net_cost, currency, pricing, contract_version_id, meter_id, "
                    "degraded, computed_at, source_binding_id FROM energy_cost_period ORDER BY period_start"
                )
            ).mappings().all()
            assert [
                {key: value for key, value in period.items() if key not in {"degraded", "source_binding_id"}}
                for period in periods
            ] == [
                {key: value for key, value in period.items() if key != "degraded"}
                for period in cost_before
            ]
            assert [(period["degraded"], period["source_binding_id"] is not None) for period in periods] == [
                (0, True), (1, False), (0, True), (1, False), (0, True)
            ]
            assert [period["period_start"] for period in periods if period["source_binding_id"] is None] == [
                (start + timedelta(minutes=5)).isoformat(sep=" "),
                (start + timedelta(minutes=25)).isoformat(sep=" "),
            ]
            bound_meter_ids = connection.execute(
                text(
                    "SELECT binding.meter_id FROM energy_cost_period AS period "
                    "LEFT JOIN meter_source_binding AS binding "
                    "ON binding.id = period.source_binding_id ORDER BY period.period_start"
                )
            ).scalars().all()
            assert bound_meter_ids == [first, None, second, None, third]
            assert connection.execute(text("PRAGMA foreign_key_check")).all() == []
    finally:
        engine.dispose()


def test_dsmr_source_timestamp_uniqueness_allows_two_sources(tmp_path: Path):
    database_url = f"sqlite:///{tmp_path / 'two_sources.db'}"
    config = _config(database_url)
    command.upgrade(config, "20260822_16_dsmr_source_adoption")
    engine = _engine(database_url)
    timestamp = datetime(2026, 8, 1, tzinfo=timezone.utc)
    try:
        with engine.begin() as connection:
            first_source = connection.execute(text("SELECT id FROM meter_source WHERE kind = 'dsmr_mqtt'")).scalar_one()
            assert connection.execute(
                text("SELECT enabled FROM meter_source WHERE id = :id"), {"id": first_source}
            ).scalar_one() == 0
            connection.execute(
                text(
                    "INSERT INTO meter_source (uuid, name, kind, enabled, config, status, created_at, updated_at) "
                    "VALUES ('22222222-2222-4222-8222-222222222222', 'Second DSMR', 'dsmr_mqtt', 0, '{}', "
                    "'unknown', :at, :at)"
                ), {"at": timestamp},
            )
            second_source = connection.execute(text("SELECT last_insert_rowid()")).scalar_one()
            for source_id in (first_source, second_source):
                connection.execute(
                    text(
                        "INSERT INTO dsmr_reading (recorded_at, telegram_id, meter_source_id, payload) "
                        "VALUES (:at, 9, :source, '{}')"
                    ), {"at": timestamp, "source": source_id},
                )
            with pytest.raises(sqlalchemy.exc.IntegrityError):
                connection.execute(
                    text(
                        "INSERT INTO dsmr_reading (recorded_at, telegram_id, meter_source_id, payload) "
                        "VALUES (:at, 10, :source, '{}')"
                    ), {"at": timestamp, "source": first_source},
                )
    finally:
        engine.dispose()
    inspector = inspect(create_engine(database_url))
    assert ("meter_source_id", "recorded_at") in {
        tuple(item["column_names"]) for item in inspector.get_unique_constraints("dsmr_reading")
    }


def test_revision_15_without_legacy_dsmr_config_creates_unconfigured_source(tmp_path: Path):
    database_url = f"sqlite:///{tmp_path / 'revision_15_no_dsmr_config.db'}"
    config = _config(database_url)
    command.upgrade(config, "20260822_15_meter_sources")
    engine = _engine(database_url)
    try:
        with engine.begin() as connection:
            connection.execute(
                text(
                    "INSERT INTO app_config (key, value, updated_at) "
                    "VALUES ('UNRELATED_CONFIG', 'untouched', :at)"
                ),
                {"at": datetime(2026, 8, 1, tzinfo=timezone.utc)},
            )
    finally:
        engine.dispose()

    command.upgrade(config, "20260822_16_dsmr_source_adoption")
    engine = _engine(database_url)
    try:
        with engine.connect() as connection:
            source = connection.execute(
                text("SELECT enabled, config FROM meter_source WHERE kind = 'dsmr_mqtt'")
            ).one()
            assert source.enabled == 0
            assert json.loads(source.config) == {}
            assert dict(connection.execute(text("SELECT key, value FROM app_config")).all()) == {
                "UNRELATED_CONFIG": "untouched"
            }
    finally:
        engine.dispose()