diff --git a/applications/abri/dispatch.py b/applications/abri/dispatch.py new file mode 100644 index 000000000..aef12a3b2 --- /dev/null +++ b/applications/abri/dispatch.py @@ -0,0 +1,59 @@ +"""Dispatches the flows named in a trigger message through the orchestrator. + +The scraper decides which flows fire; this is a dumb dispatcher. Flows run +in a fixed, retry-safe order — amend, then log, then tenant sync — so that +redelivery re-runs idempotent flows before non-idempotent ones: amend is +naturally idempotent, log is guarded by the job-number check, and tenant +contact creation (not idempotent) runs last so its failure can never cause +a duplicate job log, and a log failure can never duplicate contacts. +""" + +from typing import Dict, Protocol + +from applications.abri.abri_trigger_request import AbriTriggerRequest +from domain.abri.models import AbriRequestRejected, PlaceRef +from orchestration.abri_orchestrator import ( + AmendJobOrchestrationResult, + AppointmentChange, + ConfirmedSurveyBooking, + DealDatabaseGateway, + LogJobOrchestrationResult, + TenantDataSyncResult, +) + + +class AbriFlows(Protocol): + """The per-flow surface of the AbriOrchestrator that dispatch drives.""" + + def amend_job(self, change: AppointmentChange) -> AmendJobOrchestrationResult: ... + + def log_job(self, booking: ConfirmedSurveyBooking) -> LogJobOrchestrationResult: ... + + def sync_tenant_data( + self, place_ref: PlaceRef, deal_id: str + ) -> TenantDataSyncResult: ... + + +class AbriFlowRejectedError(Exception): + """An OpenHousing rejection, surfaced as a failure so the task fails. + + Retrying a permanent rejection is wasteful but harmless — every flow is + safe to re-run — and redelivery eventually parks the message in the + dead-letter queue, which is the ops inbox. + """ + + def __init__(self, flow: str, rejection: AbriRequestRejected) -> None: + self.flow = flow + self.rejection = rejection + super().__init__( + f"{flow} rejected by OpenHousing: {rejection.code}: {rejection.message}" + ) + + +def dispatch_abri_flows( + request: AbriTriggerRequest, + flows: AbriFlows, + deal_database: DealDatabaseGateway, +) -> Dict[str, str]: + """Run the fired flows in the fixed order; returns a PII-free summary.""" + raise NotImplementedError diff --git a/tests/applications/abri/test_dispatch.py b/tests/applications/abri/test_dispatch.py new file mode 100644 index 000000000..b41aa4d4b --- /dev/null +++ b/tests/applications/abri/test_dispatch.py @@ -0,0 +1,254 @@ +from datetime import date +from typing import Any, Dict, List, Optional + +import pytest + +from applications.abri.abri_trigger_request import AbriTriggerRequest +from applications.abri.dispatch import AbriFlowRejectedError, dispatch_abri_flows +from domain.abri.models import AbriRequestRejected, AppointmentAmended, PlaceRef +from orchestration.abri_orchestrator import ( + AmendJobOrchestrationResult, + AppointmentChange, + ConfirmedSurveyBooking, + LogJobOrchestrationResult, + LogJobSummary, + LogJobWriteBackError, + TenantDataSyncResult, + TenantDataSyncSummary, +) +from utilities.aws_lambda.task_handler import NonRetriableTaskError + +DEAL_ID = "9876543210" + + +def _request(flows: List[str]) -> AbriTriggerRequest: + return AbriTriggerRequest.model_validate( + { + "hubspot_deal_id": DEAL_ID, + "flows": flows, + "place_ref": "1007165", + "deal_name": "49 Admers Crescent", + "confirmed_survey_date": "2026-06-24", + "confirmed_survey_time": "14:30", + } + ) + + +class FakeOrchestrator: + """Records flow invocations in order; outcomes are programmable.""" + + def __init__(self) -> None: + self.calls: List[Any] = [] + self.amend_outcome: AmendJobOrchestrationResult = AppointmentAmended( + job_no="AC0439951", + appointment_date="24/06/2026", + appointment_time="PM", + ) + self.log_outcome: LogJobOrchestrationResult = LogJobSummary( + deal_id=DEAL_ID, job_no="AD0226519" + ) + self.log_error: Optional[Exception] = None + self.sync_outcome: TenantDataSyncResult = TenantDataSyncSummary( + tenancy_reference="TEN0001", + contact_ids=("101", "102"), + vulnerable_contact_count=1, + ) + + def amend_job(self, change: AppointmentChange) -> AmendJobOrchestrationResult: + self.calls.append(("amend_job", change)) + return self.amend_outcome + + def log_job(self, booking: ConfirmedSurveyBooking) -> LogJobOrchestrationResult: + self.calls.append(("log_job", booking)) + if self.log_error is not None: + raise self.log_error + return self.log_outcome + + def sync_tenant_data( + self, place_ref: PlaceRef, deal_id: str + ) -> TenantDataSyncResult: + self.calls.append(("sync_tenant_data", place_ref, deal_id)) + return self.sync_outcome + + +class FakeDealDatabase: + """In-memory stand-in for the deal-database gateway.""" + + def __init__(self) -> None: + self.job_nos: Dict[str, str] = {} + + def record_job_no(self, deal_id: str, job_no: str) -> None: + self.job_nos[deal_id] = job_no + + def job_no_for_deal(self, deal_id: str) -> Optional[str]: + return self.job_nos.get(deal_id) + + +@pytest.fixture() +def orchestrator() -> FakeOrchestrator: + return FakeOrchestrator() + + +@pytest.fixture() +def deal_database() -> FakeDealDatabase: + return FakeDealDatabase() + + +# --- fixed dispatch order: amend -> log -> tenant sync --- + + +def test_fired_flows_run_in_the_fixed_order_regardless_of_message_order( + orchestrator: FakeOrchestrator, deal_database: FakeDealDatabase +) -> None: + # Arrange + request = _request(["sync_tenant_data", "log_job", "amend_job"]) + + # Act + dispatch_abri_flows(request, orchestrator, deal_database) + + # Assert + assert [call[0] for call in orchestrator.calls] == [ + "amend_job", + "log_job", + "sync_tenant_data", + ] + + +def test_flows_the_message_does_not_name_do_not_run( + orchestrator: FakeOrchestrator, deal_database: FakeDealDatabase +) -> None: + # Arrange + request = _request(["sync_tenant_data"]) + + # Act + dispatch_abri_flows(request, orchestrator, deal_database) + + # Assert + assert orchestrator.calls == [ + ("sync_tenant_data", PlaceRef("1007165"), DEAL_ID) + ] + + +# --- the flows receive the deal fields from the message --- + + +def test_flows_receive_the_deal_fields_carried_by_the_message( + orchestrator: FakeOrchestrator, deal_database: FakeDealDatabase +) -> None: + # Arrange + request = _request(["amend_job", "log_job"]) + + # Act + dispatch_abri_flows(request, orchestrator, deal_database) + + # Assert + amend_call, log_call = orchestrator.calls + assert amend_call == ( + "amend_job", + AppointmentChange( + deal_id=DEAL_ID, + confirmed_survey_date=date(2026, 6, 24), + confirmed_survey_time="14:30", + ), + ) + assert log_call == ( + "log_job", + ConfirmedSurveyBooking( + deal_id=DEAL_ID, + place_ref=PlaceRef("1007165"), + deal_name="49 Admers Crescent", + confirmed_survey_date=date(2026, 6, 24), + confirmed_survey_time="14:30", + ), + ) + + +# --- idempotency guard: a job is never logged twice for the same deal --- + + +def test_the_log_flow_is_skipped_when_the_deal_already_has_a_job_no( + orchestrator: FakeOrchestrator, deal_database: FakeDealDatabase +) -> None: + # Arrange + request = _request(["log_job", "sync_tenant_data"]) + deal_database.job_nos[DEAL_ID] = "AD0226519" + + # Act + dispatch_abri_flows(request, orchestrator, deal_database) + + # Assert + assert [call[0] for call in orchestrator.calls] == ["sync_tenant_data"] + + +def test_the_log_flow_runs_when_the_deal_has_no_job_no( + orchestrator: FakeOrchestrator, deal_database: FakeDealDatabase +) -> None: + # Arrange + request = _request(["log_job"]) + + # Act + dispatch_abri_flows(request, orchestrator, deal_database) + + # Assert + assert [call[0] for call in orchestrator.calls] == ["log_job"] + + +# --- rejections raise, so the task fails and the DLQ is the ops inbox --- + + +def test_an_openhousing_rejection_raises_and_stops_the_dispatch( + orchestrator: FakeOrchestrator, deal_database: FakeDealDatabase +) -> None: + # Arrange + request = _request(["amend_job", "sync_tenant_data"]) + orchestrator.amend_outcome = AbriRequestRejected( + code="RelayError", message="Job_no 1019905 is invalid. Job not found." + ) + + # Act / Assert + with pytest.raises(AbriFlowRejectedError) as exc_info: + dispatch_abri_flows(request, orchestrator, deal_database) + assert "RelayError" in str(exc_info.value) + assert "Job_no 1019905 is invalid" in str(exc_info.value) + assert [call[0] for call in orchestrator.calls] == ["amend_job"] + + +# --- write-back failure: failed task, but the message is never requeued --- + + +def test_a_log_write_back_failure_raises_a_non_retriable_error( + orchestrator: FakeOrchestrator, deal_database: FakeDealDatabase +) -> None: + # Arrange + request = _request(["log_job", "sync_tenant_data"]) + orchestrator.log_error = LogJobWriteBackError( + message="HubSpot update failed", deal_id=DEAL_ID, job_no="AD0226519" + ) + + # Act / Assert: the orphaned job_no stays visible in the failure record, + # and the tenant sync never runs on this delivery. + with pytest.raises(NonRetriableTaskError) as exc_info: + dispatch_abri_flows(request, orchestrator, deal_database) + assert "AD0226519" in str(exc_info.value) + assert [call[0] for call in orchestrator.calls] == ["log_job"] + + +# --- the task output is a PII-free summary of what ran --- + + +def test_dispatch_returns_a_pii_free_summary_of_what_ran( + orchestrator: FakeOrchestrator, deal_database: FakeDealDatabase +) -> None: + # Arrange + request = _request(["amend_job", "log_job", "sync_tenant_data"]) + deal_database.job_nos[DEAL_ID] = "AC0439951" + + # Act + summary = dispatch_abri_flows(request, orchestrator, deal_database) + + # Assert + assert summary == { + "amend_job": "appointment amended to 24/06/2026 PM (job AC0439951)", + "log_job": "skipped: job AC0439951 already logged", + "sync_tenant_data": "2 contacts created (1 vulnerable)", + }