From 7e5af6c8f4144a5c89e21bf34e47064615ee2ead Mon Sep 17 00:00:00 2001 From: Daniel Roth Date: Mon, 22 Jun 2026 15:46:18 +0000 Subject: [PATCH] process multiple properties in one message --- applications/modelling_e2e/handler.py | 322 ++++++--- .../local_handler/invoke_local_lambda.py | 4 +- .../modelling_e2e_trigger_body.py | 2 +- tests/applications/modelling_e2e/__init__.py | 0 .../modelling_e2e/test_handler.py | 679 ++++++++++++++++++ 5 files changed, 918 insertions(+), 89 deletions(-) create mode 100644 tests/applications/modelling_e2e/__init__.py create mode 100644 tests/applications/modelling_e2e/test_handler.py diff --git a/applications/modelling_e2e/handler.py b/applications/modelling_e2e/handler.py index 7c8bc4f9..95af5269 100644 --- a/applications/modelling_e2e/handler.py +++ b/applications/modelling_e2e/handler.py @@ -1,10 +1,16 @@ -"""SQS-triggered Lambda: fetch EPC → run modelling → persist plan. +"""SQS-triggered Lambda: fetch EPC (or predict) → run modelling → persist plan. -One SQS message = one property. The handler reads ``property_id``, -``portfolio_id``, ``scenario_id``, and ``no_solar`` from the message body, -fetches the property's EPC from the gov API, runs the full modelling pipeline -(SAP10 → optimiser) via ``harness.console.run_modelling``, and persists the -resulting Plan via ``PlanPostgresRepository.save()``. +One SQS message = one batch of properties sharing a portfolio, scenario, and +(by caller convention) postcode. The handler reads ``property_ids``, +``portfolio_id``, ``scenario_id``, ``no_solar``, and ``dry_run`` from the +message body, fetches or predicts each property's EPC, runs the full modelling +pipeline (SAP10 → optimiser) via ``harness.console.run_modelling``, and +persists the resulting Plan via ``PostgresUnitOfWork`` in one atomic transaction +per property. + +When no lodged EPC is found, EPC Prediction (Path 3, ADR-0031) synthesises one +from the postcode cohort. ``_cohort_cache`` is module-level so warm Lambda +containers re-processing the same postcode avoid redundant fetches. ``secondary_heating_removal`` is excluded unconditionally: the live ``material`` catalogue does not yet carry this measure type, causing a crash during catalogue @@ -18,6 +24,7 @@ from __future__ import annotations import io import os +from collections.abc import Callable from typing import Any, Optional, cast import boto3 @@ -25,7 +32,17 @@ import pandas as pd # pyright: ignore[reportMissingTypeStubs] from sqlalchemy import Engine, text from sqlmodel import Session -from datatypes.epc.domain.epc_property_data import EpcPropertyData +from datatypes.epc.domain.epc_property_data import ( + BuildingPartIdentifier, + EpcPropertyData, +) +from domain.epc_prediction.comparable_properties import ( + ComparableProperty, + select_comparables, +) +from domain.epc_prediction.epc_prediction import EpcPrediction +from domain.epc_prediction.prediction_target import build_prediction_target +from domain.geospatial.coordinates import Coordinates from domain.geospatial.planning_restrictions import PlanningRestrictions from domain.geospatial.spatial_reference import SpatialReference from domain.modelling.measure_type import MeasureType @@ -42,20 +59,22 @@ from infrastructure.solar.google_solar_api_client import ( from applications.modelling_e2e.modelling_e2e_trigger_body import ( ModellingE2ETriggerBody, ) +from repositories.comparable_properties.epc_comparable_properties_repository import ( + EpcComparablePropertiesRepository, +) from repositories.geospatial.geospatial_s3_repository import ( GeospatialS3Repository, ParquetReader, ) -from repositories.epc.epc_postgres_repository import EpcPostgresRepository -from repositories.plan.plan_postgres_repository import PlanPostgresRepository +from repositories.postgres_unit_of_work import PostgresUnitOfWork from repositories.product.product_postgres_repository import ProductPostgresRepository from repositories.property.landlord_override_overlays import overlays_from +from repositories.property.override_backed_prediction_attributes_reader import ( + OverrideBackedPredictionAttributesReader, +) from repositories.property.property_overrides_postgres_reader import ( PropertyOverridesPostgresReader, ) -from repositories.property.property_postgres_repository import ( - PropertyPostgresRepository, -) from repositories.scenario.scenario_postgres_repository import ( ScenarioPostgresRepository, ) @@ -63,6 +82,7 @@ from utilities.aws_lambda.task_handler import task_handler from utilities.logger import setup_logger _engine: Optional[Engine] = None +_cohort_cache: dict[str, list[ComparableProperty]] = {} logger = setup_logger() @@ -109,17 +129,53 @@ def _solar_insights_for( return None +def _predict_epc( + *, + property_id: int, + uprn: int, + postcode: str, + portfolio_id: int, + attributes_reader: OverrideBackedPredictionAttributesReader, + coordinates: Optional[Coordinates], + cohort_for: Callable[[str], list[ComparableProperty]], + predictor: EpcPrediction, +) -> Optional[EpcPropertyData]: + """Synthesise an EpcPropertyData for an EPC-less property from its postcode + cohort (EPC Prediction Path 3, ADR-0031), or None when ineligible. + + Returns None when property_type is unresolvable (hard cohort filter cannot + fire) or when the postcode cohort is empty after filtering. + """ + attributes = attributes_reader.attributes_for(property_id) + identity = PropertyIdentity( + portfolio_id=portfolio_id, postcode=postcode, address="", uprn=uprn + ) + target = build_prediction_target(identity, coordinates, attributes) + if target is None: + return None + comparables = select_comparables(target, cohort_for(target.postcode)) + if not comparables.members: + return None + predicted = predictor.predict(target, comparables) + if not any( + part.identifier is BuildingPartIdentifier.MAIN + for part in predicted.sap_building_parts + ): + return None + return predicted + + @task_handler(task_source="modelling_e2e", source=Source.PROPERTY) def handler(body: dict[str, Any], context: Any) -> None: trigger = ModellingE2ETriggerBody.model_validate(body) - property_id = trigger.property_id + property_ids = trigger.property_ids portfolio_id = trigger.portfolio_id scenario_id = trigger.scenario_id no_solar = trigger.no_solar dry_run = trigger.dry_run logger.info( - f"start property={property_id} portfolio={portfolio_id} " + f"start property_ids={property_ids} portfolio={portfolio_id} " f"scenario={scenario_id} no_solar={no_solar} dry_run={dry_run}" ) @@ -129,84 +185,178 @@ def handler(body: dict[str, Any], context: Any) -> None: solar_client = GoogleSolarApiClient(os.environ["GOOGLE_SOLAR_API_KEY"]) with engine.connect() as conn: - row = conn.execute( - text("SELECT uprn FROM property WHERE id = :pid"), - {"pid": property_id}, - ).one() - uprn = int(row[0]) - logger.info(f"resolved uprn={uprn}") + uprn_rows = conn.execute( + text("SELECT id, uprn FROM property WHERE id = ANY(:ids)"), + {"ids": property_ids}, + ).fetchall() + postcode_rows = conn.execute( + text("SELECT id, postcode FROM property WHERE id = ANY(:ids)"), + {"ids": property_ids}, + ).fetchall() - epc: Optional[EpcPropertyData] = epc_client.get_by_uprn(uprn) - if epc is None: - raise ValueError(f"no EPC found for UPRN {uprn} (property {property_id})") - logger.info(f"fetched EPC (energy_rating_current={epc.energy_rating_current})") + uprns: dict[int, int] = {int(row[0]): int(row[1]) for row in uprn_rows} + postcodes: dict[int, str] = { + int(row[0]): (row[1] or "") for row in postcode_rows + } overrides_reader = PropertyOverridesPostgresReader(lambda: Session(engine)) - overlaid = Property( - identity=PropertyIdentity( - portfolio_id=portfolio_id, postcode="", address="", uprn=uprn - ), - epc=epc, - landlord_overrides=overlays_from(overrides_reader.overrides_for(property_id)), - ) - effective_epc = overlaid.effective_epc + prediction_attrs_reader = OverrideBackedPredictionAttributesReader(overrides_reader) + comparables_repo = EpcComparablePropertiesRepository(epc_client, geospatial) + predictor = EpcPrediction() - spatial = _spatial_for(geospatial, uprn) - restrictions = ( - spatial.restrictions if spatial is not None else PlanningRestrictions() - ) - logger.info(f"spatial={'found' if spatial is not None else 'not found'}") + def _get_cohort(postcode: str) -> list[ComparableProperty]: + if postcode not in _cohort_cache: + _cohort_cache[postcode] = ( + comparables_repo.candidates_for(postcode) if postcode else [] + ) + return _cohort_cache[postcode] - if no_solar: - solar_insights = None - logger.info("solar skipped (no_solar=True)") - else: - solar_insights = _solar_insights_for(solar_client, spatial) - logger.info(f"solar={'found' if solar_insights is not None else 'not found'}") + read_session = Session(engine) + try: + scenario = ScenarioPostgresRepository(read_session).get_many([scenario_id])[0] + products = ProductPostgresRepository(read_session) - with Session(engine) as session: - scenario = ScenarioPostgresRepository(session).get_many([scenario_id])[0] - logger.info(f"loaded scenario goal={scenario.goal!r} goal_value={scenario.goal_value!r}") - products = ProductPostgresRepository(session) + errors: list[int] = [] - # secondary_heating_removal is absent from the live material.type enum; - # exclude it unconditionally until the catalogue gap is resolved. - considered: Optional[frozenset[MeasureType]] = frozenset(MeasureType) - { - MeasureType.SECONDARY_HEATING_REMOVAL - } + for property_id in property_ids: + try: + uprn = uprns[property_id] + postcode = postcodes.get(property_id, "") + logger.info( + f"property={property_id} uprn={uprn} postcode={postcode!r}" + ) - logger.info("running modelling pipeline") - plan = run_modelling( - effective_epc, - planning_restrictions=restrictions, - solar_insights=solar_insights, - considered_measures=considered, - products=products, - scenario=scenario, - print_table=False, - ) - logger.info( - f"modelling complete: SAP {plan.baseline.sap_continuous:.1f}→" - f"{plan.post_sap_continuous:.1f} measures={len(plan.measures)} " - f"cost=£{plan.cost_of_works:,.0f}" - ) + spatial = _spatial_for(geospatial, uprn) + restrictions = ( + spatial.restrictions + if spatial is not None + else PlanningRestrictions() + ) + coordinates: Optional[Coordinates] = ( + spatial.coordinates if spatial is not None else None + ) - if dry_run: - measure_types = ", ".join(m.measure_type for m in plan.measures) or "none" - logger.info(f"[dry_run] measures=[{measure_types}] — skipping DB write") - return - EpcPostgresRepository(session).save( - epc, property_id=property_id, portfolio_id=portfolio_id - ) - PlanPostgresRepository(session).save( - plan, - property_id=property_id, - scenario_id=scenario_id, - portfolio_id=portfolio_id, - is_default=scenario.is_default, - ) - PropertyPostgresRepository(session).mark_modelled( - property_id, has_recommendations=bool(plan.measures) - ) - session.commit() - logger.info("plan saved") + epc: Optional[EpcPropertyData] = epc_client.get_by_uprn(uprn) + overrides = overlays_from(overrides_reader.overrides_for(property_id)) + + if epc is not None: + logger.info(f"property={property_id} lodged EPC found") + effective_epc = Property( + identity=PropertyIdentity( + portfolio_id=portfolio_id, + postcode=postcode, + address="", + uprn=uprn, + ), + epc=epc, + landlord_overrides=overrides, + ).effective_epc + else: + logger.info( + f"property={property_id} no lodged EPC — attempting prediction" + ) + predicted_epc = _predict_epc( + property_id=property_id, + uprn=uprn, + postcode=postcode, + portfolio_id=portfolio_id, + attributes_reader=prediction_attrs_reader, + coordinates=coordinates, + cohort_for=_get_cohort, + predictor=predictor, + ) + if predicted_epc is None: + raise ValueError( + f"no EPC for UPRN {uprn} and not predictable " + f"(unresolved property_type or empty '{postcode}' cohort)" + ) + effective_epc = Property( + identity=PropertyIdentity( + portfolio_id=portfolio_id, + postcode=postcode, + address="", + uprn=uprn, + ), + epc=None, + predicted_epc=predicted_epc, + landlord_overrides=overrides, + ).effective_epc + + solar_insights: Optional[dict[str, Any]] = ( + None + if no_solar + else _solar_insights_for(solar_client, spatial) + ) + + # secondary_heating_removal is absent from the live material.type + # enum; exclude unconditionally until the catalogue gap is resolved. + considered: Optional[frozenset[MeasureType]] = frozenset( + MeasureType + ) - {MeasureType.SECONDARY_HEATING_REMOVAL} + + plan = run_modelling( + effective_epc, + planning_restrictions=restrictions, + solar_insights=solar_insights, + considered_measures=considered, + products=products, + scenario=scenario, + print_table=False, + ) + logger.info( + f"property={property_id} modelling complete " + f"measures={len(plan.measures)}" + ) + + if dry_run: + measure_types = ( + ", ".join(m.measure_type for m in plan.measures) or "none" + ) + logger.info( + f"[dry_run] property={property_id} " + f"measures=[{measure_types}] — skipping DB write" + ) + continue + + with PostgresUnitOfWork(lambda: Session(engine)) as uow: + if epc is not None: + uow.epc.save( + epc, property_id=property_id, portfolio_id=portfolio_id + ) + if spatial is not None: + uow.spatial.save(uprn, spatial) + if ( + solar_insights is not None + and spatial is not None + and spatial.coordinates is not None + ): + uow.solar.save( + uprn, + longitude=spatial.coordinates.longitude, + latitude=spatial.coordinates.latitude, + insights=solar_insights, + ) + uow.plan.save( + plan, + property_id=property_id, + scenario_id=scenario_id, + portfolio_id=portfolio_id, + is_default=scenario.is_default, + ) + uow.property.mark_modelled( + property_id, has_recommendations=bool(plan.measures) + ) + uow.commit() + logger.info(f"property={property_id} plan saved") + + except Exception as error: # noqa: BLE001 + logger.error( + f"property={property_id}: {type(error).__name__}: {error}", + exc_info=True, + ) + errors.append(property_id) + + if errors: + raise RuntimeError(f"failed property_ids: {errors}") + finally: + read_session.close() diff --git a/applications/modelling_e2e/local_handler/invoke_local_lambda.py b/applications/modelling_e2e/local_handler/invoke_local_lambda.py index c88d4362..c667ffc6 100644 --- a/applications/modelling_e2e/local_handler/invoke_local_lambda.py +++ b/applications/modelling_e2e/local_handler/invoke_local_lambda.py @@ -12,11 +12,11 @@ payload = { { "body": json.dumps( { - "property_id": 709774, + "property_ids": [709773, 709774], "portfolio_id": 796, "scenario_id": 1268, "no_solar": False, - "dry_run": True, + "dry_run": False, } ) } diff --git a/applications/modelling_e2e/modelling_e2e_trigger_body.py b/applications/modelling_e2e/modelling_e2e_trigger_body.py index 908f9bd2..cb83118a 100644 --- a/applications/modelling_e2e/modelling_e2e_trigger_body.py +++ b/applications/modelling_e2e/modelling_e2e_trigger_body.py @@ -4,7 +4,7 @@ from pydantic import BaseModel, ConfigDict class ModellingE2ETriggerBody(BaseModel): model_config = ConfigDict(extra="allow") - property_id: int + property_ids: list[int] portfolio_id: int scenario_id: int no_solar: bool = False diff --git a/tests/applications/modelling_e2e/__init__.py b/tests/applications/modelling_e2e/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/applications/modelling_e2e/test_handler.py b/tests/applications/modelling_e2e/test_handler.py new file mode 100644 index 00000000..5979fe1b --- /dev/null +++ b/tests/applications/modelling_e2e/test_handler.py @@ -0,0 +1,679 @@ +"""Tests for the modelling_e2e Lambda handler. + +Tests exercise the handler's external behaviour through handler.__wrapped__, +patching I/O boundaries (EPC client, DB reads, UoW) so no real DB or network +is needed. One test per distinct behaviour path. +""" + +from __future__ import annotations + +from contextlib import ExitStack +from typing import Any +from unittest.mock import MagicMock, call, patch + +import pytest +from pydantic import ValidationError + +from applications.modelling_e2e.modelling_e2e_trigger_body import ( + ModellingE2ETriggerBody, +) + +PROPERTY_ID = 12345 +UPRN = 987654321 +POSTCODE = "SW1A 1AA" +PORTFOLIO_ID = 100 +SCENARIO_ID = 200 + +_ENV = { + "OPEN_EPC_API_TOKEN": "test-token", + "DATA_BUCKET": "test-bucket", + "GOOGLE_SOLAR_API_KEY": "test-solar-key", +} + +_BODY = { + "property_ids": [PROPERTY_ID], + "portfolio_id": PORTFOLIO_ID, + "scenario_id": SCENARIO_ID, + "no_solar": True, + "dry_run": False, +} + + +def _call_handler(body: dict[str, Any]) -> Any: + from applications.modelling_e2e.handler import handler + + return handler.__wrapped__(body, None) # type: ignore[attr-defined] + + +def _engine_mock( + property_ids: list[int], + uprns: list[int], + postcodes: list[str], +) -> MagicMock: + """Mock engine whose connect() returns UPRN then postcode rows.""" + mock_engine = MagicMock() + mock_conn = mock_engine.connect.return_value.__enter__.return_value + + uprn_result = MagicMock() + uprn_result.fetchall.return_value = list(zip(property_ids, uprns)) + + postcode_result = MagicMock() + postcode_result.fetchall.return_value = list(zip(property_ids, postcodes)) + + mock_conn.execute.side_effect = [uprn_result, postcode_result] + return mock_engine + + +def _plan_mock() -> MagicMock: + plan = MagicMock() + plan.measures = [] + plan.cost_of_works = 0.0 + return plan + + +# --------------------------------------------------------------------------- +# Fixture: clear module-level cohort cache between tests +# --------------------------------------------------------------------------- + + +@pytest.fixture(autouse=True) +def _clear_cohort_cache() -> None: + import applications.modelling_e2e.handler as h + + h._cohort_cache.clear() + + +# --------------------------------------------------------------------------- +# Trigger body validation +# --------------------------------------------------------------------------- + + +def test_trigger_body_requires_property_ids() -> None: + """property_ids (plural) must be provided and must be a list of ints.""" + # Arrange + body = { + "property_ids": [1, 2], + "portfolio_id": PORTFOLIO_ID, + "scenario_id": SCENARIO_ID, + } + + # Act + result = ModellingE2ETriggerBody.model_validate(body) + + # Assert + assert result.property_ids == [1, 2] + + +def test_trigger_body_rejects_missing_property_ids() -> None: + with pytest.raises(ValidationError): + ModellingE2ETriggerBody.model_validate( + {"portfolio_id": PORTFOLIO_ID, "scenario_id": SCENARIO_ID} + ) + + +# --------------------------------------------------------------------------- +# Lodged EPC path +# --------------------------------------------------------------------------- + + +def test_lodged_epc_path_saves_epc_plan_and_marks_modelled() -> None: + """When get_by_uprn returns an EPC the handler saves it, saves the plan, + and marks the property as modelled — all inside one UoW per property.""" + # Arrange + mock_engine = _engine_mock([PROPERTY_ID], [UPRN], [POSTCODE]) + mock_epc = MagicMock() + mock_plan = _plan_mock() + mock_uow = MagicMock() + + 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 = mock_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") + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.ScenarioPostgresRepository") + ).return_value.get_many.return_value = [MagicMock()] + stack.enter_context( + patch("applications.modelling_e2e.handler.ProductPostgresRepository") + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.Session") + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler.run_modelling", + return_value=mock_plan, + ) + ) + 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) + + # Assert — EPC saved (lodged path), plan saved, property marked modelled + mock_uow.epc.save.assert_called_once_with( + mock_epc, property_id=PROPERTY_ID, portfolio_id=PORTFOLIO_ID + ) + mock_uow.plan.save.assert_called_once() + mock_uow.property.mark_modelled.assert_called_once_with( + PROPERTY_ID, has_recommendations=False + ) + mock_uow.commit.assert_called_once() + + +# --------------------------------------------------------------------------- +# EPC Prediction path +# --------------------------------------------------------------------------- + + +def test_prediction_path_saves_plan_without_epc_save() -> None: + """When get_by_uprn returns None the handler synthesises an EPC via + prediction and saves the plan — but never calls epc.save.""" + # Arrange + mock_engine = _engine_mock([PROPERTY_ID], [UPRN], [POSTCODE]) + mock_plan = _plan_mock() + mock_uow = MagicMock() + + mock_predicted_epc = MagicMock() + # _predict_epc checks for a MAIN building part + from datatypes.epc.domain.epc_property_data import BuildingPartIdentifier + + mock_part = MagicMock() + mock_part.identifier = BuildingPartIdentifier.MAIN + mock_predicted_epc.sap_building_parts = [mock_part] + + mock_comparables = MagicMock() + mock_comparables.members = [MagicMock()] # non-empty cohort + + 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 = None # no lodged 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") + ) + # Prediction infrastructure + from domain.epc_prediction.prediction_target import PredictionTargetAttributes + + stack.enter_context( + patch( + "applications.modelling_e2e.handler.OverrideBackedPredictionAttributesReader" + ) + ).return_value.attributes_for.return_value = PredictionTargetAttributes( + property_type="2" + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.select_comparables") + ).return_value = mock_comparables + stack.enter_context( + patch("applications.modelling_e2e.handler.EpcPrediction") + ).return_value.predict.return_value = mock_predicted_epc + stack.enter_context( + patch( + "applications.modelling_e2e.handler.EpcComparablePropertiesRepository" + ) + ).return_value.candidates_for.return_value = [MagicMock()] + stack.enter_context( + patch("applications.modelling_e2e.handler.ScenarioPostgresRepository") + ).return_value.get_many.return_value = [MagicMock()] + stack.enter_context( + patch("applications.modelling_e2e.handler.ProductPostgresRepository") + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.Session") + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler.run_modelling", + return_value=mock_plan, + ) + ) + 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) + + # Assert — epc.save NOT called (no lodged cert), plan IS saved + mock_uow.epc.save.assert_not_called() + mock_uow.plan.save.assert_called_once() + mock_uow.commit.assert_called_once() + + +# --------------------------------------------------------------------------- +# Prediction gate-out (empty cohort) +# --------------------------------------------------------------------------- + + +def test_empty_cohort_gates_property_out_and_raises() -> None: + """When candidates_for returns an empty list the property cannot be + predicted; the handler records it as an error and raises RuntimeError.""" + # Arrange + mock_engine = _engine_mock([PROPERTY_ID], [UPRN], [POSTCODE]) + + empty_comparables = MagicMock() + empty_comparables.members = [] # empty cohort → gate-out + + 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 = None + 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.overlays_from", return_value=[] + ) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.PropertyOverridesPostgresReader") + ) + from domain.epc_prediction.prediction_target import PredictionTargetAttributes + + stack.enter_context( + patch( + "applications.modelling_e2e.handler.OverrideBackedPredictionAttributesReader" + ) + ).return_value.attributes_for.return_value = PredictionTargetAttributes( + property_type="2" + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.select_comparables") + ).return_value = empty_comparables + stack.enter_context( + patch("applications.modelling_e2e.handler.EpcPrediction") + ) + stack.enter_context( + patch( + "applications.modelling_e2e.handler.EpcComparablePropertiesRepository" + ) + ).return_value.candidates_for.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.ProductPostgresRepository") + ) + stack.enter_context(patch("applications.modelling_e2e.handler.Session")) + MockUoW = stack.enter_context( + patch("applications.modelling_e2e.handler.PostgresUnitOfWork") + ) + + # Act + with pytest.raises(RuntimeError, match=str(PROPERTY_ID)): + _call_handler(_BODY) + + # UoW never entered — the property errored before the write block + MockUoW.return_value.__enter__.assert_not_called() + + +# --------------------------------------------------------------------------- +# Partial batch failure +# --------------------------------------------------------------------------- + + +def test_partial_batch_failure_raises_runtime_error_listing_failed_ids() -> None: + """Two properties: property 1 succeeds, property 2 raises during modelling. + Handler raises RuntimeError naming only the failed property; property 1's + UoW was committed.""" + # Arrange + pid1, pid2 = 111, 222 + mock_engine = _engine_mock([pid1, pid2], [1001, 1002], [POSTCODE, POSTCODE]) + mock_plan = _plan_mock() + mock_uow = MagicMock() + + def _run_modelling_side_effect(*args: Any, **kwargs: Any) -> Any: + # Fail on second call (pid2) + if not hasattr(_run_modelling_side_effect, "_calls"): + _run_modelling_side_effect._calls = 0 # type: ignore[attr-defined] + _run_modelling_side_effect._calls += 1 # type: ignore[attr-defined] + if _run_modelling_side_effect._calls == 2: # type: ignore[attr-defined] + raise ValueError("modelling exploded") + return mock_plan + + 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() # lodged 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") + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.ScenarioPostgresRepository") + ).return_value.get_many.return_value = [MagicMock()] + stack.enter_context( + patch("applications.modelling_e2e.handler.ProductPostgresRepository") + ) + stack.enter_context(patch("applications.modelling_e2e.handler.Session")) + stack.enter_context( + patch( + "applications.modelling_e2e.handler.run_modelling", + side_effect=_run_modelling_side_effect, + ) + ) + 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 + with pytest.raises(RuntimeError, match=str(pid2)): + _call_handler( + { + "property_ids": [pid1, pid2], + "portfolio_id": PORTFOLIO_ID, + "scenario_id": SCENARIO_ID, + "no_solar": True, + "dry_run": False, + } + ) + + # Property 1 succeeded — its UoW was committed + mock_uow.commit.assert_called_once() + + +# --------------------------------------------------------------------------- +# Cohort cache hit +# --------------------------------------------------------------------------- + + +def test_cohort_cache_prevents_duplicate_candidates_for_calls() -> None: + """Two properties sharing a postcode: candidates_for is fetched once and + cached — EpcComparablePropertiesRepository.candidates_for called once.""" + # Arrange + pid1, pid2 = 301, 302 + mock_engine = _engine_mock( + [pid1, pid2], [3001, 3002], [POSTCODE, POSTCODE] + ) + mock_plan = _plan_mock() + mock_uow = MagicMock() + + mock_predicted_epc = MagicMock() + from datatypes.epc.domain.epc_property_data import BuildingPartIdentifier + + mock_part = MagicMock() + mock_part.identifier = BuildingPartIdentifier.MAIN + mock_predicted_epc.sap_building_parts = [mock_part] + + mock_comparables = MagicMock() + mock_comparables.members = [MagicMock()] + + 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 = None # force prediction path + 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.overlays_from", return_value=[] + ) + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.PropertyOverridesPostgresReader") + ) + from domain.epc_prediction.prediction_target import PredictionTargetAttributes + + stack.enter_context( + patch( + "applications.modelling_e2e.handler.OverrideBackedPredictionAttributesReader" + ) + ).return_value.attributes_for.return_value = PredictionTargetAttributes( + property_type="2" + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.select_comparables") + ).return_value = mock_comparables + stack.enter_context( + patch("applications.modelling_e2e.handler.EpcPrediction") + ).return_value.predict.return_value = mock_predicted_epc + MockCandidates = stack.enter_context( + patch( + "applications.modelling_e2e.handler.EpcComparablePropertiesRepository" + ) + ) + MockCandidates.return_value.candidates_for.return_value = [MagicMock()] + stack.enter_context( + patch("applications.modelling_e2e.handler.ScenarioPostgresRepository") + ).return_value.get_many.return_value = [MagicMock()] + stack.enter_context( + patch("applications.modelling_e2e.handler.ProductPostgresRepository") + ) + stack.enter_context(patch("applications.modelling_e2e.handler.Session")) + stack.enter_context( + patch( + "applications.modelling_e2e.handler.run_modelling", + return_value=mock_plan, + ) + ) + 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( + { + "property_ids": [pid1, pid2], + "portfolio_id": PORTFOLIO_ID, + "scenario_id": SCENARIO_ID, + "no_solar": True, + "dry_run": False, + } + ) + + # Assert — cohort fetched exactly once despite two properties + MockCandidates.return_value.candidates_for.assert_called_once_with(POSTCODE) + + +# --------------------------------------------------------------------------- +# Dry-run +# --------------------------------------------------------------------------- + + +def test_dry_run_skips_all_db_writes() -> None: + """dry_run=True: run_modelling executes but PostgresUnitOfWork is never + entered — no DB writes occur for any property in the batch.""" + # Arrange + mock_engine = _engine_mock([PROPERTY_ID], [UPRN], [POSTCODE]) + + 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") + ) + stack.enter_context( + patch("applications.modelling_e2e.handler.ScenarioPostgresRepository") + ).return_value.get_many.return_value = [MagicMock()] + stack.enter_context( + patch("applications.modelling_e2e.handler.ProductPostgresRepository") + ) + stack.enter_context(patch("applications.modelling_e2e.handler.Session")) + stack.enter_context( + patch( + "applications.modelling_e2e.handler.run_modelling", + return_value=_plan_mock(), + ) + ) + MockUoW = stack.enter_context( + patch("applications.modelling_e2e.handler.PostgresUnitOfWork") + ) + + # Act + _call_handler({**_BODY, "dry_run": True}) + + # Assert — UoW never entered + MockUoW.return_value.__enter__.assert_not_called()