M8-R03: hand off channel binding during meter swap

tests/test_meter_source_services.py

@@ -23,6 +23,7 @@ from app.services.meter_sources import (
     SourceDeleteRestrictedError,
     close_binding,
     create_binding,
+    create_binding_for_meter_swap,
     create_source,
     delete_source,
     upsert_discovered_channel,
@@ -178,6 +179,131 @@ def test_binding_rejects_incompatible_unit_and_close_keeps_transaction_open(sess
     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 = 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, "warmtelink_serial", "GJ")
+    old_binding = create_binding(
+        session, meter_id=old_meter.id, channel_id=channel.id, started_at=start
+    )
+    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 = boundary
+    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)
+    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_source_delete_is_restricted_by_discovered_channel(session):
     source, _ = _source_and_channel(session, "dsmr_mqtt", "kWh")
     with pytest.raises(SourceDeleteRestrictedError):