Mark unmappable cohort certs as a subtask failure

Skipped cohort certs were previously surfaced only as outputs.result on a
completed subtask, so they were easy to miss. Treat them as a failure too:
once the batch has run to completion (so every modellable property is already
written to DB), raise if there were any per-property errors OR any skipped
certs. The run gets flagged for debugging without discarding the work done.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Jun-te Kim 2026-06-23 18:03:04 +00:00
parent 290097b1c7
commit af5b2b5f80
2 changed files with 40 additions and 33 deletions

View file

@ -453,22 +453,29 @@ def handler(body: dict[str, Any], context: Any) -> Optional[dict[str, Any]]:
f"{[s['certificate_number'] for s in skipped_certs]}" f"{[s['certificate_number'] for s in skipped_certs]}"
) )
if failures: # A property that errored AND a cohort cert the mapper could not consume
failed_ids = [f["property_id"] for f in failures] # are both surfaced as failures, so the subtask is marked failed and
# Persisted verbatim into the subtask's outputs.error (via # shows up for debugging. The whole batch has already run by this point —
# SubTask.fail), so include each property's error type + message — # every property that could be modelled was written to DB above — so
# not just the IDs — to make failed runs diagnosable without # failing here flags the run without discarding the work done so far.
# cross-referencing CloudWatch. if failures or skipped_certs:
message = ( parts: list[str] = []
f"failed property_ids: {failed_ids}; " if failures:
f"details: {json.dumps(failures)}" failed_ids = [f["property_id"] for f in failures]
) # Persisted verbatim into the subtask's outputs.error (via
if skipped_certs: # SubTask.fail): include each property's error type + message,
message += ( # not just the IDs, so failed runs are diagnosable without
f"; skipped_unmappable_cohort_certs: {json.dumps(skipped_certs)}" # cross-referencing CloudWatch.
parts.append(
f"failed property_ids: {failed_ids}; "
f"details: {json.dumps(failures)}"
) )
raise RuntimeError(message) if skipped_certs:
parts.append(
f"skipped_unmappable_cohort_certs: {json.dumps(skipped_certs)}"
)
raise RuntimeError("; ".join(parts))
return {"skipped_unmappable_cohort_certs": skipped_certs} if skipped_certs else None return None
finally: finally:
read_session.close() read_session.close()

View file

@ -214,16 +214,19 @@ def test_lodged_epc_path_saves_epc_plan_and_marks_modelled(
_baseline_orchestrator.return_value.run.assert_called_once_with([PROPERTY_ID]) _baseline_orchestrator.return_value.run.assert_called_once_with([PROPERTY_ID])
def test_skipped_cohort_certs_are_surfaced_in_the_outputs() -> None: 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 """Cohort certs the mapper can't consume are skipped (so prediction is not
aborted) and surfaced with cert numbers in the subtask outputs, so the aborted), then surfaced as a failure the subtask is marked failed (the
mapper gaps can be reported and closed.""" cert numbers land in outputs.error via the raised RuntimeError) so the
mapper gaps get debugged. The batch still ran to completion first, so the
property's plan was committed before the handler raised."""
from repositories.comparable_properties.epc_comparable_properties_repository import ( from repositories.comparable_properties.epc_comparable_properties_repository import (
SkippedCohortCert, SkippedCohortCert,
) )
mock_engine = _engine_mock([PROPERTY_ID], [UPRN], [POSTCODE]) mock_engine = _engine_mock([PROPERTY_ID], [UPRN], [POSTCODE])
mock_plan = _plan_mock() mock_plan = _plan_mock()
mock_uow = MagicMock()
skipped = [ skipped = [
SkippedCohortCert( SkippedCohortCert(
certificate_number="8257-7539-1649-0633-4992", certificate_number="8257-7539-1649-0633-4992",
@ -287,24 +290,21 @@ def test_skipped_cohort_certs_are_surfaced_in_the_outputs() -> None:
MockUoW = stack.enter_context( MockUoW = stack.enter_context(
patch("applications.modelling_e2e.handler.PostgresUnitOfWork") patch("applications.modelling_e2e.handler.PostgresUnitOfWork")
) )
MockUoW.return_value.__enter__.return_value = MagicMock() MockUoW.return_value.__enter__.return_value = mock_uow
MockUoW.return_value.__exit__.return_value = False MockUoW.return_value.__exit__.return_value = False
# Act # Act — the skipped cert fails the subtask, but only after the batch ran.
result = _call_handler(_BODY) with pytest.raises(RuntimeError) as excinfo:
_call_handler(_BODY)
# Assert — the handler's return (→ subtask outputs.result) carries the cert # Assert — the cert number + error reach outputs.error (the raised message),
# numbers + errors of every skipped cohort cert. # and the property's plan was still committed before the handler raised.
assert result == { message = str(excinfo.value)
"skipped_unmappable_cohort_certs": [ assert "skipped_unmappable_cohort_certs" in message
{ assert "8257-7539-1649-0633-4992" in message
"certificate_number": "8257-7539-1649-0633-4992", assert "RdSapSchema17_1: missing required field 'window'" in message
"error": ( mock_uow.plan.save.assert_called_once()
"ValueError: RdSapSchema17_1: missing required field 'window'" mock_uow.commit.assert_called_once()
),
}
]
}
# --------------------------------------------------------------------------- # ---------------------------------------------------------------------------