home-automation/tests/test_warmtelink_ingest.py

"""Behaviour tests for the WarmteLink per-source admission state machine."""

from __future__ import annotations

from dataclasses import replace
from datetime import UTC, datetime
from decimal import Decimal
from pathlib import Path
from types import SimpleNamespace

import pytest
from alembic import command
from alembic.config import Config
from sqlalchemy import create_engine, func, select
from sqlalchemy.orm import Session, sessionmaker

from app.integrations.p1 import IntegrityStatus, ObisField, P1Channel, P1Telegram, dsmr_crc16, parse_telegram
from app.models.meter_source import MeterSource, MeterSourceBinding, MeterSourceChannel, WarmteLinkReading
from app.services.warmtelink_ingest import WarmteLinkIngestor


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


@pytest.fixture()
def database(tmp_path: Path):
    url = f"sqlite:///{tmp_path / 'warmtelink_ingest.db'}"
    command.upgrade(_config(url), "head")
    engine = create_engine(url, connect_args={"check_same_thread": False})
    factory = sessionmaker(bind=engine, autoflush=False, class_=Session)
    with factory() as session:
        for name in ("one", "two"):
            now = datetime(2026, 8, 22, tzinfo=UTC)
            session.add(
                MeterSource(
                    name=name,
                    kind="warmtelink_serial",
                    enabled=True,
                    config={"path": f"/dev/{name}", "baudrate": 115200, "data_bits": 7, "parity": "N", "stop_bits": 1},
                    created_at=now,
                    updated_at=now,
                )
            )
        session.commit()
        ids = list(session.scalars(select(MeterSource.id).order_by(MeterSource.id)))
    yield engine, factory, ids[-2:]
    engine.dispose()


def _telegram(
    *,
    second: int,
    minute: int = 0,
    integrity: IntegrityStatus = IntegrityStatus.UNVERIFIABLE,
    water: str = "5.900",
    heat: str = "0.017",
    channels: tuple[int, ...] = (1, 2),
    device_type: str = "006",
    fingerprint: str | None = "a" * 64,
    channel_fingerprint: str | None = None,
    channel_fingerprints: dict[int, str | None] | None = None,
) -> P1Telegram:
    fields: list[ObisField] = []
    parsed_channels: list[P1Channel] = []
    for number in channels:
        value, unit = (water, "m3") if number == 1 else (heat, "GJ")
        code = f"0-{number}:24.2.1"
        field = ObisField(code, (value + "*" + unit,), Decimal(value), unit)
        fields.append(field)
        parsed_channels.append(
            P1Channel(
                number,
                device_type if number == 1 else "012",
                (
                    channel_fingerprints[number]
                    if channel_fingerprints is not None
                    else fingerprint if channel_fingerprint is None else channel_fingerprint
                ),
                (field,),
            )
        )
    return P1Telegram(
        100,
        integrity,
        "fixture",
        f"26082212{minute:02d}{second:02d}S",
        fingerprint,
        tuple(fields),
        tuple(parsed_channels),
    )


def _counts(factory) -> tuple[int, int]:
    with factory() as session:
        return (
            # Revision 16 supplies one DSMR channel; this task must not touch it.
            int(session.scalar(select(func.count(MeterSourceChannel.id))) or 0) - 1,
            int(session.scalar(select(func.count(WarmteLinkReading.id))) or 0),
        )


def _all_counts(factory) -> tuple[int, int, int]:
    channels, readings = _counts(factory)
    with factory() as session:
        bindings = int(session.scalar(select(func.count(MeterSourceBinding.id))) or 0)
    return channels, readings, bindings


def _fixed_clock() -> datetime:
    return datetime(2026, 8, 22, 10, tzinfo=UTC)


