Merge pull request #1547 from Hestia-Homes/feature/fetch-from-epc-property-on-uprn

Fetch a property's lodged EPC by UPRN, not property_id

applications/modelling_e2e/handler.py

@@ -533,8 +533,12 @@ def handler(
             SolarPostgresRepository(read_session).get_many(list(set(uprns.values())))
         )
         epc_repo = EpcPostgresRepository(read_session)
+        # Lodged EPCs are read by UPRN (keyed by UPRN below); predicted EPCs stay
+        # keyed on property_id.
         stored_lodged_epcs: dict[int, EpcPropertyData] = (
-            epc_repo.get_for_properties(property_ids) if not refetch_epc else {}
+            epc_repo.get_for_properties(list(set(uprns.values())))
+            if not refetch_epc
+            else {}
         )
         stored_predicted_epcs: dict[int, EpcPropertyData] = (
             epc_repo.get_predicted_for_properties(property_ids)
@@ -565,7 +569,7 @@ def handler(
                 spatial.coordinates if spatial is not None else None
             )
 
-            stored_lodged = stored_lodged_epcs.get(pid)
+            stored_lodged = stored_lodged_epcs.get(uprn)
             lodged_epc_is_new = False
             if refetch_epc:
                 epc: Optional[EpcPropertyData] = epc_client.get_by_uprn(uprn)

harness/console.py

@@ -16,7 +16,7 @@ REPL at the worktree root::
 
 from __future__ import annotations
 
-from dataclasses import dataclass
+from dataclasses import dataclass, replace
 from pathlib import Path
 from typing import Any, Optional
 
@@ -106,6 +106,9 @@ def run_one(
     Valuation Uplift in £ — otherwise the uplift is percentage-only (ADR-0018).
     ``epc`` must carry lodged recorded-performance + the RHI block (a real lodged
     EPC does) so the Baseline stage can run."""
+    # A lodged EPC fetched for a UPRN carries that UPRN; stamp it so the lodged read
+    # (keyed on UPRN) links it to the synthetic property.
+    epc = replace(epc, uprn=_UPRN)
     epc_repo = FakeEpcRepo()
     plan_repo = FakePlanRepository()
     property_repo = FakePropertyRepo(

repositories/epc/epc_postgres_repository.py

@@ -416,53 +416,81 @@ class EpcPostgresRepository(EpcRepository):
             delete(EpcPropertyModel).where(col(EpcPropertyModel.id).in_(epc_ids))
         )
 
-    def get_for_property(self, property_id: int) -> Optional[EpcPropertyData]:
-        return self._get_for_property(property_id, source="lodged")
+    def get_for_property(self, uprn: Optional[int]) -> Optional[EpcPropertyData]:
+        # UPRN is not unique on epc_property, so the tie-break is load-bearing: the
+        # latest-surveyed lodged row wins (inspection_date is non-null), with id as
+        # the deterministic secondary.
+        if uprn is None:
+            return None
+        row = self._session.exec(
+            select(EpcPropertyModel)
+            .where(EpcPropertyModel.uprn == uprn)
+            .where(col(EpcPropertyModel.source).in_(_slot_sources("lodged")))
+            .order_by(
+                col(EpcPropertyModel.inspection_date).desc(),
+                col(EpcPropertyModel.id).desc(),
+            )
+        ).first()
+        if row is None or row.id is None:
+            return None
+        return self.get(row.id)
 
     def get_predicted_for_property(self, property_id: int) -> Optional[EpcPropertyData]:
-        return self._get_for_property(property_id, source="predicted")
-
-    def _get_for_property(
-        self, property_id: int, source: EpcSource
-    ) -> Optional[EpcPropertyData]:
         row = self._session.exec(
             select(EpcPropertyModel)
             .where(EpcPropertyModel.property_id == property_id)
-            .where(col(EpcPropertyModel.source).in_(_slot_sources(source)))
+            .where(col(EpcPropertyModel.source).in_(_slot_sources("predicted")))
             .order_by(EpcPropertyModel.id)  # type: ignore[arg-type]
         ).first()
         if row is None or row.id is None:
             return None
         return self.get(row.id)
 
-    def get_for_properties(self, property_ids: list[int]) -> dict[int, EpcPropertyData]:
-        """Bulk-hydrate a batch's LODGED EPCs, keyed by property_id."""
-        return self._for_properties(property_ids, source="lodged")
+    def get_for_properties(self, uprns: list[int]) -> dict[int, EpcPropertyData]:
+        """Bulk-hydrate a batch's LODGED EPCs, keyed by UPRN."""
+        if not uprns:
+            return {}
+        parents = self._session.exec(
+            select(EpcPropertyModel)
+            .where(col(EpcPropertyModel.uprn).in_(uprns))
+            .where(col(EpcPropertyModel.source).in_(_slot_sources("lodged")))
+            .order_by(
+                col(EpcPropertyModel.inspection_date).desc(),
+                col(EpcPropertyModel.id).desc(),
+            )
+        ).all()
+        # UPRN is not unique, so setdefault keeps the latest-surveyed row per UPRN.
+        by_uprn: dict[int, EpcPropertyModel] = {}
+        for parent in parents:
+            if parent.uprn is not None and parent.id is not None:
+                by_uprn.setdefault(parent.uprn, parent)
+        return self._hydrate(by_uprn)
 
     def get_predicted_for_properties(
         self, property_ids: list[int]
     ) -> dict[int, EpcPropertyData]:
         """Bulk-hydrate a batch's PREDICTED EPCs (ADR-0031), keyed by property_id."""
-        return self._for_properties(property_ids, source="predicted")
-
-    def _for_properties(
-        self, property_ids: list[int], source: EpcSource
-    ) -> dict[int, EpcPropertyData]:
-        """Bulk-hydrate a batch's EPCs of one `source` in a handful of per-table IN
-        queries (ADR-0012), not N x per-property. Load-whole per ADR-0002."""
         if not property_ids:
             return {}
         parents = self._session.exec(
             select(EpcPropertyModel)
             .where(col(EpcPropertyModel.property_id).in_(property_ids))
-            .where(col(EpcPropertyModel.source).in_(_slot_sources(source)))
+            .where(col(EpcPropertyModel.source).in_(_slot_sources("predicted")))
             .order_by(EpcPropertyModel.id)  # type: ignore[arg-type]
         ).all()
-        parent_by_property: dict[int, EpcPropertyModel] = {}
+        by_property: dict[int, EpcPropertyModel] = {}
         for parent in parents:
             if parent.property_id is not None and parent.id is not None:
-                parent_by_property.setdefault(parent.property_id, parent)
-        epc_ids = [p.id for p in parent_by_property.values() if p.id is not None]
+                by_property.setdefault(parent.property_id, parent)
+        return self._hydrate(by_property)
+
+    def _hydrate(
+        self, parents_by_key: dict[int, EpcPropertyModel]
+    ) -> dict[int, EpcPropertyData]:
+        """Load each parent's child tables in a handful of per-table IN queries
+        (ADR-0012), not N x per-property, and compose them into EpcPropertyData
+        keyed the same way as `parents_by_key`. Load-whole per ADR-0002."""
+        epc_ids = [p.id for p in parents_by_key.values() if p.id is not None]
         if not epc_ids:
             return {}
 
@@ -531,9 +559,9 @@ class EpcPostgresRepository(EpcRepository):
         floor_dims_by_part = self._floor_dims_by_part(part_ids)
 
         result: dict[int, EpcPropertyData] = {}
-        for property_id, parent in parent_by_property.items():
+        for key, parent in parents_by_key.items():
             epc_id = _require(parent.id, "id")
-            result[property_id] = self._compose(
+            result[key] = self._compose(
                 p=parent,
                 perf=perf_by.get(epc_id),
                 elements=elements_by.get(epc_id, []),

repositories/epc/epc_repository.py

@@ -46,23 +46,25 @@ class EpcRepository(ABC):
     def get(self, epc_property_id: int) -> EpcPropertyData: ...
 
     @abstractmethod
-    def get_for_property(self, property_id: int) -> Optional[EpcPropertyData]:
-        """The property's LODGED EPC (the predicted slot is read separately)."""
+    def get_for_property(self, uprn: Optional[int]) -> Optional[EpcPropertyData]:
+        """The property's LODGED EPC by UPRN (an epc_property row may carry a null
+        property_id, so UPRN is its durable key), or None. The predicted slot is
+        read separately, and stays keyed on property_id."""
         ...
 
     @abstractmethod
     def get_predicted_for_property(
         self, property_id: int
     ) -> Optional[EpcPropertyData]:
+        # Keyed on property_id, not UPRN: a predicted EPC deep-copies a neighbour's
+        # UPRN, so its UPRN column is never the property's own.
         """The property's PREDICTED EPC (EPC Prediction gap-fill), or None."""
         ...
 
     @abstractmethod
-    def get_for_properties(
-        self, property_ids: list[int]
-    ) -> dict[int, EpcPropertyData]:
-        """Bulk-hydrate a batch's LODGED EPCs, keyed by property_id (only those
-        with one are present). A handful of per-table queries, not N per property."""
+    def get_for_properties(self, uprns: list[int]) -> dict[int, EpcPropertyData]:
+        """Bulk-hydrate a batch's LODGED EPCs, keyed by UPRN (only those with one
+        are present). A handful of per-table queries, not N per property."""
         ...
 
     @abstractmethod

repositories/property/property_postgres_repository.py

@@ -89,7 +89,7 @@ class PropertyPostgresRepository(PropertyRepository):
         )
         return Property(
             identity=identity,
-            epc=self._epc().get_for_property(property_id),
+            epc=self._epc().get_for_property(row.uprn),
             predicted_epc=self._epc().get_predicted_for_property(property_id),
             landlord_overrides=self._landlord_overrides(property_id),
             planning_restrictions=_restrictions_of(row.uprn, restrictions),
@@ -102,11 +102,10 @@ class PropertyPostgresRepository(PropertyRepository):
             select(PropertyRow).where(col(PropertyRow.id).in_(property_ids))
         ).all()
         row_by_id = {row.id: row for row in rows}
-        epcs = self._epc().get_for_properties(property_ids)
+        uprns = [row.uprn for row in rows if row.uprn is not None]
+        epcs = self._epc().get_for_properties(uprns)
         predicted_epcs = self._epc().get_predicted_for_properties(property_ids)
-        restrictions: dict[int, PlanningRestrictions] = self._restrictions_for(
-            [row.uprn for row in rows if row.uprn is not None]
-        )
+        restrictions: dict[int, PlanningRestrictions] = self._restrictions_for(uprns)
         items: list[Property] = []
         for property_id in property_ids:
             row = row_by_id.get(property_id)
@@ -121,7 +120,7 @@ class PropertyPostgresRepository(PropertyRepository):
                         uprn=row.uprn,
                         landlord_property_id=row.landlord_property_id,
                     ),
-                    epc=epcs.get(property_id),
+                    epc=epcs.get(row.uprn) if row.uprn is not None else None,
                     predicted_epc=predicted_epcs.get(property_id),
                     landlord_overrides=self._landlord_overrides(property_id),
                     planning_restrictions=_restrictions_of(row.uprn, restrictions),

tests/applications/modelling_e2e/test_handler.py

@@ -1450,9 +1450,10 @@ def test_refetch_epc_false_with_stored_epc_skips_api_call() -> None:
             patch("applications.modelling_e2e.handler.run_modelling", return_value=mock_plan)
         )
         # Bulk-read of stored lodged EPCs returns the stored EPC for this property
+        # (lodged reads key on UPRN)
         stack.enter_context(
             patch("applications.modelling_e2e.handler.EpcPostgresRepository")
-        ).return_value.get_for_properties.return_value = {PROPERTY_ID: stored_epc}
+        ).return_value.get_for_properties.return_value = {UPRN: stored_epc}
         MockUoW = stack.enter_context(
             patch("applications.modelling_e2e.handler.PostgresUnitOfWork")
         )

