From 72ac7e7300f2d0d0cd7c4d503ac22b41ed10d85d Mon Sep 17 00:00:00 2001 From: Tianyu Liu Date: Mon, 24 Aug 2026 04:23:24 +0200 Subject: [PATCH] M8-R04: interpret WarmteLink timestamps as Amsterdam wall time --- app/services/warmtelink_ingest.py | 85 +++++--- tests/test_warmtelink_ingest.py | 328 ++++++++++++++++++++++++++++-- 2 files changed, 373 insertions(+), 40 deletions(-) diff --git a/app/services/warmtelink_ingest.py b/app/services/warmtelink_ingest.py index bbdd655..b38d995 100644 --- a/app/services/warmtelink_ingest.py +++ b/app/services/warmtelink_ingest.py @@ -13,6 +13,7 @@ from dataclasses import dataclass from datetime import UTC, datetime, timedelta from decimal import Decimal import re +from zoneinfo import ZoneInfo from sqlalchemy import select from sqlalchemy.orm import Session @@ -52,7 +53,14 @@ class WarmteLinkIngestor: self._clock = clock or (lambda: datetime.now(UTC)) self._previous: dict[int, _FrameSnapshot] = {} - def ingest(self, session: Session, *, source_id: int, telegram: P1Telegram) -> bool: + def ingest( + self, + session: Session, + *, + source_id: int, + telegram: P1Telegram, + received_at: datetime | None = None, + ) -> bool: """Apply one parsed telegram in the caller's transaction. Returns whether the frame was admitted. Callers that own a session @@ -65,44 +73,47 @@ class WarmteLinkIngestor: if source.kind != "warmtelink_serial": raise ValueError("Source is not a WarmteLink serial source") + received_at = _utc(self._clock()) if received_at is None else _utc(received_at) try: integrity = _integrity_status(telegram.integrity) except Exception: self._previous.pop(source_id, None) - self._diagnose(source, "WarmteLink frame could not be normalized") + self._diagnose(source, "WarmteLink frame could not be normalized", now=received_at) return False if integrity is IntegrityStatus.INVALID: self._previous.pop(source_id, None) - self._diagnose(source, "WarmteLink frame checksum is invalid") + self._diagnose(source, "WarmteLink frame checksum is invalid", now=received_at) return False try: - snapshot = _snapshot(telegram) + snapshot = _snapshot(telegram, received_at) fingerprints = _final_fingerprints(snapshot) except Exception: # A parser DTO remains an untrusted boundary. Do not let an # unnormalizable DTO bridge two otherwise matching candidates. self._previous.pop(source_id, None) - self._diagnose(source, "WarmteLink frame could not be normalized") + self._diagnose(source, "WarmteLink frame could not be normalized", now=received_at) return False if fingerprints is None: self._previous.pop(source_id, None) - self._diagnose(source, "WarmteLink frame fingerprint is invalid") + self._diagnose(source, "WarmteLink frame fingerprint is invalid", now=received_at) return False if integrity is IntegrityStatus.UNVERIFIABLE: previous = self._previous.get(source_id) if previous is None: self._previous[source_id] = snapshot - self._diagnose(source, "Awaiting a second matching unverifiable WarmteLink frame") + self._diagnose( + source, "Awaiting a second matching unverifiable WarmteLink frame", now=received_at + ) return False reason = _continuity_problem(previous, snapshot) if reason is not None: self._previous[source_id] = snapshot - self._diagnose(source, reason) + self._diagnose(source, reason, now=received_at) return False try: - self._admit(session, source, snapshot, integrity.value, fingerprints) + self._admit(session, source, snapshot, integrity.value, fingerprints, received_at) except Exception: # ``ingest`` owns flushes and may be used directly by tests or # future callers. A failed write must never become a speculative @@ -133,20 +144,25 @@ class WarmteLinkIngestor: a generic source error. Consequently no partial latest/history update survives and the following frame may recover normally. """ + received_at = _utc(self._clock()) try: telegram = parser(frame) except Exception: self._previous.pop(source_id, None) - self._record_error(session_factory, source_id, "WarmteLink frame could not be parsed") + self._record_error( + session_factory, source_id, "WarmteLink frame could not be parsed", now=received_at + ) return False try: with session_factory() as session: - admitted = self.ingest(session, source_id=source_id, telegram=telegram) + admitted = self.ingest( + session, source_id=source_id, telegram=telegram, received_at=received_at + ) session.commit() return admitted except Exception: self._previous.pop(source_id, None) - self._record_error(session_factory, source_id, "WarmteLink ingest failed") + self._record_error(session_factory, source_id, "WarmteLink ingest failed", now=received_at) return False def _admit( @@ -156,8 +172,8 @@ class WarmteLinkIngestor: snapshot: _FrameSnapshot, quality: str, fingerprints: tuple[str, ...], + received_at: datetime, ) -> None: - received_at = _utc(self._clock()) for sample, fingerprint in zip(snapshot.samples, fingerprints, strict=True): channel = upsert_discovered_channel( session, @@ -200,27 +216,33 @@ class WarmteLinkIngestor: source.last_error = None source.updated_at = received_at - def _diagnose(self, source: MeterSource, reason: str) -> None: - now = _utc(self._clock()) + def _diagnose(self, source: MeterSource, reason: str, *, now: datetime | None = None) -> None: + now = _utc(self._clock()) if now is None else now source.status = "error" source.last_error = reason source.updated_at = now def _record_error( - self, session_factory: Callable[[], Session], source_id: int, message: str) -> None: + self, + session_factory: Callable[[], Session], + source_id: int, + message: str, + *, + now: datetime, + ) -> None: try: with session_factory() as session: source = session.get(MeterSource, source_id) if source is not None: - self._diagnose(source, message) + self._diagnose(source, message, now=now) session.commit() except Exception: # Error reporting itself must not kill another source worker. return -def _snapshot(telegram: P1Telegram) -> _FrameSnapshot: - recorded_at = _parse_timestamp(telegram.timestamp) +def _snapshot(telegram: P1Telegram, received_at: datetime) -> _FrameSnapshot: + recorded_at = _parse_timestamp(telegram.timestamp, received_at) samples = tuple(_sample(channel) for channel in telegram.channels) if not samples: raise ValueError("WarmteLink telegram contains no cumulative channels") @@ -340,13 +362,30 @@ def _is_canonical_fingerprint(value: str | None) -> bool: return value is not None and _FINGERPRINT_PATTERN.fullmatch(value) is not None -def _parse_timestamp(value: str | None) -> datetime: +_AMSTERDAM = ZoneInfo("Europe/Amsterdam") +_MAX_CLOCK_SKEW = timedelta(minutes=5) + + +def _parse_timestamp(value: str | None, received_at: datetime) -> datetime: if value is None or len(value) != 13 or value[-1] not in {"S", "W"} or not value[:-1].isdigit(): raise ValueError("WarmteLink timestamp is unavailable") naive = datetime.strptime(value[:-1], "%y%m%d%H%M%S") - # The telegram's S/W marker is evidence of the intended Amsterdam offset. - offset = 2 if value[-1] == "S" else 1 - return (naive.replace(tzinfo=UTC) - timedelta(hours=offset)).replace(tzinfo=UTC) + # The S/W marker is only advisory: deployed devices have emitted W while + # on CEST. Validate both folds by a UTC round trip, which rejects spring + # gaps and leaves one (ordinary) or two (fall-back) real instants. + candidates: list[datetime] = [] + for fold in (0, 1): + candidate = naive.replace(tzinfo=_AMSTERDAM, fold=fold).astimezone(UTC) + local = candidate.astimezone(_AMSTERDAM) + if local.replace(tzinfo=None) == naive and local.fold == fold and candidate not in candidates: + candidates.append(candidate) + if not candidates: + raise ValueError("WarmteLink timestamp is unavailable") + received_at = _utc(received_at) + recorded_at = min(candidates, key=lambda candidate: abs(candidate - received_at)) + if abs(recorded_at - received_at) > _MAX_CLOCK_SKEW: + raise ValueError("WarmteLink timestamp is unavailable") + return recorded_at def _suggestion(unit: str) -> str | None: diff --git a/tests/test_warmtelink_ingest.py b/tests/test_warmtelink_ingest.py index 48092a2..a5e8135 100644 --- a/tests/test_warmtelink_ingest.py +++ b/tests/test_warmtelink_ingest.py @@ -3,7 +3,7 @@ from __future__ import annotations from dataclasses import replace -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta from decimal import Decimal from pathlib import Path from types import SimpleNamespace @@ -62,6 +62,7 @@ def _telegram( fingerprint: str | None = "a" * 64, channel_fingerprint: str | None = None, channel_fingerprints: dict[int, str | None] | None = None, + timestamp: str | None = None, ) -> P1Telegram: fields: list[ObisField] = [] parsed_channels: list[P1Channel] = [] @@ -86,7 +87,7 @@ def _telegram( 100, integrity, "fixture", - f"26082212{minute:02d}{second:02d}S", + timestamp or f"26082212{minute:02d}{second:02d}S", fingerprint, tuple(fields), tuple(parsed_channels), @@ -113,6 +114,260 @@ def _fixed_clock() -> datetime: return datetime(2026, 8, 22, 10, tzinfo=UTC) +def _clock_at(value: datetime): + return lambda: value + + +class _AdvancingClock: + def __init__(self, initial: datetime) -> None: + self._next = initial + self.calls = 0 + + def __call__(self) -> datetime: + value = self._next + self._next += timedelta(minutes=1) + self.calls += 1 + return value + + +def _as_utc(value: datetime) -> datetime: + return value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC) + + +def test_summer_wall_clock_ignores_incorrect_w_marker(database): + _engine, factory, (source_id, _) = database + received_at = datetime(2026, 8, 23, 21, 40, tzinfo=UTC) + ingest = WarmteLinkIngestor(clock=_clock_at(received_at)) + with factory() as session: + assert ingest.ingest( + session, + source_id=source_id, + telegram=_telegram(second=0, integrity=IntegrityStatus.VALID, timestamp="260823234000W"), + ) + session.commit() + with factory() as session: + reading = session.scalar(select(WarmteLinkReading).order_by(WarmteLinkReading.id)) + assert reading is not None + assert _as_utc(reading.recorded_at) == datetime(2026, 8, 23, 21, 40, tzinfo=UTC) + + +def test_winter_wall_clock_uses_amsterdam_standard_time_even_with_s_marker(database): + _engine, factory, (source_id, _) = database + received_at = datetime(2026, 1, 23, 22, 40, tzinfo=UTC) + ingest = WarmteLinkIngestor(clock=_clock_at(received_at)) + with factory() as session: + assert ingest.ingest( + session, + source_id=source_id, + telegram=_telegram(second=0, integrity=IntegrityStatus.VALID, timestamp="260123234000S"), + ) + session.commit() + with factory() as session: + reading = session.scalar(select(WarmteLinkReading).order_by(WarmteLinkReading.id)) + assert reading is not None + assert _as_utc(reading.recorded_at) == datetime(2026, 1, 23, 22, 40, tzinfo=UTC) + + +@pytest.mark.parametrize( + ("received_at", "expected"), + [ + (datetime(2026, 10, 25, 0, 30, tzinfo=UTC), datetime(2026, 10, 25, 0, 30, tzinfo=UTC)), + (datetime(2026, 10, 25, 1, 30, tzinfo=UTC), datetime(2026, 10, 25, 1, 30, tzinfo=UTC)), + ], +) +def test_fall_back_fold_is_selected_from_received_at(database, received_at, expected): + _engine, factory, (source_id, _) = database + ingest = WarmteLinkIngestor(clock=_clock_at(received_at)) + with factory() as session: + assert ingest.ingest( + session, + source_id=source_id, + telegram=_telegram(second=0, integrity=IntegrityStatus.VALID, timestamp="261025023000W"), + ) + session.commit() + with factory() as session: + reading = session.scalar(select(WarmteLinkReading).order_by(WarmteLinkReading.id)) + assert reading is not None and _as_utc(reading.recorded_at) == expected + + +@pytest.mark.parametrize("timestamp", ["260329023000S", "260823234000X"]) +def test_nonexistent_or_invalid_marker_timestamp_fails_closed_with_generic_diagnostic(database, timestamp): + _engine, factory, (source_id, _) = database + ingest = WarmteLinkIngestor(clock=_clock_at(datetime(2026, 3, 29, 1, 30, tzinfo=UTC))) + with factory() as session: + assert not ingest.ingest( + session, + source_id=source_id, + telegram=_telegram(second=0, integrity=IntegrityStatus.VALID, timestamp=timestamp), + ) + session.commit() + source = session.get(MeterSource, source_id) + assert source is not None and source.last_error == "WarmteLink frame could not be normalized" + assert _counts(factory) == (0, 0) + + +def test_timestamp_beyond_clock_skew_fails_closed_with_generic_diagnostic(database): + _engine, factory, (source_id, _) = database + ingest = WarmteLinkIngestor(clock=_clock_at(datetime(2026, 8, 23, 20, tzinfo=UTC))) + with factory() as session: + assert not ingest.ingest( + session, + source_id=source_id, + telegram=_telegram(second=0, integrity=IntegrityStatus.VALID, timestamp="260823234000W"), + ) + session.commit() + source = session.get(MeterSource, source_id) + assert source is not None and source.last_error == "WarmteLink frame could not be normalized" + assert _counts(factory) == (0, 0) + + +@pytest.mark.parametrize("timestamp", ["260823234000X", "260823234600W"]) +def test_post_clock_rejections_reuse_received_at_for_diagnostic(database, timestamp): + _engine, factory, (source_id, _) = database + received_at = datetime(2026, 8, 23, 21, 40, tzinfo=UTC) + clock = _AdvancingClock(received_at) + ingest = WarmteLinkIngestor(clock=clock) + with factory() as session: + assert not ingest.ingest( + session, + source_id=source_id, + telegram=_telegram(second=0, integrity=IntegrityStatus.VALID, timestamp=timestamp), + ) + session.commit() + source = session.get(MeterSource, source_id) + assert source is not None + assert _as_utc(source.updated_at) == received_at + assert clock.calls == 1 + + +def test_first_unverifiable_candidate_reuses_received_at_for_diagnostic(database): + _engine, factory, (source_id, _) = database + received_at = datetime(2026, 8, 22, 10, tzinfo=UTC) + clock = _AdvancingClock(received_at) + ingest = WarmteLinkIngestor(clock=clock) + with factory() as session: + assert not ingest.ingest(session, source_id=source_id, telegram=_telegram(second=0)) + session.commit() + source = session.get(MeterSource, source_id) + assert source is not None + assert _as_utc(source.updated_at) == received_at + assert clock.calls == 1 + + +def test_accepted_frame_uses_received_at_for_all_source_timestamps(database): + _engine, factory, (source_id, _) = database + received_at = datetime(2026, 8, 22, 10, tzinfo=UTC) + clock = _AdvancingClock(received_at) + ingest = WarmteLinkIngestor(clock=clock) + with factory() as session: + assert ingest.ingest( + session, source_id=source_id, telegram=_telegram(second=0, integrity=IntegrityStatus.VALID) + ) + session.commit() + with factory() as session: + source = session.get(MeterSource, source_id) + reading = session.scalar(select(WarmteLinkReading).order_by(WarmteLinkReading.id)) + assert source is not None and reading is not None + assert _as_utc(reading.received_at) == received_at + assert _as_utc(source.last_seen_at) == received_at + assert _as_utc(source.updated_at) == received_at + assert clock.calls == 1 + + +@pytest.mark.parametrize( + "path", + ["parser", "integrity", "normalization", "fingerprint", "candidate", "continuity", "accepted"], +) +def test_handle_frame_samples_the_receive_clock_once_for_each_admission_path(database, path): + _engine, factory, (source_id, _) = database + received_at = datetime(2026, 8, 22, 10, tzinfo=UTC) + clock = _AdvancingClock(received_at) + ingest = WarmteLinkIngestor(clock=clock) + + if path == "parser": + assert not ingest.handle_frame( + source_id, + b"private", + session_factory=factory, + parser=lambda _raw: (_ for _ in ()).throw(ValueError()), + ) + expected_at = received_at + elif path == "integrity": + assert not ingest.handle_frame( + source_id, + b"private", + session_factory=factory, + parser=lambda _raw: _telegram(second=0, integrity=IntegrityStatus.INVALID), + ) + expected_at = received_at + elif path == "normalization": + assert not ingest.handle_frame( + source_id, + b"private", + session_factory=factory, + parser=lambda _raw: _telegram(second=0, timestamp="invalid-time"), + ) + expected_at = received_at + elif path == "fingerprint": + assert not ingest.handle_frame( + source_id, + b"private", + session_factory=factory, + parser=lambda _raw: _telegram(second=0, integrity=IntegrityStatus.VALID, fingerprint="invalid"), + ) + expected_at = received_at + elif path == "candidate": + assert not ingest.handle_frame( + source_id, b"private", session_factory=factory, parser=lambda _raw: _telegram(second=0) + ) + expected_at = received_at + elif path == "continuity": + assert not ingest.handle_frame( + source_id, b"private", session_factory=factory, parser=lambda _raw: _telegram(second=0) + ) + assert not ingest.handle_frame( + source_id, + b"private", + session_factory=factory, + parser=lambda _raw: _telegram(second=20, water="5.901"), + ) + expected_at = received_at + timedelta(minutes=1) + else: + assert ingest.handle_frame( + source_id, + b"private", + session_factory=factory, + parser=lambda _raw: _telegram(second=0, integrity=IntegrityStatus.VALID), + ) + expected_at = received_at + + with factory() as session: + source = session.get(MeterSource, source_id) + assert source is not None + assert _as_utc(source.updated_at) == expected_at + assert clock.calls == (2 if path == "continuity" else 1) + + +def test_unverifiable_summer_w_frames_keep_utc_continuity(database): + _engine, factory, (source_id, _) = database + received_at = datetime(2026, 8, 23, 21, 40, 10, tzinfo=UTC) + ingest = WarmteLinkIngestor(clock=_clock_at(received_at)) + with factory() as session: + assert not ingest.ingest(session, source_id=source_id, telegram=_telegram(second=0, timestamp="260823234000W")) + assert ingest.ingest( + session, + source_id=source_id, + telegram=_telegram(second=10, water="5.901", timestamp="260823234010W"), + ) + session.commit() + with factory() as session: + readings = list(session.scalars(select(WarmteLinkReading).order_by(WarmteLinkReading.id))) + assert [_as_utc(reading.recorded_at) for reading in readings] == [ + datetime(2026, 8, 23, 21, 40, 10, tzinfo=UTC), + datetime(2026, 8, 23, 21, 40, 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) @@ -143,14 +398,14 @@ def test_unverifiable_requires_two_continuous_frames_and_preserves_true_metadata def test_unverifiable_cadence_requires_exactly_ten_seconds_and_restarts_after_gaps(database): _engine, factory, (source_id, _) = database - ingest = WarmteLinkIngestor() + ingest = WarmteLinkIngestor(clock=_fixed_clock) 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() + restarted = WarmteLinkIngestor(clock=_fixed_clock) 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. @@ -176,7 +431,7 @@ def test_unverifiable_cadence_requires_exactly_ten_seconds_and_restarts_after_ga ) def test_unverifiable_discontinuities_restart_two_frame_confirmation(database, first, second): _engine, factory, (source_id, _) = database - ingest = WarmteLinkIngestor() + ingest = WarmteLinkIngestor(clock=_fixed_clock) 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) @@ -189,7 +444,7 @@ def test_unverifiable_discontinuities_restart_two_frame_confirmation(database, f def test_discontinuity_candidate_needs_a_new_matching_successor(database): _engine, factory, (source_id, _) = database - ingest = WarmteLinkIngestor() + ingest = WarmteLinkIngestor(clock=_fixed_clock) 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,))) @@ -213,7 +468,7 @@ def test_each_metadata_or_channel_discontinuity_requires_its_own_two_frames( database, first, discontinuous, successor ): _engine, factory, (source_id, _) = database - ingest = WarmteLinkIngestor() + ingest = WarmteLinkIngestor(clock=_fixed_clock) 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. @@ -224,7 +479,7 @@ def test_each_metadata_or_channel_discontinuity_requires_its_own_two_frames( def test_parser_failure_clears_only_its_source_candidate_and_requires_two_new_frames(database): _engine, factory, (one, two) = database - ingest = WarmteLinkIngestor() + ingest = WarmteLinkIngestor(clock=_fixed_clock) 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)) @@ -296,7 +551,7 @@ def test_dto_normalization_failure_clears_only_its_source_candidate_and_requires database, kind ): _engine, factory, (one, two) = database - ingest = WarmteLinkIngestor() + ingest = WarmteLinkIngestor(clock=_fixed_clock) 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)) @@ -366,7 +621,9 @@ def test_history_keeps_first_utc_minute_sample_and_creates_next_minute(database) def test_handle_frame_rolls_back_database_error_marks_source_and_recovers(database, monkeypatch): _engine, factory, (source_id, _) = database - ingest = WarmteLinkIngestor() + received_at = datetime(2026, 8, 22, 10, tzinfo=UTC) + clock = _AdvancingClock(received_at) + ingest = WarmteLinkIngestor(clock=clock) telegram = _telegram(second=0, integrity=IntegrityStatus.VALID) import app.services.warmtelink_ingest as module @@ -382,14 +639,51 @@ def test_handle_frame_rolls_back_database_error_marks_source_and_recovers(databa source = session.get(MeterSource, source_id) assert source is not None and source.status == "error" assert "private" not in (source.last_error or "") + assert _as_utc(source.updated_at) == received_at + assert clock.calls == 1 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_handle_frame_commit_failure_reuses_its_receive_clock_for_error_diagnostic(database, monkeypatch): + _engine, factory, (source_id, _) = database + received_at = datetime(2026, 8, 22, 10, tzinfo=UTC) + clock = _AdvancingClock(received_at) + ingest = WarmteLinkIngestor(clock=clock) + original_commit = Session.commit + sessions_created = 0 + + def session_factory(): + nonlocal sessions_created + session = factory() + session.info["fail_commit"] = sessions_created == 0 + sessions_created += 1 + return session + + def fail_initial_commit(self): + if self.info.get("fail_commit"): + raise RuntimeError("database commit failed") + return original_commit(self) + + monkeypatch.setattr(Session, "commit", fail_initial_commit) + assert not ingest.handle_frame( + source_id, + b"private", + session_factory=session_factory, + parser=lambda _raw: _telegram(second=0, integrity=IntegrityStatus.VALID), + ) + with factory() as session: + source = session.get(MeterSource, source_id) + assert source is not None + assert source.status == "error" + assert _as_utc(source.updated_at) == received_at + assert clock.calls == 1 + + def test_second_channel_failure_rolls_back_first_channel_latest_and_history(database, monkeypatch): _engine, factory, (source_id, _) = database - ingest = WarmteLinkIngestor() + ingest = WarmteLinkIngestor(clock=_fixed_clock) import app.services.warmtelink_ingest as module original = module.upsert_discovered_channel @@ -419,7 +713,7 @@ def test_second_channel_failure_rolls_back_first_channel_latest_and_history(data def test_unverifiable_database_failure_clears_only_its_source_sliding_candidate(database, monkeypatch): _engine, factory, (one, two) = database - ingest = WarmteLinkIngestor() + ingest = WarmteLinkIngestor(clock=_fixed_clock) import app.services.warmtelink_ingest as module with factory() as session: @@ -462,7 +756,7 @@ def _valid_crc_frame(*, device_type: str = "006") -> bytes: def test_non_enum_integrity_and_fake_metadata_reject_whole_frame_per_source(database): _engine, factory, (one, two) = database - ingest = WarmteLinkIngestor() + ingest = WarmteLinkIngestor(clock=_fixed_clock) 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)) @@ -487,7 +781,7 @@ def test_non_enum_integrity_and_fake_metadata_reject_whole_frame_per_source(data def test_valid_crc_parser_frame_with_raw_device_type_is_rejected_without_leakage(database): _engine, factory, (one, two) = database - ingest = WarmteLinkIngestor() + ingest = WarmteLinkIngestor(clock=_fixed_clock) 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)) @@ -523,7 +817,7 @@ def test_invalid_fingerprint_fails_closed_without_persisting_identifiers( ): _engine, factory, (source_id, _) = database raw_identifier = "serial-number-which-must-not-persist" - ingest = WarmteLinkIngestor() + ingest = WarmteLinkIngestor(clock=_fixed_clock) with factory() as session: assert not ingest.ingest(session, source_id=source_id, telegram=_telegram(second=0, integrity=IntegrityStatus.INVALID)) assert not ingest.ingest( @@ -561,7 +855,7 @@ def test_invalid_channel_fingerprint_rejects_whole_frame_and_clears_candidate( _engine, factory, (source_id, _) = database raw_identifier = "serial-number-which-must-not-persist" valid = "a" * 64 - ingest = WarmteLinkIngestor() + ingest = WarmteLinkIngestor(clock=_fixed_clock) 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)) @@ -596,7 +890,7 @@ def test_invalid_top_fingerprint_rejects_whole_frame_and_clears_candidate(databa _engine, factory, (source_id, _) = database raw_identifier = "serial-number-which-must-not-persist" valid = "a" * 64 - ingest = WarmteLinkIngestor() + ingest = WarmteLinkIngestor(clock=_fixed_clock) with factory() as session: assert not ingest.ingest(session, source_id=source_id, telegram=_telegram(second=0)) assert not ingest.ingest(