junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-06-30 13:10:47 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #1296 from Hestia-Homes/feature/e2e-subtasks

Browse source 
Separate subtask for each property ID when e2e modelling is triggered
This commit is contained in:
Daniel Roth 2026-06-24 12:36:12 +01:00 • committed by GitHub
commit ecd9a5c622
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
6 changed files with 401 additions and 286 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

268
.github/workflows/lambda_smoke_tests.yml vendored
View file

@ -1,144 +1,154 @@
name: Lambda Smoke Tests
# name: Lambda Smoke Tests
on:
pull_request:
branches:
- main
# on:
# pull_request:
# branches:
# - main
jobs:
# ============================================================
# Ara Engine
# ============================================================
ara_engine_smoke_test:
uses: ./.github/workflows/_smoke_test_lambda.yml
with:
dockerfile_path: backend/docker/engine.Dockerfile
build_context: .
service_name: ara-engine
# jobs:
# # ============================================================
# # Ara Engine
# # ============================================================
# ara_engine_smoke_test:
# uses: ./.github/workflows/_smoke_test_lambda.yml
# with:
# dockerfile_path: backend/docker/engine.Dockerfile
# build_context: .
# service_name: ara-engine
# ============================================================
# Address 2 UPRN
# ============================================================
address2uprn_smoke_test:
uses: ./.github/workflows/_smoke_test_lambda.yml
with:
dockerfile_path: backend/address2UPRN/handler/Dockerfile
build_context: .
service_name: address2uprn
# # ============================================================
# # Address 2 UPRN
# # ============================================================
# address2uprn_smoke_test:
# uses: ./.github/workflows/_smoke_test_lambda.yml
# with:
# dockerfile_path: backend/address2UPRN/handler/Dockerfile
# build_context: .
# service_name: address2uprn
# ============================================================
# Postcode Splitter
# ============================================================
postcode_splitter_smoke_test:
uses: ./.github/workflows/_smoke_test_lambda.yml
with:
dockerfile_path: backend/postcode_splitter/handler/Dockerfile
build_context: .
service_name: postcode-splitter
# # ============================================================
# # Postcode Splitter
# # ============================================================
# postcode_splitter_smoke_test:
# uses: ./.github/workflows/_smoke_test_lambda.yml
# with:
# dockerfile_path: backend/postcode_splitter/handler/Dockerfile
# build_context: .
# service_name: postcode-splitter
postcode_splitter_ddd_smoke_test:
uses: ./.github/workflows/_smoke_test_lambda.yml
with:
dockerfile_path: applications/postcode_splitter/Dockerfile
build_context: .
service_name: postcode-splitter-ddd
# postcode_splitter_ddd_smoke_test:
# uses: ./.github/workflows/_smoke_test_lambda.yml
# with:
# dockerfile_path: applications/postcode_splitter/Dockerfile
# build_context: .
# service_name: postcode-splitter-ddd
# ============================================================
# Landlord Description Overrides
# ============================================================
landlord_description_overrides_smoke_test:
uses: ./.github/workflows/_smoke_test_lambda.yml
with:
dockerfile_path: applications/landlord_description_overrides/Dockerfile
build_context: .
service_name: landlord-description-overrides
# # ============================================================
# # Landlord Description Overrides
# # ============================================================
# landlord_description_overrides_smoke_test:
# uses: ./.github/workflows/_smoke_test_lambda.yml
# with:
# dockerfile_path: applications/landlord_description_overrides/Dockerfile
# build_context: .
# service_name: landlord-description-overrides
# ============================================================
# Bulk Address2UPRN Combiner
# ============================================================
bulk_address2uprn_combiner_smoke_test:
uses: ./.github/workflows/_smoke_test_lambda.yml
with:
dockerfile_path: backend/bulk_address2uprn_combiner/handler/Dockerfile
build_context: .
service_name: bulk-address2uprn-combiner
# # ============================================================
# # Bulk Address2UPRN Combiner
# # ============================================================
# bulk_address2uprn_combiner_smoke_test:
# uses: ./.github/workflows/_smoke_test_lambda.yml
# with:
# dockerfile_path: backend/bulk_address2uprn_combiner/handler/Dockerfile
# build_context: .
# service_name: bulk-address2uprn-combiner
# ============================================================
# Bulk Upload Finaliser
# ============================================================
bulk_upload_finaliser_smoke_test:
uses: ./.github/workflows/_smoke_test_lambda.yml
with:
dockerfile_path: applications/bulk_upload_finaliser/Dockerfile
build_context: .
service_name: bulk-upload-finaliser
# # ============================================================
# # Bulk Upload Finaliser
# # ============================================================
# bulk_upload_finaliser_smoke_test:
# uses: ./.github/workflows/_smoke_test_lambda.yml
# with:
# dockerfile_path: applications/bulk_upload_finaliser/Dockerfile
# build_context: .
# service_name: bulk-upload-finaliser
# ============================================================
# Condition ETL
# ============================================================
condition_etl_smoke_test:
uses: ./.github/workflows/_smoke_test_lambda.yml
with:
dockerfile_path: backend/condition/handler/Dockerfile
build_context: .
service_name: condition-etl
# # ============================================================
# # Condition ETL
# # ============================================================
# condition_etl_smoke_test:
# uses: ./.github/workflows/_smoke_test_lambda.yml
# with:
# dockerfile_path: backend/condition/handler/Dockerfile
# build_context: .
# service_name: condition-etl
# ============================================================
# Categorisation
# ============================================================
categorisation_smoke_test:
uses: ./.github/workflows/_smoke_test_lambda.yml
with:
dockerfile_path: backend/categorisation/handler/Dockerfile
build_context: .
service_name: categorisation
# # ============================================================
# # Categorisation
# # ============================================================
# categorisation_smoke_test:
# uses: ./.github/workflows/_smoke_test_lambda.yml
# with:
# dockerfile_path: backend/categorisation/handler/Dockerfile
# build_context: .
# service_name: categorisation
# ============================================================
# Ordnance Survey
# ============================================================
ordnance_survey_smoke_test:
uses: ./.github/workflows/_smoke_test_lambda.yml
with:
dockerfile_path: backend/ordnanceSurvey/handler/Dockerfile
build_context: .
service_name: ordnance-survey
# # ============================================================
# # Ordnance Survey
# # ============================================================
# ordnance_survey_smoke_test:
# uses: ./.github/workflows/_smoke_test_lambda.yml
# with:
# dockerfile_path: backend/ordnanceSurvey/handler/Dockerfile
# build_context: .
# service_name: ordnance-survey
# ============================================================
# Pas Hub Fetcher
# ============================================================
pashub_smoke_test:
uses: ./.github/workflows/_smoke_test_lambda.yml
with:
dockerfile_path: backend/pashub_fetcher/handler/Dockerfile
build_context: .
service_name: pashub
# # ============================================================
# # Pas Hub Fetcher
# # ============================================================
# pashub_smoke_test:
# uses: ./.github/workflows/_smoke_test_lambda.yml
# with:
# dockerfile_path: backend/pashub_fetcher/handler/Dockerfile
# build_context: .
# service_name: pashub
# ============================================================
# MagicPlan
# ============================================================
magic_plan_smoke_test:
uses: ./.github/workflows/_smoke_test_lambda.yml
with:
dockerfile_path: applications/magic_plan/handler/Dockerfile
build_context: .
service_name: magic-plan
# # ============================================================
# # MagicPlan
# # ============================================================
# magic_plan_smoke_test:
# uses: ./.github/workflows/_smoke_test_lambda.yml
# with:
# dockerfile_path: applications/magic_plan/handler/Dockerfile
# build_context: .
# service_name: magic-plan
# ============================================================
# Audit Generator
# ============================================================
audit_generator_smoke_test:
uses: ./.github/workflows/_smoke_test_lambda.yml
with:
dockerfile_path: applications/audit_generator/handler/Dockerfile
build_context: .
service_name: audit-generator
# # ============================================================
# # Audit Generator
# # ============================================================
# audit_generator_smoke_test:
# uses: ./.github/workflows/_smoke_test_lambda.yml
# with:
# dockerfile_path: applications/audit_generator/handler/Dockerfile
# build_context: .
# service_name: audit-generator
# ============================================================
# HubSpot Scraper
# ============================================================
hubspot_scraper_smoke_test:
uses: ./.github/workflows/_smoke_test_lambda.yml
with:
dockerfile_path: etl/hubspot/scripts/scraper/handler/Dockerfile
build_context: .
service_name: hubspot-scraper
# # ============================================================
# # HubSpot Scraper
# # ============================================================
# hubspot_scraper_smoke_test:
# uses: ./.github/workflows/_smoke_test_lambda.yml
# with:
# dockerfile_path: etl/hubspot/scripts/scraper/handler/Dockerfile
# build_context: .
# service_name: hubspot-scraper
# # ============================================================
# # Modelling E2E
# # ============================================================
# hubspot_scraper_smoke_test:
# uses: ./.github/workflows/_smoke_test_lambda.yml
# with:
# dockerfile_path: applications/modelling_e2e/Dockerfile
# build_context: .
# service_name: modelling_e2e

