junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-07-12 13:29:04 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions
Model/tests/applications/modelling_e2e/test_handler.py
Khalim Conn-Kowlessar 0c70280dea guard(modelling_e2e): quarantine predicted Properties the calculator mis-scores 
TEMPORARY guard (remove once the SAP calculator's oil-heating under-score is
fixed): a predicted oil-boiler picture scores SAP 13/G against its own
synthesised recorded SAP of 50/E, so the optimiser overshoots goal C all the
way to band A and publishes nonsense.

A predicted EpcPropertyData carries its recorded SAP (energy_rating_current).
When the calculator baseline diverges from it by more than ~one band (20 SAP
points), withhold the Plan: raise inside the per-property loop so the existing
failure isolation drops just that property into `failures` and fails the
subtask, while every other property still models and persists. Lodged
Properties are untouched — they have a real recorded cert and the Rebaseliner
already owns this check.

Verified end-to-end against property 713406 (UPRN 100061849247): baseline 13.2
vs recorded 50 -> quarantined, no Plan written.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-06-24 09:07:24 +00:00

950 lines
35 KiB
Python
Raw Blame History

"""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, Iterator
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
# A plausible baseline so the predicted-baseline guard stays silent (it
# compares this against the picture's recorded SAP).
plan.baseline.sap_continuous = 50.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()
h._nearby_cohort_cache.clear()
@pytest.fixture(autouse=True)
def _baseline_orchestrator() -> Iterator[MagicMock]:
"""Patch the Baseline orchestrator for every test — construction stays cheap
and ``.run`` is a no-op mock the baseline tests assert against."""
with patch(
"applications.modelling_e2e.handler.PropertyBaselineOrchestrator"
) as orchestrator:
yield orchestrator
# ---------------------------------------------------------------------------
# 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(
_baseline_orchestrator: MagicMock,
) -> None:
"""When get_by_uprn returns an EPC the handler saves it, saves the plan,
marks the property as modelled — all inside one UoW per property — and
re-establishes its Baseline Performance after the plan commits."""
# 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.catalogue_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,
)
)
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()
# Baseline Performance is re-established for the lodged property.
_baseline_orchestrator.return_value.run.assert_called_once_with([PROPERTY_ID])
def test_skipped_cohort_certs_fail_the_subtask_but_the_plan_is_still_saved() -> None:
"""Cohort certs the mapper can't consume are skipped (so prediction is not
aborted), then surfaced as a failure — the subtask is marked failed (the
cert numbers land in outputs.error via the raised RuntimeError) so the
mapper gaps get debugged. The batch still ran to completion first, so the
property's plan was committed before the handler raised."""
from repositories.comparable_properties.epc_comparable_properties_repository import (
SkippedCohortCert,
)
mock_engine = _engine_mock([PROPERTY_ID], [UPRN], [POSTCODE])
mock_plan = _plan_mock()
mock_uow = MagicMock()
skipped = [
SkippedCohortCert(
certificate_number="8257-7539-1649-0633-4992",
error="ValueError: RdSapSchema17_1: missing required field 'window'",
)
]
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.catalogue_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 repo accumulated a skipped (unmappable) cohort cert during the run.
stack.enter_context(
patch("applications.modelling_e2e.handler.EpcComparablePropertiesRepository")
).return_value.skipped = skipped
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 — the skipped cert fails the subtask, but only after the batch ran.
with pytest.raises(RuntimeError) as excinfo:
_call_handler(_BODY)
# Assert — the cert number + error reach outputs.error (the raised message),
# and the property's plan was still committed before the handler raised.
message = str(excinfo.value)
assert "skipped_unmappable_cohort_certs" in message
assert "8257-7539-1649-0633-4992" in message
assert "RdSapSchema17_1: missing required field 'window'" in message
mock_uow.plan.save.assert_called_once()
mock_uow.commit.assert_called_once()
# ---------------------------------------------------------------------------
# EPC Prediction path
# ---------------------------------------------------------------------------
def test_prediction_path_saves_predicted_epc_plan_and_baseline(
_baseline_orchestrator: MagicMock,
) -> None:
"""When get_by_uprn returns None the handler synthesises an EPC via
prediction, persists it in the predicted slot (source='predicted'), saves the
plan, and re-establishes the Baseline from the predicted picture."""
# 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_predicted_epc.energy_rating_current = 50 # matches plan baseline -> guard silent
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.catalogue_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,
)
)
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 — predicted EPC persisted in the predicted slot, plan saved, baseline run
mock_uow.epc.save.assert_called_once_with(
mock_predicted_epc,
property_id=PROPERTY_ID,
portfolio_id=PORTFOLIO_ID,
source="predicted",
)
mock_uow.plan.save.assert_called_once()
mock_uow.commit.assert_called_once()
_baseline_orchestrator.return_value.run.assert_called_once_with([PROPERTY_ID])
# ---------------------------------------------------------------------------
# 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.catalogue_with_off_catalogue_overrides")
)
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()
# ---------------------------------------------------------------------------
# Broadened cohort — sparse own postcode falls back to nearby postcodes
# ---------------------------------------------------------------------------
def test_empty_own_postcode_broadens_to_nearby_and_predicts() -> None:
"""When the property's own postcode holds no same-type comparables, the
handler broadens to the nearby-postcode cohort (candidates_near) and, finding
comparables there, synthesises the EPC and saves the plan."""
# Arrange
mock_engine = _engine_mock([PROPERTY_ID], [UPRN], [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_predicted_epc.energy_rating_current = 50 # matches plan baseline -> guard silent
# First select_comparables (own postcode) is empty → broaden; the second
# (nearby cohort) finds comparables.
empty_comparables = MagicMock()
empty_comparables.members = []
found_comparables = MagicMock()
found_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 # 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")
)
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"
)
MockRepo = stack.enter_context(
patch(
"applications.modelling_e2e.handler.EpcComparablePropertiesRepository"
)
)
MockRepo.return_value.candidates_for.return_value = []
MockRepo.return_value.candidates_near.return_value = [MagicMock()]
stack.enter_context(
patch(
"applications.modelling_e2e.handler.select_comparables",
side_effect=[empty_comparables, found_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.ScenarioPostgresRepository")
).return_value.get_many.return_value = [MagicMock()]
stack.enter_context(
patch(
"applications.modelling_e2e.handler.catalogue_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,
)
)
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 — broadening fired, and the broadened cohort produced a saved plan
# with its predicted EPC persisted in the predicted slot.
MockRepo.return_value.candidates_near.assert_called_once()
mock_uow.epc.save.assert_called_once_with(
mock_predicted_epc,
property_id=PROPERTY_ID,
portfolio_id=PORTFOLIO_ID,
source="predicted",
)
mock_uow.plan.save.assert_called_once()
mock_uow.commit.assert_called_once()
# ---------------------------------------------------------------------------
# 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.catalogue_with_off_catalogue_overrides")
)
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_predicted_epc.energy_rating_current = 50 # matches plan baseline -> guard silent
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.catalogue_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,
)
)
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.catalogue_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=_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()
def test_predicted_baseline_within_band_is_plausible() -> None:
# A predicted picture whose calculator baseline tracks its recorded SAP
# (here 50 vs 47) is trusted — the guard does not fire.
from applications.modelling_e2e.handler import _predicted_baseline_is_implausible
assert _predicted_baseline_is_implausible(47.0, 50) is False
def test_predicted_baseline_beyond_band_is_implausible() -> None:
# The 713406 case: calculator scores the oil-boiler picture at 13 while the
# synthesised cert records 50 — a >20-point contradiction the guard rejects.
from applications.modelling_e2e.handler import _predicted_baseline_is_implausible
assert _predicted_baseline_is_implausible(13.2, 50) is True
def test_predicted_baseline_without_a_recorded_sap_is_not_judged() -> None:
# No recorded SAP means no reference to contradict, so the guard stays silent.
from applications.modelling_e2e.handler import _predicted_baseline_is_implausible
assert _predicted_baseline_is_implausible(13.2, None) is False
Reference in a new issue View git blame Copy permalink