def test_unverifiable_requires_two_continuous_frames_and_preserves_true_metadata(database):
    _engine, factory, (source_id, _) = database
    ingest = WarmteLinkIngestor(clock=_fixed_clock)
    with factory() as session:
        assert not ingest.ingest(session, source_id=source_id, telegram=_telegram(second=0))
        session.commit()
    assert _counts(factory) == (0, 0)
    with factory() as session:
        assert ingest.ingest(session, source_id=source_id, telegram=_telegram(second=10, water="5.901"))
        assert ingest.ingest(session, source_id=source_id, telegram=_telegram(second=20, water="5.902"))
        session.commit()
    assert _counts(factory) == (2, 2)
    with factory() as session:
        channels = list(
            session.scalars(
                select(MeterSourceChannel)
                .where(MeterSourceChannel.source_id == source_id)
                .order_by(MeterSourceChannel.channel_key)
            )
        )
        assert [(item.channel_key, item.unit, item.device_type, item.latest_quality) for item in channels] == [
            ("channel-1", "m³", "006", "unverifiable"),
            ("channel-2", "GJ", "012", "unverifiable"),
        ]
        assert all(item.fingerprint == "a" * 64 for item in channels)
        assert channels[0].latest_value == Decimal("5.902")


def test_unverifiable_cadence_requires_exactly_ten_seconds_and_restarts_after_gaps(database):
    _engine, factory, (source_id, _) = database
    ingest = WarmteLinkIngestor()
    with factory() as session:
        assert not ingest.ingest(session, source_id=source_id, telegram=_telegram(second=0))
        assert ingest.ingest(session, source_id=source_id, telegram=_telegram(second=10, water="5.901"))
        session.commit()
    assert _counts(factory) == (2, 2)

    restarted = WarmteLinkIngestor()
    with factory() as session:
        assert not restarted.ingest(session, source_id=source_id, telegram=_telegram(second=0, water="5.902"))
        # 20 seconds misses one expected frame; it is only a new candidate.
        assert not restarted.ingest(session, source_id=source_id, telegram=_telegram(second=20, water="5.903"))
        assert restarted.ingest(session, source_id=source_id, telegram=_telegram(second=30, water="5.904"))
        # A larger gap likewise needs its own new matching successor.
        assert not restarted.ingest(session, source_id=source_id, telegram=_telegram(second=0, minute=2, water="5.905"))
        assert restarted.ingest(session, source_id=source_id, telegram=_telegram(second=10, minute=2, water="5.906"))
        session.commit()
    assert _counts(factory) == (2, 4)


@pytest.mark.parametrize(
    "first, second",
    [
        (_telegram(second=10), _telegram(second=9)),
        (_telegram(second=0), _telegram(second=30)),
        (_telegram(second=10), _telegram(second=20, water="5.899")),
        (_telegram(second=10), _telegram(second=20, device_type="007")),
        (_telegram(second=10), _telegram(second=20, channels=(1,))),
        (_telegram(second=10, channels=(1,)), _telegram(second=20)),
    ],
)
def test_unverifiable_discontinuities_restart_two_frame_confirmation(database, first, second):
    _engine, factory, (source_id, _) = database
    ingest = WarmteLinkIngestor()
    with factory() as session:
        assert not ingest.ingest(session, source_id=source_id, telegram=first)
        assert not ingest.ingest(session, source_id=source_id, telegram=second)
        session.commit()
    assert _counts(factory) == (0, 0)
    with factory() as session:
        source = session.get(MeterSource, source_id)
        assert source is not None and source.status == "error" and source.last_error is not None


def test_discontinuity_candidate_needs_a_new_matching_successor(database):
    _engine, factory, (source_id, _) = database
    ingest = WarmteLinkIngestor()
    with factory() as session:
        assert not ingest.ingest(session, source_id=source_id, telegram=_telegram(second=0))
        assert not ingest.ingest(session, source_id=source_id, telegram=_telegram(second=10, channels=(1,)))
        # The restored channel set is a fresh candidate, not an admission.
        assert not ingest.ingest(session, source_id=source_id, telegram=_telegram(second=20))
        assert ingest.ingest(session, source_id=source_id, telegram=_telegram(second=30, water="5.901"))
        session.commit()
    assert _counts(factory) == (2, 2)