101
applications/modelling_e2e/handler.py
View file

@ -25,7 +25,6 @@ from __future__ import annotations
import dataclasses
import io
import json
import os
from collections.abc import Callable
from typing import Any, Optional, cast
@ -56,6 +55,7 @@ from domain.property_baseline.calculator_rebaseliner import CalculatorRebaseline
from domain.sap10_calculator.calculator import Sap10Calculator
from domain.tasks.tasks import Source
from harness.console import run_modelling
from orchestration.task_orchestrator import TaskOrchestrator
from orchestration.property_baseline_orchestrator import (
PropertyBaselineOrchestrator,
)
@ -97,6 +97,7 @@ from repositories.scenario.scenario_postgres_repository import (
)
from repositories.solar.solar_postgres_repository import SolarPostgresRepository
from utilities.aws_lambda.task_handler import task_handler
from uuid import UUID
from utilities.logger import setup_logger
_engine: Optional[Engine] = None
@ -212,8 +213,8 @@ def _predict_epc(
return predicted
@task_handler(task_source="modelling_e2e", source=Source.PROPERTY)
def handler(body: dict[str, Any], context: Any) -> Optional[dict[str, Any]]:
@task_handler(task_source="modelling_e2e", source=Source.PROPERTY, pass_task_orchestrator=True)
def handler(body: dict[str, Any], context: Any, orchestrator: TaskOrchestrator, task_id: UUID) -> None:
trigger = ModellingE2ETriggerBody.model_validate(body)
property_ids = trigger.property_ids
portfolio_id = trigger.portfolio_id
@ -292,13 +293,15 @@ def handler(body: dict[str, Any], context: Any) -> Optional[dict[str, Any]]:
products = catalogue_with_off_catalogue_overrides(read_session)
solar_reader = SolarPostgresRepository(read_session)
failures: list[dict[str, Any]] = []
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}")
child = orchestrator.create_child_subtask(
task_id, inputs={"property_id": property_id}
)
def _work(pid: int = property_id) -> None:
uprn = uprns[pid]
postcode = postcodes.get(pid, "")
logger.info(f"property={pid} uprn={uprn} postcode={postcode!r}")
spatial = _spatial_for(geospatial, uprn)
restrictions = (
@ -311,11 +314,11 @@ def handler(body: dict[str, Any], context: Any) -> Optional[dict[str, Any]]:
)
epc: Optional[EpcPropertyData] = epc_client.get_by_uprn(uprn)
overrides = overlays_from(overrides_reader.overrides_for(property_id))
overrides = overlays_from(overrides_reader.overrides_for(pid))
predicted_epc: Optional[EpcPropertyData] = None
if epc is not None:
logger.info(f"property={property_id} lodged EPC found")
logger.info(f"property={pid} lodged EPC found")
effective_epc = Property(
identity=PropertyIdentity(
portfolio_id=portfolio_id,
@ -328,10 +331,10 @@ def handler(body: dict[str, Any], context: Any) -> Optional[dict[str, Any]]:
).effective_epc
else:
logger.info(
f"property={property_id} no lodged EPC — attempting prediction"
f"property={pid} no lodged EPC — attempting prediction"
)
predicted_epc = _predict_epc(
property_id=property_id,
property_id=pid,
uprn=uprn,
postcode=postcode,
portfolio_id=portfolio_id,
@ -359,9 +362,6 @@ def handler(body: dict[str, Any], context: Any) -> Optional[dict[str, Any]]:
landlord_overrides=overrides,
).effective_epc
# Read-before-fetch: the Google Solar call is paid, so skip it
# when this UPRN's insights are already persisted. Only a cache
# miss hits Google — re-runs cost nothing for solar.
solar_insights: Optional[dict[str, Any]]
solar_was_fetched = False
if no_solar:
@ -372,9 +372,6 @@ def handler(body: dict[str, Any], context: Any) -> Optional[dict[str, Any]]:
solar_insights = _solar_insights_for(solar_client, spatial)
solar_was_fetched = solar_insights is not None
# All Measure Types are considered: the off-catalogue overlay
# (catalogue_with_off_catalogue_overrides) prices the measures the
# live material catalogue cannot supply, so none need excluding.
plan = run_modelling(
effective_epc,
planning_restrictions=restrictions,
@ -385,7 +382,7 @@ def handler(body: dict[str, Any], context: Any) -> Optional[dict[str, Any]]:
print_table=False,
)
logger.info(
f"property={property_id} modelling complete "
f"property={pid} modelling complete "
f"measures={len(plan.measures)}"
)
@ -394,15 +391,15 @@ def handler(body: dict[str, Any], context: Any) -> Optional[dict[str, Any]]:
", ".join(m.measure_type for m in plan.measures) or "none"
)
logger.info(
f"[dry_run] property={property_id} "
f"[dry_run] property={pid} "
f"measures=[{measure_types}] — skipping DB write"
)
continue
return
with PostgresUnitOfWork(lambda: Session(engine)) as uow:
if epc is not None:
uow.epc.save(
epc, property_id=property_id, portfolio_id=portfolio_id
epc, property_id=pid, portfolio_id=portfolio_id
)
elif predicted_epc is not None:
# Persist the synthesised EPC in the predicted slot (ADR-0031),
@ -410,7 +407,7 @@ def handler(body: dict[str, Any], context: Any) -> Optional[dict[str, Any]]:
# the picture the Plan was modelled from.
uow.epc.save(
predicted_epc,
property_id=property_id,
property_id=pid,
portfolio_id=portfolio_id,
source="predicted",
)
@ -430,38 +427,24 @@ def handler(body: dict[str, Any], context: Any) -> Optional[dict[str, Any]]:
)
uow.plan.save(
plan,
property_id=property_id,
property_id=pid,
scenario_id=scenario_id,
portfolio_id=portfolio_id,
is_default=scenario.is_default,
)
uow.property.mark_modelled(
property_id, has_recommendations=bool(plan.measures)
pid, has_recommendations=bool(plan.measures)
)
uow.commit()
logger.info(f"property={property_id} plan saved")
logger.info(f"property={pid} plan saved")
# Baseline Performance is re-established from the persisted EPC
# (lodged or predicted), so it runs after the Plan UoW commits. By
# here the property always has a persisted EPC — a property that
# could be neither fetched nor predicted raised earlier.
baseline_orchestrator.run([property_id])
logger.info(f"property={property_id} baseline saved")
baseline_orchestrator.run([pid])
logger.info(f"property={pid} baseline saved")
except Exception as error: # noqa: BLE001
logger.error(
f"property={property_id} uprn={uprns.get(property_id)}: "
f"{type(error).__name__}: {error}",
exc_info=True,
)
failures.append(
{
"property_id": property_id,
"uprn": uprns.get(property_id),
"error_type": type(error).__name__,
"error": str(error),
}
)
try:
orchestrator.run_subtask(child.id, work=_work)
except Exception: # noqa: BLE001
pass
# Cohort certs the mapper could not consume were skipped (not aborted on)
# so prediction could proceed; surface them — with cert numbers — in the
@ -476,29 +459,5 @@ def handler(body: dict[str, Any], context: Any) -> Optional[dict[str, Any]]:
f"{[s['certificate_number'] for s in skipped_certs]}"
)
# A property that errored AND a cohort cert the mapper could not consume
# are both surfaced as failures, so the subtask is marked failed and
# shows up for debugging. The whole batch has already run by this point —
# every property that could be modelled was written to DB above — so
# failing here flags the run without discarding the work done so far.
if failures or skipped_certs:
parts: list[str] = []
if failures:
failed_ids = [f["property_id"] for f in failures]
# Persisted verbatim into the subtask's outputs.error (via
# SubTask.fail): include each property's error type + message,
# not just the IDs, so failed runs are diagnosable without
# cross-referencing CloudWatch.
parts.append(
f"failed property_ids: {failed_ids}; "
f"details: {json.dumps(failures)}"
)
if skipped_certs:
parts.append(
f"skipped_unmappable_cohort_certs: {json.dumps(skipped_certs)}"
)
raise RuntimeError("; ".join(parts))
return None
finally:
read_session.close()

