From 4c038ae8dcfa7a0fbabccdee76cd7566a0a5d537 Mon Sep 17 00:00:00 2001 From: Jun-te Kim Date: Tue, 16 Jun 2026 17:29:31 +0000 Subject: [PATCH] =?UTF-8?q?Hydrate=20landlord-override=20overlays=20onto?= =?UTF-8?q?=20the=20Property=20from=20property=5Foverrides=20=F0=9F=9F=A9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 4.8 (1M context) --- .../property/landlord_override_overlays.py | 27 +++++ .../property/property_postgres_repository.py | 14 +++ ...est_property_postgres_overlay_hydration.py | 114 ++++++++++++++++++ 3 files changed, 155 insertions(+) create mode 100644 repositories/property/landlord_override_overlays.py create mode 100644 tests/repositories/property/test_property_postgres_overlay_hydration.py diff --git a/repositories/property/landlord_override_overlays.py b/repositories/property/landlord_override_overlays.py new file mode 100644 index 00000000..3f73163c --- /dev/null +++ b/repositories/property/landlord_override_overlays.py @@ -0,0 +1,27 @@ +"""Map a Property's resolved Landlord Overrides to Simulation Overlays (ADR-0032). + +The boundary between the faithful `property_overrides` read model +(`ResolvedPropertyOverrides`, value-space) and the domain overlay surface +(`EpcSimulation`). Lives in `repositories/` because it consumes a repository +type — `domain/` never imports `repositories/`. Per-component and partial: an +override produces an overlay only where a component mapping exists and resolves +(the tracer covers `wall_type`); everything else is left to the lodged EPC. +""" + +from __future__ import annotations + +from domain.epc.wall_type_overlay import wall_overlay_for +from domain.modelling.simulation import EpcSimulation +from repositories.property.property_overrides_reader import ResolvedPropertyOverrides + +_WALL_TYPE = "wall_type" + + +def overlays_from(overrides: ResolvedPropertyOverrides) -> list[EpcSimulation]: + overlays: list[EpcSimulation] = [] + for row in overrides.rows: + if row.override_component == _WALL_TYPE: + overlay = wall_overlay_for(row.override_value, row.building_part) + if overlay is not None: + overlays.append(overlay) + return overlays diff --git a/repositories/property/property_postgres_repository.py b/repositories/property/property_postgres_repository.py index 3549d0fc..db0bda00 100644 --- a/repositories/property/property_postgres_repository.py +++ b/repositories/property/property_postgres_repository.py @@ -8,10 +8,13 @@ from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlmodel import Session, col, select from domain.geospatial.planning_restrictions import PlanningRestrictions +from domain.modelling.simulation import EpcSimulation from domain.property.properties import Properties from domain.property.property import Property, PropertyIdentity from infrastructure.postgres.property_table import PropertyRow from repositories.epc.epc_repository import EpcRepository +from repositories.property.landlord_override_overlays import overlays_from +from repositories.property.property_overrides_reader import PropertyOverridesReader from repositories.property.property_repository import ( PropertyIdentityInsert, PropertyRepository, @@ -36,10 +39,12 @@ class PropertyPostgresRepository(PropertyRepository): session: Session, epc_repo: Optional[EpcRepository] = None, spatial_repo: Optional[SpatialRepository] = None, + overrides_reader: Optional[PropertyOverridesReader] = None, ) -> None: self._session = session self._epc_repo = epc_repo self._spatial_repo = spatial_repo + self._overrides_reader = overrides_reader # ``__table__`` is injected at runtime on table=True classes but the stubs # don't expose it; pin to ``Table`` so the dialect insert is typed. self._table: Table = cast(Table, getattr(PropertyRow, "__table__")) @@ -52,6 +57,13 @@ class PropertyPostgresRepository(PropertyRepository): ) return self._epc_repo + def _landlord_overrides(self, property_id: int) -> list[EpcSimulation]: + """The Property's Landlord Overrides as Simulation Overlays — empty when + no reader is wired (the overlay stays off) or the Property has none.""" + if self._overrides_reader is None: + return [] + return overlays_from(self._overrides_reader.overrides_for(property_id)) + def get(self, property_id: int) -> Property: row = self._session.get(PropertyRow, property_id) if row is None: @@ -73,6 +85,7 @@ class PropertyPostgresRepository(PropertyRepository): identity=identity, epc=self._epc().get_for_property(property_id), predicted_epc=self._epc().get_predicted_for_property(property_id), + landlord_overrides=self._landlord_overrides(property_id), planning_restrictions=_restrictions_of(row.uprn, restrictions), ) @@ -104,6 +117,7 @@ class PropertyPostgresRepository(PropertyRepository): ), epc=epcs.get(property_id), predicted_epc=predicted_epcs.get(property_id), + landlord_overrides=self._landlord_overrides(property_id), planning_restrictions=_restrictions_of(row.uprn, restrictions), ) ) diff --git a/tests/repositories/property/test_property_postgres_overlay_hydration.py b/tests/repositories/property/test_property_postgres_overlay_hydration.py new file mode 100644 index 00000000..126777c3 --- /dev/null +++ b/tests/repositories/property/test_property_postgres_overlay_hydration.py @@ -0,0 +1,114 @@ +"""PropertyPostgresRepository hydrates Landlord Overrides as overlays (ADR-0032). + +Real Postgres end-to-end for the read side: a lodged EPC and a `wall_type` +override row are persisted, and the reloaded Property's Effective EPC reflects +the override folded onto the lodged wall — proving reader → overlay → aggregate. +""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +from sqlalchemy import Engine +from sqlmodel import Session + +from datatypes.epc.domain.epc_property_data import ( + BuildingPartIdentifier, + EpcPropertyData, +) +from datatypes.epc.domain.mapper import EpcPropertyDataMapper +from infrastructure.postgres.property_override_table import PropertyOverrideRow +from infrastructure.postgres.property_table import PropertyRow +from repositories.epc.epc_postgres_repository import EpcPostgresRepository +from repositories.property.property_overrides_postgres_reader import ( + PropertyOverridesPostgresReader, +) +from repositories.property.property_postgres_repository import ( + PropertyPostgresRepository, +) + +_JSON_SAMPLES = Path(__file__).resolve().parents[3] / "backend/epc_api/json_samples" + + +def _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 test_reloaded_property_effective_epc_reflects_the_wall_override( + db_engine: Engine, +) -> None: + # Arrange — a Property with a lodged EPC (cavity main wall) and a solid-brick + # / internal-insulation wall override. + with Session(db_engine) as session: + row = PropertyRow(portfolio_id=1, postcode="A0 0AA", address="1 St", uprn=1) + session.add(row) + session.commit() + property_id = row.id + assert property_id is not None + + EpcPostgresRepository(session).save(_epc(), property_id=property_id) + session.add( + PropertyOverrideRow( + property_id=property_id, + portfolio_id=1, + building_part=0, + override_component="wall_type", + override_value="Solid brick, with internal insulation", + original_spreadsheet_description="solid brick, insulated", + ) + ) + session.commit() + + # Act — reload through the real repository with the overrides reader wired. + with Session(db_engine) as session: + repo = PropertyPostgresRepository( + session, + EpcPostgresRepository(session), + overrides_reader=PropertyOverridesPostgresReader(lambda: Session(db_engine)), + ) + prop = repo.get(property_id) + + main = next( + part + for part in prop.effective_epc.sap_building_parts + if part.identifier is BuildingPartIdentifier.MAIN + ) + + # Assert — the lodged cavity wall (4) is overlaid to solid brick (3) / internal (3). + assert main.wall_construction == 3 + assert main.wall_insulation_type == 3 + + +def test_property_without_overrides_keeps_its_lodged_wall(db_engine: Engine) -> None: + # Arrange — a Property with a lodged EPC but no override rows. + with Session(db_engine) as session: + row = PropertyRow(portfolio_id=1, postcode="A0 0AA", address="2 St", uprn=2) + session.add(row) + session.commit() + property_id = row.id + assert property_id is not None + EpcPostgresRepository(session).save(_epc(), property_id=property_id) + session.commit() + + # Act + with Session(db_engine) as session: + repo = PropertyPostgresRepository( + session, + EpcPostgresRepository(session), + overrides_reader=PropertyOverridesPostgresReader(lambda: Session(db_engine)), + ) + prop = repo.get(property_id) + + main = next( + part + for part in prop.effective_epc.sap_building_parts + if part.identifier is BuildingPartIdentifier.MAIN + ) + + # Assert — the lodged cavity wall (4) is untouched. + assert main.wall_construction == 4