tests/orchestration/fakes.py

@@ -57,7 +57,7 @@ class FakePropertyRepo(PropertyRepository):
             return prop
         return Property(
             identity=prop.identity,
-            epc=self._epc_repo.get_for_property(property_id),
+            epc=self._epc_repo.get_for_property(prop.identity.uprn),
             site_notes=prop.site_notes,
             current_market_value=prop.current_market_value,
         )
@@ -83,11 +83,12 @@ class FakePropertyRepo(PropertyRepository):
 
 
 class FakeEpcRepo(EpcRepository):
-    def __init__(self, by_property: Optional[dict[int, EpcPropertyData]] = None) -> None:
+    def __init__(self, by_uprn: Optional[dict[int, EpcPropertyData]] = None) -> None:
         self.saved: list[tuple[EpcPropertyData, Optional[int]]] = []
         self.sources: list[EpcSource] = []
-        self._by_property = by_property or {}
-        # Predicted EPCs live in their own slot, coexisting with lodged (ADR-0031).
+        # Lodged EPCs read by UPRN (mirroring EpcPostgresRepository, whose uprn
+        # column is data.uprn); predicted EPCs read by property_id (ADR-0031).
+        self._lodged_by_uprn = by_uprn or {}
         self._predicted_by_property: dict[int, EpcPropertyData] = {}
 
     def save(
@@ -99,33 +100,29 @@ class FakeEpcRepo(EpcRepository):
     ) -> int:
         self.saved.append((data, property_id))
         self.sources.append(source)
