mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-30 13:10:47 +00:00
per-property failure fails child SubTask without raising from handler 🟩
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
parent
39f028f03a
commit
c2a84d3ee8
2 changed files with 12 additions and 143 deletions
|
|
@ -294,8 +294,6 @@ def handler(body: dict[str, Any], context: Any, orchestrator: TaskOrchestrator,
|
||||||
products = catalogue_with_off_catalogue_overrides(read_session)
|
products = catalogue_with_off_catalogue_overrides(read_session)
|
||||||
solar_reader = SolarPostgresRepository(read_session)
|
solar_reader = SolarPostgresRepository(read_session)
|
||||||
|
|
||||||
failures: list[dict[str, Any]] = []
|
|
||||||
|
|
||||||
for property_id in property_ids:
|
for property_id in property_ids:
|
||||||
child = orchestrator.create_child_subtask(
|
child = orchestrator.create_child_subtask(
|
||||||
task_id, inputs={"property_id": property_id}
|
task_id, inputs={"property_id": property_id}
|
||||||
|
|
@ -446,20 +444,8 @@ def handler(body: dict[str, Any], context: Any, orchestrator: TaskOrchestrator,
|
||||||
|
|
||||||
try:
|
try:
|
||||||
orchestrator.run_subtask(child.id, work=_work)
|
orchestrator.run_subtask(child.id, work=_work)
|
||||||
except Exception as error: # noqa: BLE001
|
except Exception: # noqa: BLE001
|
||||||
logger.error(
|
pass
|
||||||
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),
|
|
||||||
}
|
|
||||||
)
|
|
||||||
|
|
||||||
# Cohort certs the mapper could not consume were skipped (not aborted on)
|
# Cohort certs the mapper could not consume were skipped (not aborted on)
|
||||||
# so prediction could proceed; surface them — with cert numbers — in the
|
# 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]}"
|
f"{[s['certificate_number'] for s in skipped_certs]}"
|
||||||
)
|
)
|
||||||
|
|
||||||
# A property that errored AND a cohort cert the mapper could not consume
|
if skipped_certs:
|
||||||
# are both surfaced as failures, so the subtask is marked failed and
|
raise RuntimeError(
|
||||||
# shows up for debugging. The whole batch has already run by this point —
|
f"skipped_unmappable_cohort_certs: {json.dumps(skipped_certs)}"
|
||||||
# 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:
|
finally:
|
||||||
read_session.close()
|
read_session.close()
|
||||||
|
|
|
||||||
|
|
@ -513,9 +513,10 @@ def test_prediction_path_saves_predicted_epc_plan_and_baseline(
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
def test_empty_cohort_gates_property_out_and_raises() -> None:
|
def test_empty_cohort_gates_property_out_without_saving() -> None:
|
||||||
"""When candidates_for returns an empty list the property cannot be
|
"""When candidates_for returns an empty list the property cannot be predicted;
|
||||||
predicted; the handler records it as an error and raises RuntimeError."""
|
the failure is recorded in the child SubTask and the handler returns without
|
||||||
|
raising — no plan is saved."""
|
||||||
# Arrange
|
# Arrange
|
||||||
mock_engine = _engine_mock([PROPERTY_ID], [UPRN], [POSTCODE])
|
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")
|
patch("applications.modelling_e2e.handler.PostgresUnitOfWork")
|
||||||
)
|
)
|
||||||
|
|
||||||
# Act
|
# Act — no raise; child SubTask is failed, handler continues
|
||||||
with pytest.raises(RuntimeError, match=str(PROPERTY_ID)):
|
_call_handler(_BODY)
|
||||||
_call_handler(_BODY)
|
|
||||||
|
|
||||||
# UoW never entered — the property errored before the write block
|
# UoW never entered — the property errored before the write block
|
||||||
MockUoW.return_value.__enter__.assert_not_called()
|
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()
|
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
|
# Partial batch failure — per-property isolation
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue