junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-08-03 05:18:22 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

Merge pull request #1687 from Hestia-Homes/fix/epc-property-per-property-persistence

Browse source 
EPC persistence: write a per-property lodged copy, and refresh a stale stored cert on refetch
This commit is contained in:
KhalimCK 2026-07-27 16:44:55 +01:00 • committed by GitHub
commit 035d2f5136
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 523 additions and 12 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

71
applications/modelling_e2e/handler.py
View file

@ -207,11 +207,6 @@ def _flush_writes(engine: Engine, writes: list[_PropertyWrite]) -> None:
transaction back so the SQS message is retried — every save is an idempotent
upsert. Per-property failures are isolated earlier, in the modelling loop,
before a write is ever queued."""
lodged_requests = [
EpcSaveRequest(w.lodged_epc, property_id=w.property_id, portfolio_id=w.portfolio_id, source="lodged")
for w in writes
if w.lodged_epc is not None and w.lodged_epc_is_new
]
predicted_requests = [
EpcSaveRequest(
w.predicted_epc,
@ -223,6 +218,25 @@ def _flush_writes(engine: Engine, writes: list[_PropertyWrite]) -> None:
if w.predicted_epc is not None and w.predicted_epc_is_new
]
with PostgresUnitOfWork(lambda: Session(engine)) as uow:
# A freshly-fetched lodged cert (lodged_epc_is_new) is always persisted.
# But the lodged read keys on UPRN (ADR-0001 recency), so a property whose
# UPRN matches a *sibling's* row — the same dwelling onboarded under another
# portfolio — or a null-property_id survey row sees is_new=False while
# owning no lodged row of its own. The FE reads epc_property by property_id,
# so without a property_id-anchored copy the modelled EPC is invisible. Give
# any such property its own copy: persist when the cert is new OR when this
# property has no lodged row of its own (observed on portfolio 850 /
# property 792235, whose uprn 100021979925 also lives under property 753896
# in portfolio 830 — the sibling's row hid the EPC from property 792235).
pids_with_lodged: set[int] = uow.epc.property_ids_with_lodged_epc(
[w.property_id for w in writes if w.lodged_epc is not None]
)
lodged_requests = [
EpcSaveRequest(w.lodged_epc, property_id=w.property_id, portfolio_id=w.portfolio_id, source="lodged")
for w in writes
if w.lodged_epc is not None
and (w.lodged_epc_is_new or w.property_id not in pids_with_lodged)
]
if lodged_requests:
uow.epc.save_batch(lodged_requests)
if predicted_requests:
@ -402,6 +416,36 @@ def _newer_lodged(
return stored, False
def _reconcile_lodged(
stored: Optional[EpcPropertyData],
fetched: Optional[EpcPropertyData],
*,
stored_is_survey: bool,
) -> tuple[Optional[EpcPropertyData], bool]:
"""The ADR-0001 Recency Tie-Break, plus a same-date content refresh so a
``refetch_epc`` re-ingests a stale stored gov cert.
Recency still decides first: a strictly newer source wins. The addition is the
equal-inspection-date case — where recency alone keeps the stored row. When the
stored row is a public gov cert (``stored_is_survey`` False) and the freshly
fetched cert differs from it, the fetched cert wins and is flagged for saving,
so a DB row written before a schema field/table existed (e.g. room-in-roof
geometry) is healed by a plain re-fetch. One of OUR surveys (PasHub / ECMK /
Elmhurst — ``uploaded_file_id`` set) is preserved on the tie, keeping the
survey-wins rule (#1589). An unchanged public cert is still not re-persisted."""
chosen, chosen_is_fetched = _newer_lodged(stored, fetched)
if (
not chosen_is_fetched
and stored is not None
and fetched is not None
and not stored_is_survey
and fetched.inspection_date == stored.inspection_date
and fetched != stored
):
return fetched, True
return chosen, chosen_is_fetched
def _predict_epc(
*,
property_id: int,
@ -613,6 +657,13 @@ def handler(
stored_lodged_epcs: dict[int, EpcPropertyData] = epc_repo.get_for_properties(
list(set(uprns.values()))
)
# Which stored lodged rows are OUR surveys (uploaded_file_id set) rather
# than public gov certs — provenance the hydrated EPC drops. Gates the
# same-date refresh so a gov refetch heals a stale public cert but never
# overwrites a survey (see _reconcile_lodged).
survey_lodged_uprns: set[int] = epc_repo.survey_lodged_uprns(
list(set(uprns.values()))
)
stored_predicted_epcs: dict[int, EpcPropertyData] = (
epc_repo.get_predicted_for_properties(property_ids)
if not repredict_epc
@ -648,13 +699,15 @@ def handler(
epc: Optional[EpcPropertyData] = None
lodged_epc_is_new = False
if refetch_epc:
epc, lodged_epc_is_new = _newer_lodged(
stored_lodged, epc_client.get_by_uprn(uprn)
epc, lodged_epc_is_new = _reconcile_lodged(
stored_lodged,
epc_client.get_by_uprn(uprn),
stored_is_survey=uprn in survey_lodged_uprns,
)
if epc is not None and not lodged_epc_is_new:
logger.info(
f"property={pid} stored lodged EPC is newer than the "
f"fetched cert — keeping the stored assessment"
f"property={pid} stored lodged EPC wins the reconcile "
f"(newer, or an unchanged/survey tie) — keeping it"
)
elif stored_lodged is not None:
logger.info(

45
repositories/epc/epc_postgres_repository.py
View file

@ -577,6 +577,51 @@ class EpcPostgresRepository(EpcRepository):
by_property.setdefault(parent.property_id, parent)
return self._hydrate(by_property)
def survey_lodged_uprns(self, uprns: list[int]) -> set[int]:
"""The subset of `uprns` whose winning LODGED row is one of our uploaded
surveys (`uploaded_file_id` set) rather than a public gov-API cert.
Mirrors ``get_for_properties`` ordering (inspection_date desc, id desc) and
takes the first row per UPRN, so the provenance matches the row it
hydrates."""
if not uprns:
return set()
rows = self._session.exec(
select(EpcPropertyModel.uprn, EpcPropertyModel.uploaded_file_id)
.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()
winning_uploaded_file_id: dict[int, Optional[int]] = {}
for uprn, uploaded_file_id in rows:
if uprn is not None:
winning_uploaded_file_id.setdefault(uprn, uploaded_file_id)
return {
uprn
for uprn, uploaded_file_id in winning_uploaded_file_id.items()
if uploaded_file_id is not None
}
def property_ids_with_lodged_epc(self, property_ids: list[int]) -> set[int]:
"""The subset of `property_ids` that already own a LODGED row keyed on
their own property_id.
Unlike ``get_for_properties`` (UPRN-keyed, recency tie-break), this keys on
property_id — it answers "does the FE have an EPC to read for this
property?", which a UPRN-shared sibling's row does not. Ids only; the graph
is never hydrated."""
if not property_ids:
return set()
rows = self._session.exec(
select(EpcPropertyModel.property_id)
.where(col(EpcPropertyModel.property_id).in_(property_ids))
.where(col(EpcPropertyModel.source).in_(_slot_sources("lodged")))
).all()
return {pid for pid in rows if pid is not None}
def _hydrate(
self, parents_by_key: dict[int, EpcPropertyModel]
) -> dict[int, EpcPropertyData]:

24
repositories/epc/epc_repository.py
View file

@ -74,3 +74,27 @@ class EpcRepository(ABC):
"""Bulk-hydrate a batch's PREDICTED EPCs (ADR-0031), keyed by property_id
(only those with one are present)."""
...
@abstractmethod
def survey_lodged_uprns(self, uprns: list[int]) -> set[int]:
"""The subset of `uprns` whose winning LODGED row (the one
`get_for_properties` selects) is one of OUR uploaded surveys — a cert
pulled from PasHub / ECMK / Elmhurst, marked by a non-null
`uploaded_file_id` — rather than a public gov-API cert.
Provenance the hydrated `EpcPropertyData` does not carry. Used to protect a
survey from being overwritten when a gov refetch reconciles a stale stored
cert of the same inspection date (ADR-0001 survey-wins)."""
...
@abstractmethod
def property_ids_with_lodged_epc(self, property_ids: list[int]) -> set[int]:
"""The subset of `property_ids` that already own a LODGED epc_property row
keyed on their OWN property_id.
The lodged read keys on UPRN (a row may carry a null or a sibling's
property_id), so "the batch has a lodged EPC for this UPRN" does not imply
"this property has a lodged row the FE can read". The persist step consults
this to give a property its own property_id-anchored copy when a UPRN-shared
sibling's row would otherwise suppress the write."""
...

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

@ -1528,6 +1528,9 @@ def test_refetch_epc_true_models_the_newer_stored_survey_without_resaving_it() -
gov_epc = _lodged_epc("2023-12-01") # older public cert
mock_plan = _plan_mock()
mock_uow = MagicMock()
# This property already owns its lodged row (the survey was persisted before),
# so the per-property persist must NOT re-fire on top of the recency skip.
mock_uow.epc.property_ids_with_lodged_epc.return_value = {PROPERTY_ID}
with ExitStack() as stack:
stack.enter_context(patch("applications.modelling_e2e.handler.os.environ", _ENV))
@ -1589,6 +1592,9 @@ def test_refetch_epc_false_with_stored_epc_skips_api_call() -> None:
stored_epc = MagicMock()
mock_plan = _plan_mock()
mock_uow = MagicMock()
# This property already owns its lodged row, so the per-property persist must
# NOT re-fire when the stored EPC is reused unchanged.
mock_uow.epc.property_ids_with_lodged_epc.return_value = {PROPERTY_ID}
with ExitStack() as stack:
stack.enter_context(patch("applications.modelling_e2e.handler.os.environ", _ENV))
@ -1731,6 +1737,302 @@ def test_refetch_epc_false_without_stored_epc_skips_api_and_goes_to_prediction()
)
def test_reconcile_refreshes_a_stale_gov_cert_on_an_equal_date() -> None:
"""A stored PUBLIC gov cert (not one of our surveys) on the same inspection
date but with different content is refreshed to the fetched cert — this is how
a refetch heals a row written before a schema field/table existed. The recency
tie-break alone (strict >) would drop the same-date refetch."""
# Arrange
from applications.modelling_e2e.handler import _reconcile_lodged
stored = _lodged_epc("2025-07-18")
fetched = dataclasses.replace(_lodged_epc("2025-07-18"), post_town="REFRESHED")
# Act
chosen, chosen_is_fetched = _reconcile_lodged(
stored, fetched, stored_is_survey=False
)
# Assert
assert chosen is fetched
assert chosen_is_fetched is True
def test_reconcile_keeps_a_gov_cert_that_is_unchanged() -> None:
"""Same date, same content, public cert: nothing to refresh, so it is not
re-persisted (the RDS-CPU-saving skip still holds)."""
# Arrange
from applications.modelling_e2e.handler import _reconcile_lodged
stored = _lodged_epc("2025-07-18")
fetched = _lodged_epc("2025-07-18")
# Act
chosen, chosen_is_fetched = _reconcile_lodged(
stored, fetched, stored_is_survey=False
)
# Assert
assert chosen is stored
assert chosen_is_fetched is False
def test_reconcile_preserves_our_survey_over_a_same_date_gov_cert() -> None:
"""A stored SURVEY (uploaded_file_id set) on the same inspection date is not
overwritten by a differing gov cert — ADR-0001/#1589 survey-wins on a tie."""
# Arrange
from applications.modelling_e2e.handler import _reconcile_lodged
stored = _lodged_epc("2025-07-18")
fetched = dataclasses.replace(_lodged_epc("2025-07-18"), post_town="GOV")
# Act
chosen, chosen_is_fetched = _reconcile_lodged(
stored, fetched, stored_is_survey=True
)
# Assert
assert chosen is stored
assert chosen_is_fetched is False
def test_reconcile_newer_fetched_wins_regardless_of_survey_flag() -> None:
"""A genuinely newer gov cert supersedes even a stored survey (recency)."""
# Arrange
from applications.modelling_e2e.handler import _reconcile_lodged
stored = _lodged_epc("2023-12-01")
fetched = _lodged_epc("2026-05-01")
# Act
chosen, chosen_is_fetched = _reconcile_lodged(
stored, fetched, stored_is_survey=True
)
# Assert
assert chosen is fetched
assert chosen_is_fetched is True
def test_refetch_refreshes_a_stale_stored_gov_cert_when_content_differs() -> None:
"""End-to-end: refetch_epc=True with a stored public gov cert of the same
inspection date but stale content re-persists the freshly-fetched cert, so the
DB row is healed (e.g. room-in-roof geometry added by a later schema change).
The property already owns its row, so the refresh — not the missing-row path —
is what drives the write."""
# Arrange
mock_engine = _engine_mock([PROPERTY_ID], [UPRN], [POSTCODE])
stored_epc = _lodged_epc("2025-07-18")
fetched_epc = dataclasses.replace(_lodged_epc("2025-07-18"), post_town="REFRESHED")
mock_plan = _plan_mock()
mock_uow = MagicMock()
mock_uow.epc.property_ids_with_lodged_epc.return_value = {PROPERTY_ID}
with ExitStack() as stack:
stack.enter_context(patch("applications.modelling_e2e.handler.os.environ", _ENV))
stack.enter_context(
patch("applications.modelling_e2e.handler._get_engine", return_value=mock_engine)
)
stack.enter_context(
patch("applications.modelling_e2e.handler.EpcClientService")
).return_value.get_by_uprn.return_value = fetched_epc
stack.enter_context(patch("applications.modelling_e2e.handler.GeospatialS3Repository"))
stack.enter_context(patch("applications.modelling_e2e.handler.GoogleSolarApiClient"))
stack.enter_context(
patch("applications.modelling_e2e.handler._spatial_for", return_value=None)
)
stack.enter_context(
patch("applications.modelling_e2e.handler._solar_insights_for", return_value=None)
)
stack.enter_context(
patch("applications.modelling_e2e.handler.overlays_from", return_value=[])
)
stack.enter_context(
patch("applications.modelling_e2e.handler.PropertyOverridesPostgresReader")
).return_value.overrides_for_many.return_value = {}
stack.enter_context(
patch("applications.modelling_e2e.handler.ScenarioPostgresRepository")
).return_value.get_many.return_value = [MagicMock()]
stack.enter_context(
patch("applications.modelling_e2e.handler.catalogue_snapshot_with_off_catalogue_overrides")
)
stack.enter_context(patch("applications.modelling_e2e.handler.Session"))
mock_run_modelling = stack.enter_context(
patch("applications.modelling_e2e.handler.run_modelling", return_value=mock_plan)
)
mock_repo = stack.enter_context(
patch("applications.modelling_e2e.handler.EpcPostgresRepository")
).return_value
mock_repo.get_for_properties.return_value = {UPRN: stored_epc}
mock_repo.survey_lodged_uprns.return_value = set() # public gov cert, not a survey
MockUoW = stack.enter_context(
patch("applications.modelling_e2e.handler.PostgresUnitOfWork")
)
MockUoW.return_value.__enter__.return_value = mock_uow
MockUoW.return_value.__exit__.return_value = False
# Act
_call_handler({**_BODY, "refetch_epc": True})
# Assert — the fetched cert was modelled AND persisted to heal the stale row.
assert mock_run_modelling.call_args.args[0].post_town == "REFRESHED"
mock_uow.epc.save_batch.assert_called_once_with(
[
EpcSaveRequest(
fetched_epc,
property_id=PROPERTY_ID,
portfolio_id=PORTFOLIO_ID,
source="lodged",
)
]
)
def test_refetch_does_not_overwrite_our_survey_with_a_gov_cert() -> None:
"""End-to-end: a stored survey (uploaded_file_id set) on the same date as the
gov cert is modelled and NOT re-persisted, even though the gov cert differs."""
# Arrange
mock_engine = _engine_mock([PROPERTY_ID], [UPRN], [POSTCODE])
stored_survey = _lodged_epc("2025-07-18")
fetched_gov = dataclasses.replace(_lodged_epc("2025-07-18"), post_town="GOV")
mock_plan = _plan_mock()
mock_uow = MagicMock()
mock_uow.epc.property_ids_with_lodged_epc.return_value = {PROPERTY_ID}
with ExitStack() as stack:
stack.enter_context(patch("applications.modelling_e2e.handler.os.environ", _ENV))
stack.enter_context(
patch("applications.modelling_e2e.handler._get_engine", return_value=mock_engine)
)
stack.enter_context(
patch("applications.modelling_e2e.handler.EpcClientService")
).return_value.get_by_uprn.return_value = fetched_gov
stack.enter_context(patch("applications.modelling_e2e.handler.GeospatialS3Repository"))
stack.enter_context(patch("applications.modelling_e2e.handler.GoogleSolarApiClient"))
stack.enter_context(
patch("applications.modelling_e2e.handler._spatial_for", return_value=None)
)
stack.enter_context(
patch("applications.modelling_e2e.handler._solar_insights_for", return_value=None)
)
stack.enter_context(
patch("applications.modelling_e2e.handler.overlays_from", return_value=[])
)
stack.enter_context(
patch("applications.modelling_e2e.handler.PropertyOverridesPostgresReader")
).return_value.overrides_for_many.return_value = {}
stack.enter_context(
patch("applications.modelling_e2e.handler.ScenarioPostgresRepository")
).return_value.get_many.return_value = [MagicMock()]
stack.enter_context(
patch("applications.modelling_e2e.handler.catalogue_snapshot_with_off_catalogue_overrides")
)
stack.enter_context(patch("applications.modelling_e2e.handler.Session"))
mock_run_modelling = stack.enter_context(
patch("applications.modelling_e2e.handler.run_modelling", return_value=mock_plan)
)
mock_repo = stack.enter_context(
patch("applications.modelling_e2e.handler.EpcPostgresRepository")
).return_value
mock_repo.get_for_properties.return_value = {UPRN: stored_survey}
mock_repo.survey_lodged_uprns.return_value = {UPRN} # one of our surveys
MockUoW = stack.enter_context(
patch("applications.modelling_e2e.handler.PostgresUnitOfWork")
)
MockUoW.return_value.__enter__.return_value = mock_uow
MockUoW.return_value.__exit__.return_value = False
# Act
_call_handler({**_BODY, "refetch_epc": True})
# Assert — the survey was modelled and NOT overwritten by the gov cert.
assert mock_run_modelling.call_args.args[0].post_town != "GOV"
mock_uow.epc.save_batch.assert_not_called()
def test_stored_lodged_epc_from_a_sibling_uprn_is_persisted_for_this_property() -> None:
"""A property whose UPRN matches a lodged epc_property row owned by a DIFFERENT
property (the same dwelling onboarded under another portfolio) must still get
its OWN property_id-anchored lodged row.
The lodged read keys on UPRN (ADR-0001 recency), so a sibling's row makes the
cert look "already stored" and `lodged_epc_is_new` stays False. But writes and
the FE both key on property_id, so without a per-property copy the property has
no epc_property row the FE can read — the EPC is modelled but invisible
(portfolio 850 / property 792235, whose UPRN 100021979925 also lives under
property 753896 in portfolio 830). The persist decision must therefore follow
"does THIS property have a lodged row?", not "is the cert new?".
"""
# Arrange — a stored lodged EPC is found by UPRN (it is a sibling's row), and
# this property has no lodged row of its own. refetch_epc=False reproduces the
# "already stored under the sibling" shape without an API tie-break.
mock_engine = _engine_mock([PROPERTY_ID], [UPRN], [POSTCODE])
stored_epc = MagicMock()
mock_plan = _plan_mock()
mock_uow = MagicMock()
# No lodged epc_property row exists for THIS property_id (the UPRN-matched row
# belongs to the sibling), so the per-property persist must fire.
mock_uow.epc.property_ids_with_lodged_epc.return_value = set()
with ExitStack() as stack:
stack.enter_context(patch("applications.modelling_e2e.handler.os.environ", _ENV))
stack.enter_context(
patch("applications.modelling_e2e.handler._get_engine", return_value=mock_engine)
)
stack.enter_context(
patch("applications.modelling_e2e.handler.EpcClientService")
).return_value.get_by_uprn.return_value = MagicMock()
stack.enter_context(patch("applications.modelling_e2e.handler.GeospatialS3Repository"))
stack.enter_context(patch("applications.modelling_e2e.handler.GoogleSolarApiClient"))
stack.enter_context(
patch("applications.modelling_e2e.handler._spatial_for", return_value=None)
)
stack.enter_context(
patch("applications.modelling_e2e.handler._solar_insights_for", return_value=None)
)
stack.enter_context(
patch("applications.modelling_e2e.handler.overlays_from", return_value=[])
)
stack.enter_context(
patch("applications.modelling_e2e.handler.PropertyOverridesPostgresReader")
).return_value.overrides_for_many.return_value = {}
stack.enter_context(
patch("applications.modelling_e2e.handler.ScenarioPostgresRepository")
).return_value.get_many.return_value = [MagicMock()]
stack.enter_context(
patch("applications.modelling_e2e.handler.catalogue_snapshot_with_off_catalogue_overrides")
)
stack.enter_context(patch("applications.modelling_e2e.handler.Session"))
stack.enter_context(
patch("applications.modelling_e2e.handler.run_modelling", return_value=mock_plan)
)
# The lodged read (UPRN-keyed) surfaces the sibling's row for this UPRN.
stack.enter_context(
patch("applications.modelling_e2e.handler.EpcPostgresRepository")
).return_value.get_for_properties.return_value = {UPRN: stored_epc}
MockUoW = stack.enter_context(
patch("applications.modelling_e2e.handler.PostgresUnitOfWork")
)
MockUoW.return_value.__enter__.return_value = mock_uow
MockUoW.return_value.__exit__.return_value = False
# Act
_call_handler({**_BODY, "refetch_epc": False})
# Assert — the stored lodged EPC is persisted under THIS property_id so the FE
# (which reads epc_property by property_id) can see it.
mock_uow.epc.save_batch.assert_called_once_with(
[
EpcSaveRequest(
stored_epc,
property_id=PROPERTY_ID,
portfolio_id=PORTFOLIO_ID,
source="lodged",
)
]
)
def test_refetch_epc_true_always_calls_api_even_if_stored_epc_exists() -> None:
"""refetch_epc=True (default): EpcClientService.get_by_uprn is called even
when a stored lodged EPC exists — existing behaviour is preserved."""
@ -1846,10 +2148,13 @@ def test_repredict_epc_false_with_stored_predicted_epc_skips_prediction() -> Non
mock_predictor = stack.enter_context(
patch("applications.modelling_e2e.handler.EpcPrediction")
).return_value
# Stored predicted EPC is present
stack.enter_context(
# No lodged EPC (so the prediction path is actually exercised); a stored
# predicted EPC is present.
mock_epc_repo = stack.enter_context(
patch("applications.modelling_e2e.handler.EpcPostgresRepository")
).return_value.get_predicted_for_properties.return_value = {
).return_value
mock_epc_repo.get_for_properties.return_value = {}
mock_epc_repo.get_predicted_for_properties.return_value = {
PROPERTY_ID: stored_predicted
}
MockUoW = stack.enter_context(

17
tests/orchestration/fakes.py
View file

@ -137,6 +137,23 @@ class FakeEpcRepo(EpcRepository):
def save_batch(self, requests: list[EpcSaveRequest]) -> list[int]:
return [self.save(r.data, r.property_id, r.portfolio_id, r.source) for r in requests]
def survey_lodged_uprns(self, uprns: list[int]) -> set[int]:
# The fake writes no uploaded_file_id provenance (surveys aren't modelled
# through it), so it reports no rows as surveys — every stored lodged EPC
# behaves as a public gov cert.
return set()
def property_ids_with_lodged_epc(self, property_ids: list[int]) -> set[int]:
# Lodged saves that carried a property_id record it here (mirrors the
# postgres adapter's property_id-keyed lookup, distinct from the
# UPRN-keyed lodged read).
saved_lodged = {
pid
for (_data, pid), source in zip(self.saved, self.sources)
if source not in PREDICTED_SLOT_SOURCES and pid is not None
}
return {pid for pid in property_ids if pid in saved_lodged}
class FakeSolarRepo(SolarRepository):
"""In-memory Google Solar insights store keyed by UPRN. Seed `by_uprn` to

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

@ -167,3 +167,70 @@ def test_lodged_and_predicted_batch_slots_are_independent(db_engine: Engine) ->
# Assert — both slots are populated for both properties.
assert lodged == epc_by_uprn
assert predicted == epc_by_uprn
# ---------------------------------------------------------------------------
# property_ids_with_lodged_epc keys on property_id, not the shared UPRN
# ---------------------------------------------------------------------------
def test_property_ids_with_lodged_epc_ignores_a_uprn_shared_siblings_row(
db_engine: Engine,
) -> None:
# Arrange — one dwelling (uprn 5555) onboarded under property 5001 only. Its
# sibling 5002 shares the UPRN but has no lodged row of its own; 5003 holds
# only a predicted row.
owner, sibling, predicted_only = 5001, 5002, 5003
lodged = replace(_load_epc(), uprn=5555)
with Session(db_engine) as session:
repo = EpcPostgresRepository(session)
repo.save_batch([EpcSaveRequest(lodged, property_id=owner, source="lodged")])
repo.save_batch(
[EpcSaveRequest(lodged, property_id=predicted_only, source="predicted")]
)
session.commit()
# Act
with Session(db_engine) as session:
repo = EpcPostgresRepository(session)
result = repo.property_ids_with_lodged_epc(
[owner, sibling, predicted_only]
)
# Assert — only the property that owns a lodged row keyed on its OWN
# property_id counts; the UPRN-shared sibling and the predicted-only property
# do not (this is what drives the per-property persist so the FE sees the EPC).
assert result == {owner}
# ---------------------------------------------------------------------------
# survey_lodged_uprns distinguishes our uploaded surveys from gov certs
# ---------------------------------------------------------------------------
def test_survey_lodged_uprns_flags_only_uploaded_file_rows(db_engine: Engine) -> None:
from sqlalchemy import text
# Arrange — two lodged EPCs saved as public gov certs (uploaded_file_id null);
# then one is marked as an uploaded survey the way the survey writers do.
survey_uprn, gov_uprn = 6001, 6002
epc_by_uprn = {uprn: replace(_load_epc(), uprn=uprn) for uprn in (survey_uprn, gov_uprn)}
with Session(db_engine) as session:
repo = EpcPostgresRepository(session)
repo.save_batch(
[EpcSaveRequest(epc_by_uprn[u], property_id=u, source="lodged") for u in epc_by_uprn]
)
session.commit()
session.exec( # type: ignore[call-overload]
text("UPDATE epc_property SET uploaded_file_id = 999 WHERE uprn = :u"),
params={"u": survey_uprn},
)
session.commit()
# Act
with Session(db_engine) as session:
repo = EpcPostgresRepository(session)
result = repo.survey_lodged_uprns([survey_uprn, gov_uprn])
# Assert — only the uploaded-file row counts as a survey.
assert result == {survey_uprn}