-        if property_id is not None:
-            slot = (
-                self._predicted_by_property
-                if source in PREDICTED_SLOT_SOURCES
-                else self._by_property
-            )
-            slot[property_id] = data
+        if source in PREDICTED_SLOT_SOURCES:
+            if property_id is not None:
+                self._predicted_by_property[property_id] = data
+        elif data.uprn is not None:
+            self._lodged_by_uprn[data.uprn] = data
         return len(self.saved)
 
     def get(self, epc_property_id: int) -> EpcPropertyData:  # pragma: no cover
         raise NotImplementedError
 
-    def get_for_property(self, property_id: int) -> Optional[EpcPropertyData]:
-        return self._by_property.get(property_id)
+    def get_for_property(self, uprn: Optional[int]) -> Optional[EpcPropertyData]:
+        return self._lodged_by_uprn.get(uprn) if uprn is not None else None
 
     def get_predicted_for_property(
         self, property_id: int
     ) -> Optional[EpcPropertyData]:
         return self._predicted_by_property.get(property_id)
 
-    def get_for_properties(
-        self, property_ids: list[int]
-    ) -> dict[int, EpcPropertyData]:
+    def get_for_properties(self, uprns: list[int]) -> dict[int, EpcPropertyData]:
         return {
-            property_id: self._by_property[property_id]
-            for property_id in property_ids
-            if property_id in self._by_property
+            uprn: self._lodged_by_uprn[uprn]
+            for uprn in uprns
+            if uprn in self._lodged_by_uprn
         }
 
     def get_predicted_for_properties(

tests/orchestration/test_ara_first_run_pipeline_integration.py

@@ -92,6 +92,7 @@ def _lodged_epc() -> EpcPropertyData:
     epc = EpcPropertyDataMapper.from_api_response(raw)
     return dataclasses.replace(
         epc,
+        uprn=12345,
         energy_rating_current=72,
         current_energy_efficiency_band=Epc.C,
         co2_emissions_current=1.8,
@@ -326,7 +327,7 @@ def test_modelling_optimises_and_persists_a_multi_measure_plan(
         )
         session.commit()
         EpcPostgresRepository(session).save(
-            _build_uninsulated_cavity_and_floor_epc(),
+            dataclasses.replace(_build_uninsulated_cavity_and_floor_epc(), uprn=33333),
             property_id=30,
             portfolio_id=1,
         )
@@ -504,7 +505,7 @@ def test_modelling_recommends_nothing_when_already_at_the_target_band(
         )
         session.commit()
         EpcPostgresRepository(session).save(
-            _build_uninsulated_cavity_and_floor_epc(),
+            dataclasses.replace(_build_uninsulated_cavity_and_floor_epc(), uprn=44444),
             property_id=31,
             portfolio_id=1,
         )
@@ -663,8 +664,16 @@ def test_listed_uprn_ingested_blocks_solid_wall_insulation_in_modelling(
         )
         session.commit()
         epc_repo = EpcPostgresRepository(session)
-        epc_repo.save(solid_brick_epc, property_id=40, portfolio_id=1)
-        epc_repo.save(solid_brick_epc, property_id=41, portfolio_id=1)
+        epc_repo.save(
+            dataclasses.replace(solid_brick_epc, uprn=44444),
+            property_id=40,
+            portfolio_id=1,
+        )
+        epc_repo.save(
+            dataclasses.replace(solid_brick_epc, uprn=55555),
+            property_id=41,
+            portfolio_id=1,
+        )
         session.commit()
 
     def unit_of_work() -> PostgresUnitOfWork:

tests/orchestration/test_ingestion_prediction.py

@@ -5,6 +5,7 @@ prediction collaborators are optional; when unwired, ingestion is unchanged."""
 from __future__ import annotations
 
 import json
+from dataclasses import replace
 from datetime import date
 from pathlib import Path
 from typing import Any, Optional
@@ -139,7 +140,7 @@ def test_epc_less_property_is_predicted_and_persisted_to_the_predicted_slot() ->
     predicted = epc_repo.get_predicted_for_property(10)
     assert predicted is not None
     assert predicted.property_type == "0"
-    assert epc_repo.get_for_property(10) is None
+    assert epc_repo.get_for_property(12345) is None
     assert uow.commits == 1
 
 
@@ -170,12 +171,13 @@ def test_unknown_property_type_gates_the_property_out_of_prediction() -> None:
     # Assert — gated out: no cohort fetched, nothing predicted or persisted.
     assert comparables_repo.searched == []
     assert epc_repo.get_predicted_for_property(10) is None
-    assert epc_repo.get_for_property(10) is None
+    assert epc_repo.get_for_property(12345) is None
 
 
 def test_a_lodged_epc_is_not_predicted_over() -> None:
-    # Arrange — a real EPC is fetched, so prediction must not run.
-    lodged = _epc()
+    # Arrange — a real EPC is fetched, so prediction must not run. It carries the
+    # property's UPRN (lodged reads key on UPRN).
+    lodged = replace(_epc(), uprn=12345)
     epc_repo = FakeEpcRepo()
     uow = FakeUnitOfWork(
         property=FakePropertyRepo({10: _property(uprn=12345)}),
@@ -200,7 +202,7 @@ def test_a_lodged_epc_is_not_predicted_over() -> None:
 
     # Assert — the lodged EPC is saved; no cohort fetch, no predicted slot.
     assert comparables_repo.searched == []
-    assert epc_repo.get_for_property(10) == lodged
+    assert epc_repo.get_for_property(12345) == lodged
     assert epc_repo.get_predicted_for_property(10) is None
 
 
@@ -258,7 +260,7 @@ def test_expired_historic_epc_conditions_prediction_and_labels_the_source() -> N
     assert comparables_repo.searched == ["A0 0AA"]
     assert epc_repo.get_predicted_for_property(10) is not None
     assert epc_repo.sources == ["expired"]
-    assert epc_repo.get_for_property(10) is None
+    assert epc_repo.get_for_property(12345) is None
 
 
 def test_landlord_overrides_win_over_the_expired_historic_cert() -> None:

tests/repositories/epc/test_epc_batch_save.py

@@ -80,9 +80,10 @@ def test_multi_property_building_part_ids_are_not_crossed(db_engine: Engine) ->
     # Arrange — property A has 2 parts with distinctive areas; B has 1 with a
     # third distinctive area. If part IDs are mis-wired the floor-dimension FK
     # rows end up under the wrong property.
+    # Distinct UPRNs so the UPRN-keyed lodged read tells the two properties apart.
     base = _load_epc()
-    epc_a = _with_floor_areas(base, [10.0, 20.0])
-    epc_b = _with_floor_areas(base, [99.0])
+    epc_a = replace(_with_floor_areas(base, [10.0, 20.0]), uprn=2001)
+    epc_b = replace(_with_floor_areas(base, [99.0]), uprn=2002)
 
     with Session(db_engine) as session:
         repo = EpcPostgresRepository(session)
@@ -121,7 +122,7 @@ def test_multi_property_building_part_ids_are_not_crossed(db_engine: Engine) ->
 
 def test_save_batch_is_idempotent(db_engine: Engine) -> None:
     # Arrange
-    epc = _load_epc()
+    epc = replace(_load_epc(), uprn=3001)
     requests = [EpcSaveRequest(epc, property_id=3001)]
 
     with Session(db_engine) as session:
@@ -146,14 +147,15 @@ def test_save_batch_is_idempotent(db_engine: Engine) -> None:
 
 def test_lodged_and_predicted_batch_slots_are_independent(db_engine: Engine) -> None:
     # Arrange — two properties each get a lodged EPC and then a predicted EPC
-    # via separate save_batch() calls.
-    epc = _load_epc()
+    # via separate save_batch() calls. UPRNs equal the property_ids so the lodged
+    # (UPRN-keyed) and predicted (property_id-keyed) bulk reads line up.
     property_ids = [4001, 4002]
+    epc_by_uprn = {pid: replace(_load_epc(), uprn=pid) for pid in property_ids}
 
     with Session(db_engine) as session:
         repo = EpcPostgresRepository(session)
-        repo.save_batch([EpcSaveRequest(epc, property_id=pid, source="lodged") for pid in property_ids])
-        repo.save_batch([EpcSaveRequest(epc, property_id=pid, source="predicted") for pid in property_ids])
+        repo.save_batch([EpcSaveRequest(epc_by_uprn[pid], property_id=pid, source="lodged") for pid in property_ids])
+        repo.save_batch([EpcSaveRequest(epc_by_uprn[pid], property_id=pid, source="predicted") for pid in property_ids])
         session.commit()
 
     # Act
@@ -163,5 +165,5 @@ def test_lodged_and_predicted_batch_slots_are_independent(db_engine: Engine) ->
         predicted = repo.get_predicted_for_properties(property_ids)
 
     # Assert — both slots are populated for both properties.
-    assert lodged == {4001: epc, 4002: epc}
-    assert predicted == {4001: epc, 4002: epc}
+    assert lodged == epc_by_uprn
+    assert predicted == epc_by_uprn

tests/repositories/epc/test_epc_bulk_read.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 
 import json
 from collections.abc import Callable
+from dataclasses import replace
 from pathlib import Path
 from typing import Any
 
@@ -41,36 +42,36 @@ def _count_queries(engine: Engine, work: Callable[[], None]) -> int:
 
 
 def test_get_for_properties_hydrates_the_whole_batch(db_engine: Engine) -> None:
-    # Arrange — the same sample EPC persisted for two properties.
-    epc = _load_epc()
+    # Distinct UPRNs so the UPRN-keyed read distinguishes the two properties.
+    epc_by_uprn = {pid: replace(_load_epc(), uprn=pid) for pid in [10, 11]}
     with Session(db_engine) as session:
         repo = EpcPostgresRepository(session)
-        repo.save(epc, property_id=10)
-        repo.save(epc, property_id=11)
+        repo.save(epc_by_uprn[10], property_id=10)
+        repo.save(epc_by_uprn[11], property_id=11)
         session.commit()
 
     # Act
     with Session(db_engine) as session:
         result = EpcPostgresRepository(session).get_for_properties([10, 11])
 
-    # Assert — both fully hydrated (load-whole, ADR-0002).
-    assert result == {10: epc, 11: epc}
+    # Assert — both fully hydrated (load-whole, ADR-0002), keyed by UPRN.
+    assert result == epc_by_uprn
 
 
 def test_get_for_properties_round_trips_do_not_scale_with_batch_size(
     db_engine: Engine,
 ) -> None:
-    # Arrange
-    epc = _load_epc()
+    # Arrange — distinct UPRNs (equal to the property_ids) for the UPRN-keyed read.
+    epc_by_uprn = {pid: replace(_load_epc(), uprn=pid) for pid in [10, 11]}
     with Session(db_engine) as session:
         repo = EpcPostgresRepository(session)
-        repo.save(epc, property_id=10)
-        repo.save(epc, property_id=11)
+        repo.save(epc_by_uprn[10], property_id=10)
+        repo.save(epc_by_uprn[11], property_id=11)
         session.commit()
 
-    def _read(property_ids: list[int]) -> None:
+    def _read(uprns: list[int]) -> None:
         with Session(db_engine) as session:
-            EpcPostgresRepository(session).get_for_properties(property_ids)
+            EpcPostgresRepository(session).get_for_properties(uprns)
 
     # Act — count queries for a 1-property batch vs a 2-property batch.
     one = _count_queries(db_engine, lambda: _read([10]))

tests/repositories/epc/test_epc_idempotent_save.py

@@ -30,7 +30,7 @@ def test_resaving_an_epc_for_a_property_replaces_rather_than_duplicates(
     db_engine: Engine,
 ) -> None:
     # Arrange — same property re-ingested with a changed field.
-    original = _load_epc()
+    original = dataclasses.replace(_load_epc(), uprn=10)
     updated = dataclasses.replace(original, status="re-run-sentinel")
 
     # Act — save twice for the same property_id (a re-run).

tests/repositories/epc/test_epc_lodged_uprn_collision.py

@@ -0,0 +1,69 @@
+"""Lodged EPC reads key on UPRN, which is not unique on epc_property. When several
+lodged rows share a UPRN, the latest-surveyed (most recent inspection_date) row
+wins — regardless of insertion order."""
+
+from __future__ import annotations
+
+import json
+from dataclasses import replace
+from datetime import date
+from pathlib import Path
+from typing import Any
+
+from sqlalchemy import Engine
+from sqlmodel import Session
+
+from datatypes.epc.domain.epc_property_data import EpcPropertyData
+from datatypes.epc.domain.mapper import EpcPropertyDataMapper
+from repositories.epc.epc_postgres_repository import EpcPostgresRepository
+
+_JSON_SAMPLES = Path(__file__).resolve().parents[3] / "backend/epc_api/json_samples"
+
+
+def _load_epc() -> EpcPropertyData:
+    raw: dict[str, Any] = json.loads(
+        (_JSON_SAMPLES / "RdSAP-Schema-21.0.0" / "epc.json").read_text()
+    )
+    return EpcPropertyDataMapper.from_api_response(raw)
+
+
+def _seed_shared_uprn(db_engine: Engine) -> None:
+    # Two lodged EPCs share UPRN 500 under different property_ids. The write path
+    # dedupes per property_id, not UPRN, so both rows persist — a real collision.
+    # The latest-surveyed EPC is saved FIRST (lower id) so a passing test proves
+    # inspection_date, not insertion order, decides the winner.
+    epc = _load_epc()
+    with Session(db_engine) as session:
+        repo = EpcPostgresRepository(session)
+        repo.save(
+            replace(epc, uprn=500, inspection_date=date(2022, 6, 1), status="newer"),
+            property_id=100,
+        )
+        repo.save(
+            replace(epc, uprn=500, inspection_date=date(2018, 6, 1), status="older"),
+            property_id=101,
+        )
+        session.commit()
+
+
+def test_get_for_property_returns_the_most_recent_row_for_a_shared_uprn(
+    db_engine: Engine,
+) -> None:
+    _seed_shared_uprn(db_engine)
+
+    with Session(db_engine) as session:
+        lodged = EpcPostgresRepository(session).get_for_property(500)
+
+    assert lodged is not None
+    assert lodged.status == "newer"
+
+
+def test_get_for_properties_returns_the_most_recent_row_for_a_shared_uprn(
+    db_engine: Engine,
+) -> None:
+    _seed_shared_uprn(db_engine)
+
+    with Session(db_engine) as session:
+        result = EpcPostgresRepository(session).get_for_properties([500])
+
+    assert result[500].status == "newer"

tests/repositories/epc/test_epc_predicted_slot.py

@@ -4,6 +4,7 @@
 from __future__ import annotations
 
 import json
+from dataclasses import replace
 from pathlib import Path
 from typing import Any
 
@@ -26,7 +27,7 @@ def _epc() -> EpcPropertyData:
 
 def test_lodged_and_predicted_epcs_coexist_per_property(db_engine: Engine) -> None:
     # Arrange — the same property gets a lodged EPC and a predicted EPC.
-    epc = _epc()
+    epc = replace(_epc(), uprn=42)
     with Session(db_engine) as session:
         repo = EpcPostgresRepository(session)
         repo.save(epc, property_id=42)
@@ -48,7 +49,7 @@ def test_re_saving_a_predicted_epc_leaves_the_lodged_one_intact(
     db_engine: Engine,
 ) -> None:
     # Arrange — a lodged EPC, then a predicted one saved twice (idempotent re-run).
-    epc = _epc()
+    epc = replace(_epc(), uprn=7)
     with Session(db_engine) as session:
         repo = EpcPostgresRepository(session)
         repo.save(epc, property_id=7)
@@ -69,7 +70,7 @@ def test_re_saving_a_predicted_epc_leaves_the_lodged_one_intact(
 
 def test_a_lodged_only_property_has_no_predicted_epc(db_engine: Engine) -> None:
     # Arrange — only a lodged EPC is saved.
-    epc = _epc()
+    epc = replace(_epc(), uprn=9)
     with Session(db_engine) as session:
         repo = EpcPostgresRepository(session)
         repo.save(epc, property_id=9)
@@ -86,7 +87,7 @@ def test_an_expired_source_epc_lives_in_the_predicted_slot(db_engine: Engine) ->
     # Arrange — an Expired-Enhanced Prediction is persisted with source="expired"
     # (ADR-0054): enhanced by an expired historic observation, not totally
     # predicted — but it occupies the same predicted slot.
-    epc = _epc()
+    epc = replace(_epc(), uprn=11)
     with Session(db_engine) as session:
         repo = EpcPostgresRepository(session)
         repo.save(epc, property_id=11, source="expired")

tests/repositories/property/test_property_postgres_overlay_hydration.py

@@ -7,6 +7,7 @@ the override folded onto the lodged wall — proving reader → overlay → aggr
 
 from __future__ import annotations
 
+import dataclasses
 import json
 from pathlib import Path
 from typing import Any
@@ -51,7 +52,9 @@ def test_reloaded_property_effective_epc_reflects_the_wall_override(
         property_id = row.id
         assert property_id is not None
 
-        EpcPostgresRepository(session).save(_epc(), property_id=property_id)
+        EpcPostgresRepository(session).save(
+            dataclasses.replace(_epc(), uprn=1), property_id=property_id
+        )
         session.add(
             PropertyOverrideRow(
                 property_id=property_id,
@@ -92,7 +95,9 @@ def test_property_without_overrides_keeps_its_lodged_wall(db_engine: Engine) ->
         session.commit()
         property_id = row.id
         assert property_id is not None
-        EpcPostgresRepository(session).save(_epc(), property_id=property_id)
+        EpcPostgresRepository(session).save(
+            dataclasses.replace(_epc(), uprn=2), property_id=property_id
+        )
         session.commit()
 
     # Act

tests/repositories/property/test_property_repository.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import dataclasses
 import json
 from pathlib import Path
 from typing import Any
@@ -27,7 +28,9 @@ def test_get_hydrates_identity_and_epc_slice(db_engine: Engine) -> None:
     raw: dict[str, Any] = json.loads(
         (_JSON_SAMPLES / "RdSAP-Schema-21.0.0" / "epc.json").read_text()
     )
-    epc = EpcPropertyDataMapper.from_api_response(raw)
+    epc = dataclasses.replace(
+        EpcPropertyDataMapper.from_api_response(raw), uprn=12345
+    )
     with Session(db_engine) as session:
         row = PropertyRow(
             portfolio_id=7, postcode="A0 0AA", address="1 Some Street", uprn=12345

tests/repositories/test_unit_of_work.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import dataclasses
 import json
 from collections.abc import Callable
 from pathlib import Path
@@ -123,7 +124,7 @@ def test_unit_hydrates_a_property_with_its_landlord_overrides_folded(
         property_id = row.id
         assert property_id is not None
         EpcPostgresRepository(uow._session).save(  # pyright: ignore[reportPrivateUsage]
-            _epc(), property_id=property_id
+            dataclasses.replace(_epc(), uprn=1), property_id=property_id
         )
         uow._session.add(  # pyright: ignore[reportPrivateUsage]
             PropertyOverrideRow(
@@ -171,7 +172,7 @@ def test_hydrating_a_property_with_overrides_stays_on_one_connection(
         property_id = row.id
         assert property_id is not None
         EpcPostgresRepository(uow._session).save(  # pyright: ignore[reportPrivateUsage]
-            _epc(), property_id=property_id
+            dataclasses.replace(_epc(), uprn=1), property_id=property_id
         )
         uow._session.add(  # pyright: ignore[reportPrivateUsage]
             PropertyOverrideRow(