4
applications/modelling_e2e/local_handler/invoke_local_lambda.py
View file

@ -12,11 +12,11 @@ payload = {
{
"body": json.dumps(
{
"property_ids": [722473],
"property_ids": [727220, 727229],
"portfolio_id": 796,
"scenario_id": 1268,
"no_solar": False,
"dry_run": False,
"dry_run": True,
}
)
}

269
tests/applications/modelling_e2e/test_handler.py
View file

@ -8,8 +8,9 @@ is needed. One test per distinct behaviour path.
from __future__ import annotations
from contextlib import ExitStack
from typing import Any, Iterator
from typing import Any, Iterator, Optional
from unittest.mock import MagicMock, call, patch
from uuid import UUID, uuid4
import pytest
from pydantic import ValidationError
@ -39,10 +40,22 @@ _BODY = {
}
def _call_handler(body: dict[str, Any]) -> Any:
def _mock_orchestrator() -> MagicMock:
mock = MagicMock()
mock.run_subtask.side_effect = lambda subtask_id, work, **kwargs: work()
child = MagicMock()
child.id = uuid4()
mock.create_child_subtask.return_value = child
return mock
def _call_handler(
body: dict[str, Any], orchestrator: Optional[MagicMock] = None
) -> Any:
from applications.modelling_e2e.handler import handler
return handler.__wrapped__(body, None) # type: ignore[attr-defined]
orch = orchestrator if orchestrator is not None else _mock_orchestrator()
return handler.__wrapped__(body, None, orch, uuid4()) # type: ignore[attr-defined]
def _engine_mock(
@ -122,6 +135,72 @@ def test_trigger_body_rejects_missing_property_ids() -> None:
)
# ---------------------------------------------------------------------------
# Child SubTask creation
# ---------------------------------------------------------------------------
def test_handler_creates_one_child_subtask_per_property_id() -> None:
"""Handler creates exactly N child SubTasks, one per property in the batch,
each recording its property_id in inputs."""
# Arrange
pid1, pid2, pid3 = 111, 222, 333
mock_engine = _engine_mock(
[pid1, pid2, pid3], [1001, 1002, 1003], [POSTCODE, POSTCODE, POSTCODE]
)
mock_orch = _mock_orchestrator()
task_id = uuid4()
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"))
mock_uow = MagicMock()
MockUoW.return_value.__enter__.return_value = mock_uow
MockUoW.return_value.__exit__.return_value = False
# Act
from applications.modelling_e2e.handler import handler
handler.__wrapped__( # type: ignore[attr-defined]
{"property_ids": [pid1, pid2, pid3], "portfolio_id": PORTFOLIO_ID,
"scenario_id": SCENARIO_ID, "no_solar": True, "dry_run": False},
None, mock_orch, task_id,
)
# Assert — one child SubTask per property, inputs record the property_id
assert mock_orch.create_child_subtask.call_count == 3
calls = mock_orch.create_child_subtask.call_args_list
recorded_ids = [c.kwargs["inputs"]["property_id"] for c in calls]
assert recorded_ids == [pid1, pid2, pid3]
# All three calls used the same task_id
assert all(c.args[0] == task_id for c in calls)
# ---------------------------------------------------------------------------
# Lodged EPC path
# ---------------------------------------------------------------------------
@ -214,12 +293,9 @@ def test_lodged_epc_path_saves_epc_plan_and_marks_modelled(
_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."""
def test_skipped_cohort_certs_do_not_prevent_plan_being_saved() -> None:
"""Cohort certs the mapper can't consume are skipped so prediction can
proceed — the property's plan is committed before the handler returns."""
from repositories.comparable_properties.epc_comparable_properties_repository import (
SkippedCohortCert,
)
@ -293,20 +369,81 @@ def test_skipped_cohort_certs_fail_the_subtask_but_the_plan_is_still_saved() ->
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)
# Act — no raise even with a skipped cert
_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
# Assert — plan committed despite the skipped cert
mock_uow.plan.save.assert_called_once()
mock_uow.commit.assert_called_once()
def test_skipped_cohort_certs_are_logged_and_handler_does_not_raise() -> None:
"""Unmappable cohort certs are logged but do not raise — the coordinator
SubTask completes normally and the property plan is committed."""
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)
)
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
mock_logger = stack.enter_context(
patch("applications.modelling_e2e.handler.logger")
)
# Act — must not raise even though cohort certs were skipped
_call_handler(_BODY)
# Assert — plan committed; skipped cert number surfaced in a log call
mock_uow.plan.save.assert_called_once()
mock_uow.commit.assert_called_once()
logged_messages = " ".join(
str(c.args) + str(c.kwargs) for c in mock_logger.info.call_args_list
)
assert "8257-7539-1649-0633-4992" in logged_messages
# ---------------------------------------------------------------------------
# EPC Prediction path
# ---------------------------------------------------------------------------
@ -434,9 +571,10 @@ def test_prediction_path_saves_predicted_epc_plan_and_baseline(
# ---------------------------------------------------------------------------
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."""
def test_empty_cohort_gates_property_out_without_saving() -> None:
"""When candidates_for returns an empty list the property cannot be predicted;
the failure is recorded in the child SubTask and the handler returns without
raising — no plan is saved."""
# Arrange
mock_engine = _engine_mock([PROPERTY_ID], [UPRN], [POSTCODE])
@ -506,9 +644,8 @@ def test_empty_cohort_gates_property_out_and_raises() -> None:
patch("applications.modelling_e2e.handler.PostgresUnitOfWork")
)
# Act
with pytest.raises(RuntimeError, match=str(PROPERTY_ID)):
_call_handler(_BODY)
# Act — no raise; child SubTask is failed, handler continues
_call_handler(_BODY)
# UoW never entered — the property errored before the write block
MockUoW.return_value.__enter__.assert_not_called()
@ -639,73 +776,53 @@ def test_empty_own_postcode_broadens_to_nearby_and_predicts() -> None:
# ---------------------------------------------------------------------------
# Partial batch failure
# Partial batch failure — per-property isolation
# ---------------------------------------------------------------------------
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."""
def test_per_property_failure_fails_child_subtask_and_siblings_continue() -> None:
"""Two properties: property 1 succeeds, property 2 fails during modelling.
The handler does not raise; property 1's UoW was committed; run_subtask was
called for both properties."""
# Arrange
pid1, pid2 = 111, 222
mock_engine = _engine_mock([pid1, pid2], [1001, 1002], [POSTCODE, POSTCODE])
mock_plan = _plan_mock()
mock_uow = MagicMock()
call_count = {"n": 0}
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]
call_count["n"] += 1
if call_count["n"] == 2:
raise ValueError("modelling exploded")
return mock_plan
mock_orch = _mock_orchestrator()
with ExitStack() as stack:
stack.enter_context(patch("applications.modelling_e2e.handler.os.environ", _ENV))
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,
)
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
).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.GeospatialS3Repository")
patch("applications.modelling_e2e.handler._spatial_for", return_value=None)
)
stack.enter_context(
patch("applications.modelling_e2e.handler.GoogleSolarApiClient")
patch("applications.modelling_e2e.handler._solar_insights_for", return_value=None)
)
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")
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.catalogue_with_off_catalogue_overrides"))
stack.enter_context(patch("applications.modelling_e2e.handler.Session"))
stack.enter_context(
patch(
@ -713,25 +830,19 @@ def test_partial_batch_failure_raises_runtime_error_listing_failed_ids() -> None
side_effect=_run_modelling_side_effect,
)
)
MockUoW = stack.enter_context(
patch("applications.modelling_e2e.handler.PostgresUnitOfWork")
)
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,
}
)
# Act — must not raise even though pid2 fails
_call_handler(
{"property_ids": [pid1, pid2], "portfolio_id": PORTFOLIO_ID,
"scenario_id": SCENARIO_ID, "no_solar": True, "dry_run": False},
orchestrator=mock_orch,
)
# Property 1 succeeded — its UoW was committed
# run_subtask called for both properties; pid1 committed
assert mock_orch.run_subtask.call_count == 2
mock_uow.commit.assert_called_once()