@pytest.mark.parametrize(
    "first, discontinuous, successor",
    [
        (_telegram(second=0), _telegram(second=10, water="5.899"), _telegram(second=20, water="5.900")),
        (_telegram(second=0), _telegram(second=10, fingerprint="b" * 64), _telegram(second=20, fingerprint="b" * 64)),
        (_telegram(second=0), _telegram(second=10, channels=(1,)), _telegram(second=20, channels=(1,))),
        (_telegram(second=0, channels=(1,)), _telegram(second=10), _telegram(second=20)),
    ],
)
def test_each_metadata_or_channel_discontinuity_requires_its_own_two_frames(
    database, first, discontinuous, successor
):
    _engine, factory, (source_id, _) = database
    ingest = WarmteLinkIngestor()
    with factory() as session:
        assert not ingest.ingest(session, source_id=source_id, telegram=first)
        # The discontinuous frame becomes a new first candidate, never an admission.
        assert not ingest.ingest(session, source_id=source_id, telegram=discontinuous)
        assert ingest.ingest(session, source_id=source_id, telegram=successor)
        session.commit()


def test_parser_failure_clears_only_its_source_candidate_and_requires_two_new_frames(database):
    _engine, factory, (one, two) = database
    ingest = WarmteLinkIngestor()
    with factory() as session:
        assert not ingest.ingest(session, source_id=one, telegram=_telegram(second=0))
        assert not ingest.ingest(session, source_id=two, telegram=_telegram(second=0))
        session.commit()

    def parse_failure(_frame: bytes) -> P1Telegram:
        raise ValueError("raw telegram must never be reported")

    assert not ingest.handle_frame(one, b"private", session_factory=factory, parser=parse_failure)
    with factory() as session:
        # One's candidate was cleared; two's independent candidate survives.
        assert not ingest.ingest(session, source_id=one, telegram=_telegram(second=10, water="5.901"))
        assert ingest.ingest(session, source_id=two, telegram=_telegram(second=10, water="5.901"))
        assert ingest.ingest(session, source_id=one, telegram=_telegram(second=20, water="5.902"))
        session.commit()
    assert _counts(factory) == (4, 4)


def _normalization_failure(kind: str) -> P1Telegram:
    telegram = _telegram(second=10)
    if kind == "no-channels":
        return replace(telegram, channels=())
    first, *rest = telegram.channels
    if kind == "empty-readings":
        return replace(telegram, channels=(replace(first, readings=()), *rest))
    if kind == "ambiguous-readings":
        return replace(telegram, channels=(replace(first, readings=(first.readings[0],) * 2), *rest))
    if kind == "incomplete-reading":
        return replace(
            telegram,
            channels=(replace(first, readings=(replace(first.readings[0], value=None),)), *rest),
        )
    invalid_values: dict[str, object] = {
        "decimal-nan": Decimal("NaN"),
        "decimal-positive-infinity": Decimal("Infinity"),
        "decimal-negative-infinity": Decimal("-Infinity"),
        "float": 5.9,
        "string": "private-invalid-cumulative-value",
    }
    if kind in invalid_values:
        return replace(
            telegram,
            channels=(
                replace(first, readings=(replace(first.readings[0], value=invalid_values[kind]),)),
                *rest,
            ),
        )
    if kind == "invalid-timestamp":
        return replace(telegram, timestamp="invalid-time")
    raise AssertionError(f"unexpected normalization failure kind: {kind}")


