"""M8-T02 tests for source profiles and transaction-owned binding services.""" from __future__ import annotations from datetime import UTC, datetime, timedelta import pytest from sqlalchemy import create_engine, event from sqlalchemy.orm import Session from app.integrations.meter_sources import ( SECRET_MASK, SourceProfileError, merge_source_config, sanitize_source_config, validate_source_config, ) from app.models.energy import Meter from app.models.meter_source import MeterSource, MeterSourceBinding, MeterSourceChannel from app.services.meter_sources import ( BindingOverlapError, BindingValidationError, SourceDeleteRestrictedError, close_binding, create_binding, create_binding_for_meter_swap, create_source, delete_source, transfer_binding, upsert_discovered_channel, ) @pytest.mark.parametrize( ("kind", "config", "expected"), [ ("dsmr_mqtt", {}, {"broker_port": 1883, "topic": "dsmr/json", "sample_interval_s": 10}), ( "warmtelink_serial", {"path": "/dev/warmtelink"}, {"baudrate": 115200, "data_bits": 7, "parity": "N", "stop_bits": 1}, ), ], ) def test_profiles_fill_defaults(kind, config, expected): validated = validate_source_config(kind, config) assert {key: validated[key] for key in expected} == expected @pytest.mark.parametrize( ("kind", "config"), [ ("warmtelink_serial", {"path": "/tmp/warmtelink"}), ("warmtelink_serial", {"path": "/dev/warmtelink", "data_bits": 8}), ("dsmr_mqtt", {"extra": True}), ], ) def test_profile_rejects_invalid_or_unknown_config(kind, config): with pytest.raises(SourceProfileError): validate_source_config(kind, config) def test_secret_sanitize_and_mask_merge_keep_old_value(): original = validate_source_config("dsmr_mqtt", {"password": "not-for-response"}) assert sanitize_source_config("dsmr_mqtt", original)["password"] == SECRET_MASK merged = merge_source_config("dsmr_mqtt", original, {"password": SECRET_MASK, "topic": "new/topic"}) assert merged["password"] == "not-for-response" assert merged["topic"] == "new/topic" @pytest.fixture() def session(tmp_path): engine = create_engine(f"sqlite:///{tmp_path / 'source_services.db'}") @event.listens_for(engine, "connect") def _enable_foreign_keys(dbapi_connection, _connection_record): dbapi_connection.execute("PRAGMA foreign_keys=ON") from app.db import Base Base.metadata.create_all(engine) with Session(engine) as db_session: yield db_session engine.dispose() def _meter(session: Session, commodity: str, label: str = "Meter") -> Meter: timestamp = datetime(2026, 8, 22, tzinfo=UTC) meter = Meter( label=label, commodity=commodity, started_at=timestamp, reason="initial", created_at=timestamp, ) session.add(meter) session.flush() return meter def _source_and_channel(session: Session, kind: str, unit: str) -> tuple[MeterSource, MeterSourceChannel]: config = {"path": "/dev/warmtelink"} if kind == "warmtelink_serial" else {} source = create_source(session, name="Source", kind=kind, config=config) session.flush() channel = upsert_discovered_channel( session, source_id=source.id, channel_key=f"{unit}-total", label="Total", unit=unit, suggested_commodity="heating", ) session.flush() return source, channel def test_channel_upsert_is_idempotent_and_never_auto_binds(session): _, channel = _source_and_channel(session, "warmtelink_serial", "GJ") same = upsert_discovered_channel( session, source_id=channel.source_id, channel_key=channel.channel_key, label="Renamed total", unit="GJ", suggested_commodity="heating", ) session.flush() assert same.id == channel.id assert same.label == "Renamed total" assert session.query(MeterSourceBinding).count() == 0 def test_channel_upsert_rejects_unit_change_when_channel_is_bound(session): meter = _meter(session, "heating") _, channel = _source_and_channel(session, "warmtelink_serial", "GJ") create_binding(session, meter_id=meter.id, channel_id=channel.id, started_at=meter.started_at) session.flush() with pytest.raises(BindingValidationError, match="Cannot change unit of bound channel"): upsert_discovered_channel( session, source_id=channel.source_id, channel_key=channel.channel_key, label="Total", unit="m³", suggested_commodity="hot_water", ) assert channel.unit == "GJ" def test_binding_checks_both_sides_and_allows_equal_boundaries(session): meter_one = _meter(session, "heating", "one") meter_two = _meter(session, "heating", "two") _, channel_one = _source_and_channel(session, "warmtelink_serial", "GJ") _, channel_two = _source_and_channel(session, "warmtelink_serial", "GJ") start = datetime(2026, 8, 22, tzinfo=UTC) boundary = start + timedelta(hours=1) create_binding(session, meter_id=meter_one.id, channel_id=channel_one.id, started_at=start, ended_at=boundary) create_binding(session, meter_id=meter_one.id, channel_id=channel_two.id, started_at=boundary) create_binding(session, meter_id=meter_two.id, channel_id=channel_one.id, started_at=boundary) with pytest.raises(BindingOverlapError): create_binding(session, meter_id=meter_one.id, channel_id=channel_two.id, started_at=start + timedelta(minutes=30)) with pytest.raises(BindingOverlapError): create_binding(session, meter_id=meter_two.id, channel_id=channel_one.id, started_at=start + timedelta(minutes=30)) def test_binding_rejects_incompatible_unit_and_close_keeps_transaction_open(session): meter = _meter(session, "electricity") _, channel = _source_and_channel(session, "warmtelink_serial", "GJ") with pytest.raises(BindingValidationError): create_binding(session, meter_id=meter.id, channel_id=channel.id, started_at=meter.started_at) thermal_meter = _meter(session, "heating", "thermal") binding = create_binding(session, meter_id=thermal_meter.id, channel_id=channel.id, started_at=meter.started_at) session.flush() close_binding(session, binding.id, ended_at=meter.started_at + timedelta(minutes=1)) assert session.in_transaction() session.rollback() assert session.get(MeterSourceBinding, binding.id) is None def test_meter_swap_hands_off_only_the_previous_meter_binding(session): start = datetime(2026, 8, 22, tzinfo=UTC) boundary = start + timedelta(days=1) old_meter = _meter(session, "heating", "old") old_meter.started_at = start old_meter.ended_at = None new_meter = Meter( label="new", commodity="heating", started_at=boundary, reason="meter_swap", created_at=boundary, ) session.add(new_meter) session.flush() _, channel = _source_and_channel(session, "warmtelink_serial", "GJ") old_binding = create_binding( session, meter_id=old_meter.id, channel_id=channel.id, started_at=start ) old_meter.ended_at = boundary session.flush() new_binding = create_binding_for_meter_swap( session, old_meter_id=old_meter.id, new_meter_id=new_meter.id, channel_id=channel.id, started_at=boundary, ) session.flush() assert old_binding.ended_at == boundary assert new_binding.started_at == boundary assert new_binding.ended_at is None def test_meter_swap_rejects_channel_owned_by_a_different_meter(session): start = datetime(2026, 8, 22, tzinfo=UTC) boundary = start + timedelta(days=1) old_meter = _meter(session, "heating", "old") new_meter = _meter(session, "heating", "new") other_meter = _meter(session, "heating", "other") old_meter.started_at = start old_meter.ended_at = boundary new_meter.started_at = boundary _, channel = _source_and_channel(session, "warmtelink_serial", "GJ") create_binding(session, meter_id=other_meter.id, channel_id=channel.id, started_at=start) with pytest.raises(BindingOverlapError, match="cannot be handed off"): create_binding_for_meter_swap( session, old_meter_id=old_meter.id, new_meter_id=new_meter.id, channel_id=channel.id, started_at=boundary, ) def test_meter_swap_rejects_ambiguous_channel_without_closing_any_binding(session): start = datetime(2026, 8, 22, tzinfo=UTC) boundary = start + timedelta(days=1) old_meter = _meter(session, "heating", "old") old_meter.started_at = start old_meter.ended_at = None new_meter = Meter( label="new", commodity="heating", started_at=boundary, reason="meter_swap", created_at=boundary, ) other_meter = _meter(session, "heating", "other") session.add(new_meter) session.flush() _, channel = _source_and_channel(session, "warmtelink_serial", "GJ") old_binding = create_binding(session, meter_id=old_meter.id, channel_id=channel.id, started_at=start) old_meter.ended_at = boundary session.add( MeterSourceBinding( meter_id=other_meter.id, channel_id=channel.id, started_at=start, created_at=start, updated_at=start, ) ) session.flush() with pytest.raises(BindingOverlapError, match="occupied or has an ambiguous binding"): create_binding_for_meter_swap( session, old_meter_id=old_meter.id, new_meter_id=new_meter.id, channel_id=channel.id, started_at=boundary, ) assert old_binding.ended_at is None def test_meter_swap_rejects_incompatible_channel(session): start = datetime(2026, 8, 22, tzinfo=UTC) boundary = start + timedelta(days=1) old_meter = _meter(session, "heating", "old") old_meter.started_at = start old_meter.ended_at = boundary new_meter = Meter( label="new", commodity="heating", started_at=boundary, reason="meter_swap", created_at=boundary, ) session.add(new_meter) session.flush() _, channel = _source_and_channel(session, "dsmr_mqtt", "kWh") with pytest.raises(BindingValidationError, match="requires unit"): create_binding_for_meter_swap( session, old_meter_id=old_meter.id, new_meter_id=new_meter.id, channel_id=channel.id, started_at=boundary, ) def test_transfer_closes_and_opens_at_shared_boundary(session): start = datetime(2026, 8, 22, tzinfo=UTC) meter = _meter(session, "heating") meter.started_at = start _, old_channel = _source_and_channel(session, "warmtelink_serial", "GJ") _, new_channel = _source_and_channel(session, "warmtelink_serial", "GJ") old = create_binding(session, meter_id=meter.id, channel_id=old_channel.id, started_at=start) session.flush() closed, created = transfer_binding( session, target_meter_id=meter.id, from_binding_id=old.id, to_channel_id=new_channel.id, effective_at=start + timedelta(hours=1), ) assert closed.ended_at == created.started_at == start + timedelta(hours=1) assert created.channel_id == new_channel.id def test_transfer_rejects_future_without_mutating_source_binding(session): start = datetime.now(UTC) - timedelta(hours=2) meter = _meter(session, "heating") meter.started_at = start _, old_channel = _source_and_channel(session, "warmtelink_serial", "GJ") _, new_channel = _source_and_channel(session, "warmtelink_serial", "GJ") old = create_binding(session, meter_id=meter.id, channel_id=old_channel.id, started_at=start) session.flush() with pytest.raises(BindingValidationError, match="future"): transfer_binding(session, target_meter_id=meter.id, from_binding_id=old.id, to_channel_id=new_channel.id, effective_at=datetime.now(UTC) + timedelta(minutes=1)) assert old.ended_at is None def test_cross_meter_transfer_recovers_stranded_same_channel(session): start = datetime(2026, 8, 20, tzinfo=UTC) boundary = start + timedelta(days=1) old = _meter(session, "heating", "old") old.started_at, old.ended_at = start, None new = Meter(label="new", commodity="heating", started_at=boundary, reason="meter_swap", created_at=boundary) session.add(new) session.flush() _, channel = _source_and_channel(session, "warmtelink_serial", "GJ") stranded = create_binding(session, meter_id=old.id, channel_id=channel.id, started_at=start) old.ended_at = boundary session.flush() closed, created = transfer_binding( session, target_meter_id=new.id, from_binding_id=stranded.id, to_channel_id=channel.id, effective_at=boundary + timedelta(hours=2), ) assert closed.ended_at == boundary assert created.started_at == boundary + timedelta(hours=2) assert created.channel_id == channel.id def test_cross_meter_transfer_recovers_unique_gapped_predecessor(session): """A deliberate no-meter gap does not make the latest predecessor ambiguous.""" start = datetime(2026, 8, 20, tzinfo=UTC) old_end = start + timedelta(days=1) target_start = old_end + timedelta(hours=3) old = _meter(session, "heating", "old") old.started_at, old.ended_at = start, old_end target = Meter(label="target", commodity="heating", started_at=target_start, reason="initial", created_at=target_start) session.add(target) session.flush() _, channel = _source_and_channel(session, "warmtelink_serial", "GJ") stranded = MeterSourceBinding( meter_id=old.id, channel_id=channel.id, started_at=start, created_at=start, updated_at=start, ) session.add(stranded) session.flush() closed, created = transfer_binding( session, target_meter_id=target.id, from_binding_id=stranded.id, to_channel_id=channel.id, effective_at=target_start + timedelta(hours=1), ) assert closed.ended_at == old_end assert created.started_at == target_start + timedelta(hours=1) def test_cross_meter_transfer_rejects_closed_source_and_ambiguous_predecessor(session): start = datetime(2026, 8, 20, tzinfo=UTC) boundary = start + timedelta(days=1) old = _meter(session, "heating", "old") old.started_at, old.ended_at = start, None target = Meter(label="target", commodity="heating", started_at=boundary, reason="meter_swap", created_at=boundary) session.add(target) session.flush() _, channel = _source_and_channel(session, "warmtelink_serial", "GJ") source = create_binding(session, meter_id=old.id, channel_id=channel.id, started_at=start) old.ended_at = boundary session.flush() close_binding(session, source.id, ended_at=boundary - timedelta(hours=1)) with pytest.raises(BindingValidationError, match="open binding"): transfer_binding(session, target_meter_id=target.id, from_binding_id=source.id, to_channel_id=channel.id, effective_at=boundary) assert source.ended_at == boundary - timedelta(hours=1) source.ended_at = None # synthetic retained bad row, exactly the recovery input. duplicate = Meter(label="duplicate", commodity="heating", started_at=start + timedelta(hours=1), ended_at=boundary, reason="other", created_at=start) session.add(duplicate) session.flush() with pytest.raises(BindingValidationError, match="unique immediately preceding"): transfer_binding(session, target_meter_id=target.id, from_binding_id=source.id, to_channel_id=channel.id, effective_at=boundary) assert source.ended_at is None def test_source_delete_is_restricted_by_discovered_channel(session): source, _ = _source_and_channel(session, "dsmr_mqtt", "kWh") with pytest.raises(SourceDeleteRestrictedError): delete_source(session, source.id)