27
tests/utilities/aws_lambda/test_task_handler.py
View file

@ -77,6 +77,33 @@ def test_task_handler_records_cloudwatch_url_on_subtask(
assert "$255B$2524LATEST$255D" in saved_url
def test_task_handler_passes_orchestrator_and_task_id_when_flag_is_true(
harness: Harness,
) -> None:
# Arrange
received: list[tuple[TaskOrchestrator, UUID]] = []
@task_handler(
task_source="modelling_e2e",
source=Source.PROPERTY,
orchestrator_cm=harness.factory,
pass_task_orchestrator=True,
)
def _handler(
body: dict[str, Any], context: Any, orchestrator: TaskOrchestrator, task_id: UUID
) -> None:
received.append((orchestrator, task_id))
# Act
result = _handler(_direct_event("prop-1"), context=None)
# Assert
assert len(received) == 1
recv_orchestrator, recv_task_id = received[0]
assert recv_orchestrator is harness.orchestrator
assert recv_task_id == UUID(result[0]["task_id"])
def test_task_handler_leaves_cloudwatch_url_unset_outside_lambda(
harness: Harness, monkeypatch: pytest.MonkeyPatch
) -> None:

18
utilities/aws_lambda/task_handler.py
View file

@ -25,6 +25,7 @@ def task_handler(
task_source: str,
source: Source,
orchestrator_cm: Optional[OrchestratorCM] = None,
pass_task_orchestrator: bool = False,
) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
"""Run the wrapped function as the body of a freshly-created Task + SubTask.
@ -65,11 +66,18 @@ def task_handler(
)
try:
orchestrator.run_subtask(
subtask.id,
work=lambda body=body: func(body, context),
cloud_logs_url=cloud_logs_url,
)
if pass_task_orchestrator:
orchestrator.run_subtask(
subtask.id,
work=lambda: func(body, context, orchestrator, task.id),
cloud_logs_url=cloud_logs_url,
)
else:
orchestrator.run_subtask(
subtask.id,
work=lambda: func(body, context),
cloud_logs_url=cloud_logs_url,
)
except Exception:
logger.exception(
"subtask failed (task_source=%s source_id=%s)",