@pytest.mark.parametrize(
    "kind",
    [
        "no-channels",
        "empty-readings",
        "ambiguous-readings",
        "incomplete-reading",
        "decimal-nan",
        "decimal-positive-infinity",
        "decimal-negative-infinity",
        "float",
        "string",
        "invalid-timestamp",
    ],
)
def test_dto_normalization_failure_clears_only_its_source_candidate_and_requires_two_new_frames(
    database, kind
):
    _engine, factory, (one, two) = database
    ingest = WarmteLinkIngestor()
    with factory() as session:
        assert not ingest.ingest(session, source_id=one, telegram=_telegram(second=0))
        assert not ingest.ingest(session, source_id=two, telegram=_telegram(second=0))
        session.commit()

    failed = _normalization_failure(kind)
    assert not ingest.handle_frame(one, b"private", session_factory=factory, parser=lambda _raw: failed)
    # The rejected DTO creates no discovery/latest/history rows, and source two
    # retains its independent first candidate.
    assert _counts(factory) == (0, 0)
    with factory() as session:
        source = session.get(MeterSource, one)
        assert source is not None and source.last_error == "WarmteLink frame could not be normalized"
        assert "private" not in (source.last_error or "")
        assert not ingest.ingest(session, source_id=one, telegram=_telegram(second=10, water="5.901"))
        assert ingest.ingest(session, source_id=two, telegram=_telegram(second=10, water="5.901"))
        assert ingest.ingest(session, source_id=one, telegram=_telegram(second=20, water="5.902"))
        session.commit()
    assert _counts(factory) == (4, 4)


def test_valid_invalid_sampling_replay_restart_and_source_isolation(database):
    _engine, factory, (one, two) = database
    ingest = WarmteLinkIngestor(clock=_fixed_clock)
    with factory() as session:
        assert ingest.ingest(session, source_id=one, telegram=_telegram(second=0, integrity=IntegrityStatus.VALID))
        assert ingest.ingest(session, source_id=one, telegram=_telegram(second=10, integrity=IntegrityStatus.VALID, water="5.901"))
        assert not ingest.ingest(session, source_id=one, telegram=_telegram(second=20, integrity=IntegrityStatus.INVALID))
        # Source two cannot consume source one's candidate.
        assert not ingest.ingest(session, source_id=one, telegram=_telegram(second=30))
        assert not ingest.ingest(session, source_id=two, telegram=_telegram(second=40))
        session.commit()
    assert _counts(factory) == (2, 2)  # candidates are never discovered; first minute only once
    # A new process sees the minute bucket in DB and does not create duplicates.
    restarted = WarmteLinkIngestor(clock=_fixed_clock)
    with factory() as session:
        assert restarted.ingest(session, source_id=one, telegram=_telegram(second=50, integrity=IntegrityStatus.VALID))
        session.commit()
    assert _counts(factory) == (2, 2)


def test_history_keeps_first_utc_minute_sample_and_creates_next_minute(database):
    _engine, factory, (source_id, _) = database
    ingest = WarmteLinkIngestor(clock=_fixed_clock)
    with factory() as session:
        assert ingest.ingest(session, source_id=source_id, telegram=_telegram(second=0, integrity=IntegrityStatus.VALID))
        assert ingest.ingest(session, source_id=source_id, telegram=_telegram(second=10, integrity=IntegrityStatus.VALID, water="5.901"))
        assert ingest.ingest(
            session,
            source_id=source_id,
            telegram=_telegram(second=0, minute=1, integrity=IntegrityStatus.VALID, water="5.902"),
        )
        session.commit()
    with factory() as session:
        readings = list(session.scalars(select(WarmteLinkReading).order_by(WarmteLinkReading.recorded_at)))
        assert len(readings) == 4
        assert readings[0].recorded_at.second == 0
        assert readings[2].recorded_at.minute == 1 and readings[2].recorded_at.second == 0
        channel = session.scalar(
            select(MeterSourceChannel).where(
                MeterSourceChannel.source_id == source_id,
                MeterSourceChannel.channel_key == "channel-1",
            )
        )
        assert channel is not None and channel.latest_value == Decimal("5.902")


