From 3a66819a86c11e8d1f8ac6a5cef7751ddd737939 Mon Sep 17 00:00:00 2001 From: Daniel Roth Date: Wed, 24 Jun 2026 10:47:05 +0000 Subject: [PATCH 01/10] =?UTF-8?q?task=5Fhandler=20passes=20orchestrator=20?= =?UTF-8?q?and=20task=5Fid=20to=20wrapped=20function=20when=20flag=20is=20?= =?UTF-8?q?true=20=F0=9F=9F=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../utilities/aws_lambda/test_task_handler.py | 27 +++++++++++++++++++ utilities/aws_lambda/task_handler.py | 1 + 2 files changed, 28 insertions(+) diff --git a/tests/utilities/aws_lambda/test_task_handler.py b/tests/utilities/aws_lambda/test_task_handler.py index fae35de2..418669bc 100644 --- a/tests/utilities/aws_lambda/test_task_handler.py +++ b/tests/utilities/aws_lambda/test_task_handler.py @@ -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: diff --git a/utilities/aws_lambda/task_handler.py b/utilities/aws_lambda/task_handler.py index 43699aee..0ed680c9 100644 --- a/utilities/aws_lambda/task_handler.py +++ b/utilities/aws_lambda/task_handler.py @@ -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. From 35a794563ae3c0c3afb07b1f9731bcfcd1c6c131 Mon Sep 17 00:00:00 2001 From: Daniel Roth Date: Wed, 24 Jun 2026 10:48:40 +0000 Subject: [PATCH 02/10] =?UTF-8?q?task=5Fhandler=20passes=20orchestrator=20?= =?UTF-8?q?and=20task=5Fid=20to=20wrapped=20function=20when=20flag=20is=20?= =?UTF-8?q?true=20=F0=9F=9F=A9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Sonnet 4.6 --- utilities/aws_lambda/task_handler.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/utilities/aws_lambda/task_handler.py b/utilities/aws_lambda/task_handler.py index 0ed680c9..ada8562e 100644 --- a/utilities/aws_lambda/task_handler.py +++ b/utilities/aws_lambda/task_handler.py @@ -66,9 +66,15 @@ def task_handler( ) try: + if pass_task_orchestrator: + work = lambda body=body, t=task: func( + body, context, orchestrator, t.id + ) + else: + work = lambda body=body: func(body, context) orchestrator.run_subtask( subtask.id, - work=lambda body=body: func(body, context), + work=work, cloud_logs_url=cloud_logs_url, ) except Exception: From 2d2abc016b877d6d25ca4238673a4ddc49fefaf6 Mon Sep 17 00:00:00 2001 From: Daniel Roth Date: Wed, 24 Jun 2026 10:56:13 +0000 Subject: [PATCH 03/10] =?UTF-8?q?handler=20creates=20one=20child=20SubTask?= =?UTF-8?q?=20per=20property=20ID=20in=20the=20batch=20=F0=9F=9F=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Sonnet 4.6 --- applications/modelling_e2e/handler.py | 6 +- .../modelling_e2e/test_handler.py | 85 ++++++++++++++++++- 2 files changed, 86 insertions(+), 5 deletions(-) diff --git a/applications/modelling_e2e/handler.py b/applications/modelling_e2e/handler.py index c5aa189b..829044a7 100644 --- a/applications/modelling_e2e/handler.py +++ b/applications/modelling_e2e/handler.py @@ -56,6 +56,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 +98,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 +214,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 diff --git a/tests/applications/modelling_e2e/test_handler.py b/tests/applications/modelling_e2e/test_handler.py index baa20103..2ac4520d 100644 --- a/tests/applications/modelling_e2e/test_handler.py +++ b/tests/applications/modelling_e2e/test_handler.py @@ -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 # --------------------------------------------------------------------------- From 91c2d8a8fd40bc51bc40f7568ce0080f37a12b92 Mon Sep 17 00:00:00 2001 From: Daniel Roth Date: Wed, 24 Jun 2026 10:58:47 +0000 Subject: [PATCH 04/10] =?UTF-8?q?handler=20creates=20one=20child=20SubTask?= =?UTF-8?q?=20per=20property=20ID=20in=20the=20batch=20=F0=9F=9F=A9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Sonnet 4.6 --- applications/modelling_e2e/handler.py | 52 +++++++++++++-------------- 1 file changed, 24 insertions(+), 28 deletions(-) diff --git a/applications/modelling_e2e/handler.py b/applications/modelling_e2e/handler.py index 829044a7..c06d4b4d 100644 --- a/applications/modelling_e2e/handler.py +++ b/applications/modelling_e2e/handler.py @@ -297,10 +297,14 @@ def handler(body: dict[str, Any], context: Any, orchestrator: TaskOrchestrator, 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 = ( @@ -313,11 +317,11 @@ def handler(body: dict[str, Any], context: Any, orchestrator: TaskOrchestrator, ) 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, @@ -330,10 +334,10 @@ def handler(body: dict[str, Any], context: Any, orchestrator: TaskOrchestrator, ).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, @@ -361,9 +365,6 @@ def handler(body: dict[str, Any], context: Any, orchestrator: TaskOrchestrator, 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: @@ -374,9 +375,6 @@ def handler(body: dict[str, Any], context: Any, orchestrator: TaskOrchestrator, 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, @@ -387,7 +385,7 @@ def handler(body: dict[str, Any], context: Any, orchestrator: TaskOrchestrator, print_table=False, ) logger.info( - f"property={property_id} modelling complete " + f"property={pid} modelling complete " f"measures={len(plan.measures)}" ) @@ -396,15 +394,15 @@ def handler(body: dict[str, Any], context: Any, orchestrator: TaskOrchestrator, ", ".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), @@ -412,7 +410,7 @@ def handler(body: dict[str, Any], context: Any, orchestrator: TaskOrchestrator, # 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", ) @@ -432,24 +430,22 @@ def handler(body: dict[str, Any], context: Any, orchestrator: TaskOrchestrator, ) 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") + try: + orchestrator.run_subtask(child.id, work=_work) except Exception as error: # noqa: BLE001 logger.error( f"property={property_id} uprn={uprns.get(property_id)}: " From 39f028f03a6b962682d2c298dd59808055a3f264 Mon Sep 17 00:00:00 2001 From: Daniel Roth Date: Wed, 24 Jun 2026 11:04:29 +0000 Subject: [PATCH 05/10] =?UTF-8?q?per-property=20failure=20fails=20child=20?= =?UTF-8?q?SubTask=20without=20raising=20from=20handler=20=F0=9F=9F=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Sonnet 4.6 --- .../modelling_e2e/test_handler.py | 71 +++++++++++++++++++ 1 file changed, 71 insertions(+) diff --git a/tests/applications/modelling_e2e/test_handler.py b/tests/applications/modelling_e2e/test_handler.py index 2ac4520d..a07f7346 100644 --- a/tests/applications/modelling_e2e/test_handler.py +++ b/tests/applications/modelling_e2e/test_handler.py @@ -814,6 +814,77 @@ def test_partial_batch_failure_raises_runtime_error_listing_failed_ids() -> None mock_uow.commit.assert_called_once() +# --------------------------------------------------------------------------- +# Partial batch failure — per-property isolation +# --------------------------------------------------------------------------- + + +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: + 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._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", + 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 — 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, + ) + + # run_subtask called for both properties; pid1 committed + assert mock_orch.run_subtask.call_count == 2 + mock_uow.commit.assert_called_once() + + # --------------------------------------------------------------------------- # Cohort cache hit # --------------------------------------------------------------------------- From c2a84d3ee8a0a6d083286a5b93960e498e1241cc Mon Sep 17 00:00:00 2001 From: Daniel Roth Date: Wed, 24 Jun 2026 11:07:34 +0000 Subject: [PATCH 06/10] =?UTF-8?q?per-property=20failure=20fails=20child=20?= =?UTF-8?q?SubTask=20without=20raising=20from=20handler=20=F0=9F=9F=A9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Sonnet 4.6 --- applications/modelling_e2e/handler.py | 46 +------- .../modelling_e2e/test_handler.py | 109 +----------------- 2 files changed, 12 insertions(+), 143 deletions(-) diff --git a/applications/modelling_e2e/handler.py b/applications/modelling_e2e/handler.py index c06d4b4d..a715c7bf 100644 --- a/applications/modelling_e2e/handler.py +++ b/applications/modelling_e2e/handler.py @@ -294,8 +294,6 @@ def handler(body: dict[str, Any], context: Any, orchestrator: TaskOrchestrator, products = catalogue_with_off_catalogue_overrides(read_session) solar_reader = SolarPostgresRepository(read_session) - failures: list[dict[str, Any]] = [] - for property_id in property_ids: child = orchestrator.create_child_subtask( task_id, inputs={"property_id": property_id} @@ -446,20 +444,8 @@ def handler(body: dict[str, Any], context: Any, orchestrator: TaskOrchestrator, try: orchestrator.run_subtask(child.id, work=_work) - 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), - } - ) + 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 @@ -474,29 +460,9 @@ def handler(body: dict[str, Any], context: Any, orchestrator: TaskOrchestrator, 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 + if skipped_certs: + raise RuntimeError( + f"skipped_unmappable_cohort_certs: {json.dumps(skipped_certs)}" + ) finally: read_session.close() diff --git a/tests/applications/modelling_e2e/test_handler.py b/tests/applications/modelling_e2e/test_handler.py index a07f7346..511fc3d6 100644 --- a/tests/applications/modelling_e2e/test_handler.py +++ b/tests/applications/modelling_e2e/test_handler.py @@ -513,9 +513,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]) @@ -585,9 +586,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() @@ -717,103 +717,6 @@ def test_empty_own_postcode_broadens_to_nearby_and_predicts() -> None: 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() - - # --------------------------------------------------------------------------- # Partial batch failure — per-property isolation # --------------------------------------------------------------------------- From fc2732550c9aa3dcfecb7d711db5bc6ae6147f36 Mon Sep 17 00:00:00 2001 From: Daniel Roth Date: Wed, 24 Jun 2026 11:09:30 +0000 Subject: [PATCH 07/10] =?UTF-8?q?skipped=20cohort=20certs=20are=20logged?= =?UTF-8?q?=20without=20failing=20the=20coordinator=20SubTask=20?= =?UTF-8?q?=F0=9F=9F=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Sonnet 4.6 --- .../modelling_e2e/test_handler.py | 67 +++++++++++++++++++ 1 file changed, 67 insertions(+) diff --git a/tests/applications/modelling_e2e/test_handler.py b/tests/applications/modelling_e2e/test_handler.py index 511fc3d6..2a01e258 100644 --- a/tests/applications/modelling_e2e/test_handler.py +++ b/tests/applications/modelling_e2e/test_handler.py @@ -386,6 +386,73 @@ def test_skipped_cohort_certs_fail_the_subtask_but_the_plan_is_still_saved() -> 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 # --------------------------------------------------------------------------- From e1bc5360f2698df0fb983949ad4a35225ef8ab4f Mon Sep 17 00:00:00 2001 From: Daniel Roth Date: Wed, 24 Jun 2026 11:11:19 +0000 Subject: [PATCH 08/10] =?UTF-8?q?skipped=20cohort=20certs=20are=20logged?= =?UTF-8?q?=20without=20failing=20the=20coordinator=20SubTask=20?= =?UTF-8?q?=F0=9F=9F=A9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Sonnet 4.6 --- applications/modelling_e2e/handler.py | 5 ----- .../modelling_e2e/test_handler.py | 21 ++++++------------- 2 files changed, 6 insertions(+), 20 deletions(-) diff --git a/applications/modelling_e2e/handler.py b/applications/modelling_e2e/handler.py index a715c7bf..32c76d7d 100644 --- a/applications/modelling_e2e/handler.py +++ b/applications/modelling_e2e/handler.py @@ -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 @@ -460,9 +459,5 @@ def handler(body: dict[str, Any], context: Any, orchestrator: TaskOrchestrator, f"{[s['certificate_number'] for s in skipped_certs]}" ) - if skipped_certs: - raise RuntimeError( - f"skipped_unmappable_cohort_certs: {json.dumps(skipped_certs)}" - ) finally: read_session.close() diff --git a/tests/applications/modelling_e2e/test_handler.py b/tests/applications/modelling_e2e/test_handler.py index 2a01e258..e2664a4b 100644 --- a/tests/applications/modelling_e2e/test_handler.py +++ b/tests/applications/modelling_e2e/test_handler.py @@ -293,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, ) @@ -372,16 +369,10 @@ 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() From d06e92cd75a87c3babbb44d15544e45e4862ebea Mon Sep 17 00:00:00 2001 From: Daniel Roth Date: Wed, 24 Jun 2026 11:15:58 +0000 Subject: [PATCH 09/10] =?UTF-8?q?task=5Fhandler:=20fix=20pyright=20unknown?= =?UTF-8?q?-type=20errors=20in=20pass=5Ftask=5Forchestrator=20branch=20?= =?UTF-8?q?=F0=9F=9F=AA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Sonnet 4.6 --- utilities/aws_lambda/task_handler.py | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/utilities/aws_lambda/task_handler.py b/utilities/aws_lambda/task_handler.py index ada8562e..f2064f57 100644 --- a/utilities/aws_lambda/task_handler.py +++ b/utilities/aws_lambda/task_handler.py @@ -67,16 +67,17 @@ def task_handler( try: if pass_task_orchestrator: - work = lambda body=body, t=task: func( - body, context, orchestrator, t.id + orchestrator.run_subtask( + subtask.id, + work=lambda: func(body, context, orchestrator, task.id), + cloud_logs_url=cloud_logs_url, ) else: - work = lambda body=body: func(body, context) - orchestrator.run_subtask( - subtask.id, - work=work, - cloud_logs_url=cloud_logs_url, - ) + 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)", From 6a5a2c39d31cfd63d07509c49e9b6314fc2d932d Mon Sep 17 00:00:00 2001 From: Daniel Roth Date: Wed, 24 Jun 2026 11:35:30 +0000 Subject: [PATCH 10/10] remove smoke tests from gh workflows --- .github/workflows/lambda_smoke_tests.yml | 268 +++++++++--------- .../local_handler/invoke_local_lambda.py | 4 +- 2 files changed, 141 insertions(+), 131 deletions(-) diff --git a/.github/workflows/lambda_smoke_tests.yml b/.github/workflows/lambda_smoke_tests.yml index 1830fcd9..d958abcb 100644 --- a/.github/workflows/lambda_smoke_tests.yml +++ b/.github/workflows/lambda_smoke_tests.yml @@ -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 \ No newline at end of file diff --git a/applications/modelling_e2e/local_handler/invoke_local_lambda.py b/applications/modelling_e2e/local_handler/invoke_local_lambda.py index b9a0d584..9ed2fc02 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_ids": [722473], + "property_ids": [727220, 727229], "portfolio_id": 796, "scenario_id": 1268, "no_solar": False, - "dry_run": False, + "dry_run": True, } ) }