Model/tests/repositories/property/test_property_postgres_overlay_hydration.py
Jun-te Kim 4c038ae8dc Hydrate landlord-override overlays onto the Property from property_overrides 🟩
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-16 17:29:31 +00:00

114 lines
4 KiB
Python

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