def test_handle_frame_rolls_back_database_error_marks_source_and_recovers(database, monkeypatch):
    _engine, factory, (source_id, _) = database
    ingest = WarmteLinkIngestor()
    telegram = _telegram(second=0, integrity=IntegrityStatus.VALID)
    import app.services.warmtelink_ingest as module

    original = module.upsert_discovered_channel

    def broken(*_args, **_kwargs):
        raise RuntimeError("database unavailable")

    monkeypatch.setattr(module, "upsert_discovered_channel", broken)
    assert not ingest.handle_frame(source_id, b"private telegram", session_factory=factory, parser=lambda _raw: telegram)
    assert _counts(factory) == (0, 0)
    with factory() as session:
        source = session.get(MeterSource, source_id)
        assert source is not None and source.status == "error"
        assert "private" not in (source.last_error or "")
    monkeypatch.setattr(module, "upsert_discovered_channel", original)
    assert ingest.handle_frame(source_id, b"ignored", session_factory=factory, parser=lambda _raw: telegram)
    assert _counts(factory) == (2, 2)


def test_second_channel_failure_rolls_back_first_channel_latest_and_history(database, monkeypatch):
    _engine, factory, (source_id, _) = database
    ingest = WarmteLinkIngestor()
    import app.services.warmtelink_ingest as module

    original = module.upsert_discovered_channel

    def broken_on_second(*args, **kwargs):
        if kwargs["channel_key"] == "channel-2":
            raise RuntimeError("second channel failed")
        return original(*args, **kwargs)

    monkeypatch.setattr(module, "upsert_discovered_channel", broken_on_second)
    assert not ingest.handle_frame(
        source_id,
        b"private",
        session_factory=factory,
        parser=lambda _raw: _telegram(second=0, integrity=IntegrityStatus.VALID),
    )
    assert _counts(factory) == (0, 0)
    monkeypatch.setattr(module, "upsert_discovered_channel", original)
    assert ingest.handle_frame(
        source_id,
        b"private",
        session_factory=factory,
        parser=lambda _raw: _telegram(second=10, integrity=IntegrityStatus.VALID),
    )
    assert _counts(factory) == (2, 2)


def test_unverifiable_database_failure_clears_only_its_source_sliding_candidate(database, monkeypatch):
    _engine, factory, (one, two) = database
    ingest = WarmteLinkIngestor()
    import app.services.warmtelink_ingest as module

    with factory() as session:
        assert not ingest.ingest(session, source_id=one, telegram=_telegram(second=0))
        assert not ingest.ingest(session, source_id=two, telegram=_telegram(second=0))
        session.commit()

    original = module.upsert_discovered_channel
    monkeypatch.setattr(module, "upsert_discovered_channel", lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError()))
    assert not ingest.handle_frame(
        one, b"private", session_factory=factory, parser=lambda _raw: _telegram(second=10, water="5.901")
    )
    monkeypatch.setattr(module, "upsert_discovered_channel", original)
    assert _all_counts(factory) == (0, 0, 0)

    with factory() as session:
        # One needs a new pair after its failed write; two retains its own candidate.
        assert not ingest.ingest(session, source_id=one, telegram=_telegram(second=20, water="5.902"))
        assert ingest.ingest(session, source_id=two, telegram=_telegram(second=10, water="5.901"))
        assert ingest.ingest(session, source_id=one, telegram=_telegram(second=30, water="5.903"))
        session.commit()
    assert _all_counts(factory) == (4, 4, 0)


def _valid_crc_frame(*, device_type: str = "006") -> bytes:
    body = (
        b"/WARMTE\r\n"
        b"0-0:1.0.0(260822120000S)\r\n"
        b"0-0:96.1.1(WARMTE-REDACTED)\r\n"
        + f"0-1:24.1.0({device_type})\r\n".encode()
        + b"0-1:96.1.0(KAM-REDACTED)\r\n"
        + b"0-1:24.2.1(260822120000S)(5.900*m3)\r\n"
        + b"0-2:24.1.0(012)\r\n"
        + b"0-2:96.1.0(KAM-REDACTED)\r\n"
        + b"0-2:24.2.1(260822120000S)(0.017*GJ)\r\n"
    )
    payload = body + b"!"
    return payload + f"{dsmr_crc16(payload):04X}".encode() + b"\r\n"


