diff --git a/applications/modelling_e2e/handler.py b/applications/modelling_e2e/handler.py index 3442cb362..e0c31ca37 100644 --- a/applications/modelling_e2e/handler.py +++ b/applications/modelling_e2e/handler.py @@ -387,6 +387,15 @@ def _dedupe_skipped( return unique +def _newer_lodged( + stored: Optional[EpcPropertyData], fetched: Optional[EpcPropertyData] +) -> tuple[Optional[EpcPropertyData], bool]: + """ADR-0001 Recency Tie-Break over the two lodged sources. Returns + (chosen, chosen_is_fetched) — the bool gates the EPC save, so a stored EPC + that wins is never re-persisted.""" + raise NotImplementedError + + def _predict_epc( *, property_id: int, diff --git a/tests/applications/modelling_e2e/test_handler.py b/tests/applications/modelling_e2e/test_handler.py index b51f784a7..e2396e0c9 100644 --- a/tests/applications/modelling_e2e/test_handler.py +++ b/tests/applications/modelling_e2e/test_handler.py @@ -7,7 +7,11 @@ is needed. One test per distinct behaviour path. from __future__ import annotations +import dataclasses +import json from contextlib import ExitStack +from datetime import date +from pathlib import Path from typing import Any, Iterator, Optional from unittest.mock import MagicMock, call, patch from uuid import UUID, uuid4 @@ -18,6 +22,8 @@ from pydantic import ValidationError from applications.modelling_e2e.modelling_e2e_trigger_body import ( ModellingE2ETriggerBody, ) +from datatypes.epc.domain.epc_property_data import EpcPropertyData +from datatypes.epc.domain.mapper import EpcPropertyDataMapper from domain.tasks.subtasks import SubTask from repositories.epc.epc_postgres_repository import EpcSaveRequest @@ -1403,9 +1409,38 @@ def test_cohort_cache_prevents_duplicate_candidates_for_calls() -> None: # --------------------------------------------------------------------------- -# refetch_epc flag +# Recency Tie-Break between the two lodged sources (ADR-0001) # --------------------------------------------------------------------------- +_JSON_SAMPLES = Path(__file__).resolve().parents[3] / "backend/epc_api/json_samples" + + +def _lodged_epc(inspection: str) -> EpcPropertyData: + """A real lodged EPC surveyed on `inspection` — the tie-break reads only + inspection_date, but a real cert keeps the test honest about the type.""" + raw: dict[str, Any] = json.loads( + (_JSON_SAMPLES / "RdSAP-Schema-21.0.0" / "epc.json").read_text() + ) + epc = EpcPropertyDataMapper.from_api_response(raw) + return dataclasses.replace(epc, inspection_date=date.fromisoformat(inspection)) + + +def test_stored_survey_newer_than_gov_cert_wins_the_tie_break() -> None: + """The bug in #1589: a newer stored survey (e.g. PasHub site notes) must beat + an older gov-register cert, and must not be flagged for re-persisting.""" + # Arrange + from applications.modelling_e2e.handler import _newer_lodged + + stored = _lodged_epc("2026-05-01") + fetched = _lodged_epc("2023-12-01") + + # Act + chosen, chosen_is_fetched = _newer_lodged(stored, fetched) + + # Assert + assert chosen is stored + assert chosen_is_fetched is False + def test_refetch_epc_false_with_stored_epc_skips_api_call() -> None: """refetch_epc=False + stored lodged EPC present: EpcClientService.get_by_uprn