junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-07-27 23:35:01 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

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

Browse source 
An epc_property row can carry a null property_id (FE/historic ingestion
persists EPC rows never linked to a property row), so a property_id-keyed
read silently misses them. UPRN is the durable key both the property row
and the epc_property row share. Repoint the lodged reads (get_for_property
+ get_for_properties) onto UPRN; the predicted reads stay on property_id
because a predicted EPC deep-copies a neighbour's UPRN, so its UPRN column
is never the property's own.

UPRN is not unique on epc_property, so the read tie-break is now load-
bearing (it was dormant under property_id, where the write path guarantees
one lodged row per property): the most recently ingested (highest-id) row
wins. Write path stays keyed on (property_id, source).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Daniel Roth 2026-07-14 10:35:51 +00:00
parent e1c99edc3f
commit f7328f54c5
17 changed files with 212 additions and 101 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
applications/modelling_e2e/handler.py
View file

@ -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)

5
harness/console.py
View file

@ -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(

70
repositories/epc/epc_postgres_repository.py
View file

@ -416,53 +416,75 @@ 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
# most recently ingested (highest-id) lodged row wins.
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.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.id).desc())
).all()
# UPRN is not unique, so setdefault keeps the most recently ingested
# (highest-id) 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 +553,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, []),

16
repositories/epc/epc_repository.py
View file

@ -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

11
repositories/property/property_postgres_repository.py
View file

@ -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),

3
tests/applications/modelling_e2e/test_handler.py
View file

@ -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")
)

35
tests/orchestration/fakes.py
View file

@ -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(

17
tests/orchestration/test_ara_first_run_pipeline_integration.py
View file

@ -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:

14
tests/orchestration/test_ingestion_prediction.py
View file

@ -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:

20
tests/repositories/epc/test_epc_batch_save.py
View file

@ -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

25
tests/repositories/epc/test_epc_bulk_read.py
View file

@ -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]))

2
tests/repositories/epc/test_epc_idempotent_save.py
View file

@ -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).

59
tests/repositories/epc/test_epc_lodged_uprn_collision.py Normal file
View file

@ -0,0 +1,59 @@
"""Lodged EPC reads key on UPRN, which is not unique on epc_property. When several
lodged rows share a UPRN, the most recently ingested (highest-id) row wins."""
from __future__ import annotations
import json
from dataclasses import replace
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.
epc = _load_epc()
with Session(db_engine) as session:
repo = EpcPostgresRepository(session)
repo.save(replace(epc, uprn=500, status="older"), property_id=100)
repo.save(replace(epc, uprn=500, status="newer"), 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"

9
tests/repositories/epc/test_epc_predicted_slot.py
View file

@ -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")

9
tests/repositories/property/test_property_postgres_overlay_hydration.py
View file

@ -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

5
tests/repositories/property/test_property_repository.py
View file

@ -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

5
tests/repositories/test_unit_of_work.py
View file

@ -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(