def test_non_enum_integrity_and_fake_metadata_reject_whole_frame_per_source(database):
    _engine, factory, (one, two) = database
    ingest = WarmteLinkIngestor()
    with factory() as session:
        assert not ingest.ingest(session, source_id=one, telegram=_telegram(second=0))
        assert not ingest.ingest(session, source_id=two, telegram=_telegram(second=0))
        session.commit()

    forged_quality = replace(_telegram(second=10, integrity=IntegrityStatus.VALID), integrity=SimpleNamespace(value="valid"))
    assert not ingest.handle_frame(one, b"private", session_factory=factory, parser=lambda _raw: forged_quality)
    mixed = replace(_telegram(second=10), channels=(replace(_telegram(second=10).channels[0], device_type="raw-device"),))
    assert not ingest.handle_frame(one, b"private", session_factory=factory, parser=lambda _raw: mixed)
    assert _all_counts(factory) == (0, 0, 0)

    with factory() as session:
        source = session.get(MeterSource, one)
        assert source is not None and source.last_error == "WarmteLink frame could not be normalized"
        # Both malformed frames clear one, while two's candidate is untouched.
        assert not ingest.ingest(session, source_id=one, telegram=_telegram(second=20, water="5.901"))
        assert ingest.ingest(session, source_id=two, telegram=_telegram(second=10, water="5.901"))
        assert ingest.ingest(session, source_id=one, telegram=_telegram(second=30, water="5.902"))
        session.commit()
    assert _all_counts(factory) == (4, 4, 0)


def test_valid_crc_parser_frame_with_raw_device_type_is_rejected_without_leakage(database):
    _engine, factory, (one, two) = database
    ingest = WarmteLinkIngestor()
    with factory() as session:
        assert not ingest.ingest(session, source_id=one, telegram=_telegram(second=0))
        assert not ingest.ingest(session, source_id=two, telegram=_telegram(second=0))
        session.commit()

    raw_identifier = "serial-number-which-must-not-persist"
    assert parse_telegram(_valid_crc_frame(device_type=raw_identifier)).integrity is IntegrityStatus.VALID
    assert not ingest.handle_frame(one, _valid_crc_frame(device_type=raw_identifier), session_factory=factory)
    assert _all_counts(factory) == (0, 0, 0)
    with factory() as session:
        source = session.get(MeterSource, one)
        assert source is not None and source.last_error == "WarmteLink frame could not be normalized"
        assert raw_identifier not in (source.last_error or "")
        assert not ingest.ingest(session, source_id=one, telegram=_telegram(second=10, water="5.901"))
        assert ingest.ingest(session, source_id=two, telegram=_telegram(second=10, water="5.901"))
        assert ingest.ingest(session, source_id=one, telegram=_telegram(second=20, water="5.902"))
        session.commit()
    assert _all_counts(factory) == (4, 4, 0)


