test(orchestration): e2e — ingested listed UPRN blocks solid-wall insulation

Slice 3c.6. The integrating proof through real Postgres: two solid-brick
uninsulated dwellings, identical but for the planning status Ingestion caches
per UPRN. Ingestion writes the spatial reference; Modelling reads it back off
the Property and gates the wall measures — the listed dwelling gets neither
EWI nor IWI, the unrestricted one gets a wall measure. Closes slice 3c
(ADR-0019/ADR-0020).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

tests/orchestration/test_ara_first_run_pipeline_integration.py

@@ -23,6 +23,11 @@ from domain.sap10_calculator.calculator import Sap10Calculator
 from domain.modelling.portfolio_goal import PortfolioGoal
 from infrastructure.postgres.modelling import ScenarioModel
 from domain.geospatial.coordinates import Coordinates
+from domain.geospatial.planning_restrictions import PlanningRestrictions
+from domain.geospatial.spatial_reference import SpatialReference
+from tests.domain.modelling._elmhurst_recommendation import (
+    parse_recommendation_summary,
+)
 from infrastructure.postgres.property_baseline_performance_table import (
     PropertyBaselinePerformanceModel,
 )
@@ -439,3 +444,151 @@ def test_modelling_recommends_nothing_when_already_at_the_target_band(
     assert plan.post_energy_consumption > 0.0
     assert plan.energy_bill_savings == 0.0
     assert plan.energy_consumption_savings == 0.0
+
+
+class _NoEpcFetcher:
+    """An EPC fetcher that returns nothing — the EPC is seeded directly so this
+    e2e drives only the spatial-reference half of Ingestion."""
+
+    def get_by_uprn(self, uprn: int) -> Optional[EpcPropertyData]:
+        return None
+
+
+class _SpatialByUprn(GeospatialRepository):
+    """Resolves a per-UPRN spatial reference (coordinates nulled — the Solar leg
+    is not under test)."""
+
+    def __init__(self, by_uprn: dict[int, SpatialReference]) -> None:
+        self._by_uprn = by_uprn
+
+    def coordinates_for(self, uprn: int) -> Optional[Coordinates]:
+        return None
+
+    def spatial_for(self, uprn: int) -> Optional[SpatialReference]:
+        return self._by_uprn.get(uprn)
+
+
+def test_listed_uprn_ingested_blocks_solid_wall_insulation_in_modelling(
+    db_engine: Engine,
+) -> None:
+    # Arrange — two solid-brick uninsulated dwellings: one in a listed building,
+    # one unrestricted. Ingestion caches each UPRN's planning protections; the
+    # EPC is seeded directly (the solid-wall mechanics are pinned elsewhere).
+    listed_reference = SpatialReference(
+        coordinates=None, restrictions=PlanningRestrictions(is_listed=True)
+    )
+    unrestricted_reference = SpatialReference(
+        coordinates=None, restrictions=PlanningRestrictions()
+    )
+    solid_brick_epc = parse_recommendation_summary("solid_brick_ewi_001431_before.pdf")
+    with Session(db_engine) as session:
+        session.add_all(
+            [
+                PropertyRow(
+                    id=40,
+                    portfolio_id=1,
+                    postcode="A0 0AA",
+                    address="Listed House",
+                    uprn=44444,
+                ),
+                PropertyRow(
+                    id=41,
+                    portfolio_id=1,
+                    postcode="A0 0AA",
+                    address="Unrestricted House",
+                    uprn=55555,
+                ),
+                ScenarioModel(
+                    id=7,
+                    goal=PortfolioGoal.INCREASING_EPC,
+                    goal_value="C",
+                    is_default=True,
+                ),
+            ]
+        )
+        # The solid-brick EPC fires the floor + solid-wall Generators and the
+        # ventilation dependency, so every Product they reach for must exist.
+        session.add_all(
+            [
+                MaterialRow(
+                    id=1,
+                    type="external_wall_insulation",
+                    total_cost=100.0,
+                    cost_unit="gbp_per_m2",
+                    is_active=True,
+                    description="External wall insulation",
+                ),
+                MaterialRow(
+                    id=2,
+                    type="internal_wall_insulation",
+                    total_cost=90.0,
+                    cost_unit="gbp_per_m2",
+                    is_active=True,
+                    description="Internal wall insulation",
+                ),
+                MaterialRow(
+                    id=3,
+                    type="solid_floor_insulation",
+                    total_cost=25.0,
+                    cost_unit="gbp_per_m2",
+                    is_active=True,
+                    description="Solid floor insulation",
+                ),
+                MaterialRow(
+                    id=4,
+                    type="mechanical_ventilation",
+                    total_cost=450.0,
+                    cost_unit="gbp_per_unit",
+                    is_active=True,
+                    description="Mechanical extract ventilation unit",
+                ),
+            ]
+        )
+        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)
+        session.commit()
+
+    def unit_of_work() -> PostgresUnitOfWork:
+        return PostgresUnitOfWork(lambda: Session(db_engine))
+
+    geospatial_repo = _SpatialByUprn(
+        {44444: listed_reference, 55555: unrestricted_reference}
+    )
+
+    # Act — Ingestion caches the protections per UPRN, then Modelling reads them
+    # back off the Property (through the repo) and gates the solid-wall measures.
+    IngestionOrchestrator(
+        unit_of_work=unit_of_work,
+        epc_fetcher=_NoEpcFetcher(),
+        geospatial_repo=geospatial_repo,
+        solar_fetcher=_UnusedSolarFetcher(),
+    ).run([40, 41])
+    ModellingOrchestrator(
+        unit_of_work=unit_of_work,
+        calculator=Sap10Calculator(),
+        fuel_rates=FuelRatesStaticFileRepository(),
+    ).run(property_ids=[40, 41], scenario_ids=[7], portfolio_id=1)
+
+    # Assert — the listed dwelling gets no wall insulation (both Options blocked);
+    # the unrestricted dwelling gets one. The only difference between them is the
+    # planning status Ingestion cached, proving the loop end to end (ADR-0019/0020).
+    _WALL_TYPES = {"external_wall_insulation", "internal_wall_insulation"}
+    with Session(db_engine) as session:
+        listed_types = _plan_measure_types(session, property_id=40)
+        unrestricted_types = _plan_measure_types(session, property_id=41)
+
+    assert _WALL_TYPES.isdisjoint(listed_types)
+    assert _WALL_TYPES & unrestricted_types
+
+
+def _plan_measure_types(session: Session, *, property_id: int) -> set[str]:
+    plan = session.exec(
+        select(PlanModel).where(col(PlanModel.property_id) == property_id)
+    ).first()
+    assert plan is not None
+    rec_rows = session.exec(
+        select(RecommendationModel).where(col(RecommendationModel.plan_id) == plan.id)
+    ).all()
+    return {rec.type for rec in rec_rows}