@pytest.mark.parametrize(
    "fingerprint, channel_fingerprint",
    [
        (None, None),
        ("", ""),
        ("a" * 63, "a" * 63),
        ("g" * 64, "g" * 64),
        ("serial-number-which-must-not-persist", "serial-number-which-must-not-persist"),
    ],
)
def test_invalid_fingerprint_fails_closed_without_persisting_identifiers(
    database, fingerprint, channel_fingerprint
):
    _engine, factory, (source_id, _) = database
    raw_identifier = "serial-number-which-must-not-persist"
    ingest = WarmteLinkIngestor()
    with factory() as session:
        assert not ingest.ingest(session, source_id=source_id, telegram=_telegram(second=0, integrity=IntegrityStatus.INVALID))
        assert not ingest.ingest(
            session,
            source_id=source_id,
            telegram=_telegram(
                second=10,
                integrity=IntegrityStatus.VALID,
                fingerprint=fingerprint,
                channel_fingerprint=channel_fingerprint,
            ),
        )
        session.commit()
    assert _counts(factory) == (0, 0)
    with factory() as session:
        source = session.get(MeterSource, source_id)
        assert source is not None and source.last_error == "WarmteLink frame fingerprint is invalid"
        assert raw_identifier not in (source.last_error or "")
        assert ingest.ingest(session, source_id=source_id, telegram=_telegram(second=20, integrity=IntegrityStatus.VALID))
        session.commit()
        rendered = " ".join(str(row) for row in session.execute(select(WarmteLinkReading)).all())
        assert raw_identifier not in rendered
        assert all("raw" not in column.name for column in WarmteLinkReading.__table__.columns)


@pytest.mark.parametrize(
    "invalid_fingerprint",
    [None, "", "a" * 63, "serial-number-which-must-not-persist"],
    ids=["none", "empty", "wrong-length", "raw-like"],
)
@pytest.mark.parametrize("invalid_channel", [1, 2])
def test_invalid_channel_fingerprint_rejects_whole_frame_and_clears_candidate(
    database, invalid_fingerprint, invalid_channel
):
    _engine, factory, (source_id, _) = database
    raw_identifier = "serial-number-which-must-not-persist"
    valid = "a" * 64
    ingest = WarmteLinkIngestor()
    with factory() as session:
        # Establish a candidate, then prove an invalid channel clears only it.
        assert not ingest.ingest(session, source_id=source_id, telegram=_telegram(second=0))
        assert not ingest.ingest(
            session,
            source_id=source_id,
            telegram=_telegram(
                second=10,
                fingerprint=valid,
                channel_fingerprints={1: valid, 2: valid} | {invalid_channel: invalid_fingerprint},
            ),
        )
        session.commit()
    assert _counts(factory) == (0, 0)
    with factory() as session:
        source = session.get(MeterSource, source_id)
        assert source is not None and source.last_error == "WarmteLink frame fingerprint is invalid"
        assert raw_identifier not in (source.last_error or "")
        # The first legal frame is a fresh candidate; only its successor admits.
        assert not ingest.ingest(session, source_id=source_id, telegram=_telegram(second=20, water="5.901"))
        assert ingest.ingest(session, source_id=source_id, telegram=_telegram(second=30, water="5.902"))
        session.commit()
    assert _counts(factory) == (2, 2)


@pytest.mark.parametrize(
    "invalid_fingerprint",
    [None, "", "a" * 63, "serial-number-which-must-not-persist"],
    ids=["none", "empty", "wrong-length", "raw-like"],
)
def test_invalid_top_fingerprint_rejects_whole_frame_and_clears_candidate(database, invalid_fingerprint):
    _engine, factory, (source_id, _) = database
    raw_identifier = "serial-number-which-must-not-persist"
    valid = "a" * 64
    ingest = WarmteLinkIngestor()
    with factory() as session:
        assert not ingest.ingest(session, source_id=source_id, telegram=_telegram(second=0))
        assert not ingest.ingest(
            session,
            source_id=source_id,
            telegram=_telegram(
                second=10,
                fingerprint=invalid_fingerprint,
                channel_fingerprints={1: valid, 2: valid},
            ),
        )
        session.commit()
    assert _counts(factory) == (0, 0)
    with factory() as session:
        source = session.get(MeterSource, source_id)
        assert source is not None and source.last_error == "WarmteLink frame fingerprint is invalid"
        assert raw_identifier not in (source.last_error or "")
        assert not ingest.ingest(session, source_id=source_id, telegram=_telegram(second=20, water="5.901"))
        assert ingest.ingest(session, source_id=source_id, telegram=_telegram(second=30, water="5.902"))
        session.commit()
    assert _counts(factory) == (2, 2)