diff --git a/.github/workflows/_deploy_lambda.yml b/.github/workflows/_deploy_lambda.yml index 9f6b07beb..e8e2de981 100644 --- a/.github/workflows/_deploy_lambda.yml +++ b/.github/workflows/_deploy_lambda.yml @@ -100,6 +100,15 @@ on: TF_VAR_open_epc_api_token: required: false + + TF_VAR_abri_relay_url: + required: false + TF_VAR_abri_relay_username: + required: false + TF_VAR_abri_relay_password: + required: false + TF_VAR_abri_relay_default_resource: + required: false jobs: deploy: runs-on: ubuntu-latest @@ -175,6 +184,10 @@ jobs: TF_VAR_magicplan_api_key: ${{ secrets.TF_VAR_magicplan_api_key }} TF_VAR_openai_api_key: ${{ secrets.TF_VAR_openai_api_key }} TF_VAR_open_epc_api_token: ${{ secrets.TF_VAR_open_epc_api_token }} + TF_VAR_abri_relay_url: ${{ secrets.TF_VAR_abri_relay_url }} + TF_VAR_abri_relay_username: ${{ secrets.TF_VAR_abri_relay_username }} + TF_VAR_abri_relay_password: ${{ secrets.TF_VAR_abri_relay_password }} + TF_VAR_abri_relay_default_resource: ${{ secrets.TF_VAR_abri_relay_default_resource }} run: | ECR_REPO_URL_VAR="" if [[ -n "${{ inputs.ecr_repo }}" ]]; then @@ -228,6 +241,10 @@ jobs: TF_VAR_magicplan_api_key: ${{ secrets.TF_VAR_magicplan_api_key }} TF_VAR_openai_api_key: ${{ secrets.TF_VAR_openai_api_key }} TF_VAR_open_epc_api_token: ${{ secrets.TF_VAR_open_epc_api_token }} + TF_VAR_abri_relay_url: ${{ secrets.TF_VAR_abri_relay_url }} + TF_VAR_abri_relay_username: ${{ secrets.TF_VAR_abri_relay_username }} + TF_VAR_abri_relay_password: ${{ secrets.TF_VAR_abri_relay_password }} + TF_VAR_abri_relay_default_resource: ${{ secrets.TF_VAR_abri_relay_default_resource }} run: | EXTRA_VARS="" if [[ -n "${{ inputs.ecr_repo }}" ]]; then diff --git a/.github/workflows/deploy_terraform.yml b/.github/workflows/deploy_terraform.yml index 0f72a33f1..510ac147a 100644 --- a/.github/workflows/deploy_terraform.yml +++ b/.github/workflows/deploy_terraform.yml @@ -700,6 +700,48 @@ jobs: TF_VAR_magicplan_customer_id: ${{ secrets.MAGICPLAN_CUSTOMER_ID }} TF_VAR_magicplan_api_key: ${{ secrets.MAGICPLAN_API_KEY }} + # ============================================================ + # Build Abri Lambda image + # ============================================================ + abri_image: + needs: [determine_stage, shared_terraform] + uses: ./.github/workflows/_build_image.yml + with: + ecr_repo: abri-${{ needs.determine_stage.outputs.stage }} + dockerfile_path: applications/abri/handler/Dockerfile + build_context: . + secrets: + AWS_ACCESS_KEY_ID: ${{ secrets.DEV_AWS_ACCESS_KEY_ID }} + AWS_SECRET_ACCESS_KEY: ${{ secrets.DEV_AWS_SECRET_ACCESS_KEY }} + AWS_REGION: ${{ secrets.DEV_AWS_REGION }} + + # ============================================================ + # Deploy Abri Lambda + # ============================================================ + abri_api_lambda: + needs: [abri_image, determine_stage] + uses: ./.github/workflows/_deploy_lambda.yml + with: + lambda_name: abri + lambda_path: deployment/terraform/lambda/abri + stage: ${{ needs.determine_stage.outputs.stage }} + ecr_repo: abri-${{ needs.determine_stage.outputs.stage }} + image_digest: ${{ needs.abri_image.outputs.image_digest }} + terraform_apply: ${{ needs.determine_stage.outputs.terraform_apply }} + secrets: + AWS_ACCESS_KEY_ID: ${{ secrets.DEV_AWS_ACCESS_KEY_ID }} + AWS_SECRET_ACCESS_KEY: ${{ secrets.DEV_AWS_SECRET_ACCESS_KEY }} + AWS_REGION: ${{ secrets.DEV_AWS_REGION }} + TF_VAR_db_host: ${{ secrets.DEV_DB_HOST }} + TF_VAR_db_name: ${{ secrets.DEV_DB_NAME }} + TF_VAR_db_port: ${{ secrets.DEV_DB_PORT }} + TF_VAR_hubspot_api_key: ${{ secrets.HUBSPOT_API_KEY }} + # Awaiting API Key from Abri + TF_VAR_abri_relay_url: ${{ secrets.ABRI_RELAY_URL }} + TF_VAR_abri_relay_username: ${{ secrets.ABRI_RELAY_USERNAME }} + TF_VAR_abri_relay_password: ${{ secrets.ABRI_RELAY_PASSWORD }} + TF_VAR_abri_relay_default_resource: ${{ secrets.ABRI_RELAY_DEFAULT_RESOURCE }} + # ============================================================ # Build Audit Generator image # ============================================================ @@ -778,7 +820,7 @@ jobs: # Deploy Hubspot ETL Lambda # ============================================================ hubspot_etl_lambda: - needs: [hubspot_etl_image, determine_stage, pashub_to_ara_lambda, magic_plan_lambda] + needs: [hubspot_etl_image, determine_stage, pashub_to_ara_lambda, magic_plan_lambda, abri_lambda] uses: ./.github/workflows/_deploy_lambda.yml with: lambda_name: hubspot-etl-to-ara diff --git a/applications/abri/__init__.py b/applications/abri/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/applications/abri/abri_trigger_request.py b/applications/abri/abri_trigger_request.py new file mode 100644 index 000000000..0a9346ca9 --- /dev/null +++ b/applications/abri/abri_trigger_request.py @@ -0,0 +1,44 @@ +"""The message contract between the HubSpot scraper and the abri lambda. + +One message per scrape, naming every Abri flow that fired plus the deal +fields those flows need. A message missing a field its flow needs fails +validation — the resulting failed task / DLQ entry is the agreed +visibility surface. +""" + +from datetime import date +from typing import Dict, List, Literal, Optional, Tuple + +from pydantic import BaseModel, ConfigDict, Field, model_validator + +AbriFlow = Literal["amend_job", "log_job", "sync_tenant_data"] + +# The deal fields each flow reads; hubspot_deal_id is always required. +_FIELDS_BY_FLOW: Dict[AbriFlow, Tuple[str, ...]] = { + "amend_job": ("confirmed_survey_date",), + "log_job": ("place_ref", "deal_name", "confirmed_survey_date"), + "sync_tenant_data": ("place_ref",), +} + + +class AbriTriggerRequest(BaseModel): + model_config = ConfigDict(extra="ignore") + + hubspot_deal_id: str + flows: List[AbriFlow] = Field(min_length=1) + place_ref: Optional[str] = None + deal_name: Optional[str] = None + confirmed_survey_date: Optional[date] = None + confirmed_survey_time: Optional[str] = None + + @model_validator(mode="after") + def _each_fired_flow_has_its_fields(self) -> "AbriTriggerRequest": + missing = [ + f"flow {flow} requires {field}" + for flow in self.flows + for field in _FIELDS_BY_FLOW[flow] + if getattr(self, field) is None + ] + if missing: + raise ValueError("; ".join(missing)) + return self diff --git a/applications/abri/dispatch.py b/applications/abri/dispatch.py new file mode 100644 index 000000000..f69981162 --- /dev/null +++ b/applications/abri/dispatch.py @@ -0,0 +1,137 @@ +"""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, + LogJobWriteBackError, + TenantDataSyncResult, +) +from utilities.aws_lambda.task_handler import NonRetriableTaskError + + +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.""" + summary: Dict[str, str] = {} + + if "amend_job" in request.flows: + summary["amend_job"] = _run_amend(request, flows) + if "log_job" in request.flows: + summary["log_job"] = _run_log(request, flows, deal_database) + if "sync_tenant_data" in request.flows: + summary["sync_tenant_data"] = _run_tenant_sync(request, flows) + + return summary + + +def _run_amend(request: AbriTriggerRequest, flows: AbriFlows) -> str: + if request.confirmed_survey_date is None: + raise ValueError("amend_job fired without a confirmed_survey_date") + outcome = flows.amend_job( + AppointmentChange( + deal_id=request.hubspot_deal_id, + confirmed_survey_date=request.confirmed_survey_date, + confirmed_survey_time=request.confirmed_survey_time, + ) + ) + if isinstance(outcome, AbriRequestRejected): + raise AbriFlowRejectedError(flow="amend_job", rejection=outcome) + return ( + f"appointment amended to {outcome.appointment_date} " + f"{outcome.appointment_time} (job {outcome.job_no})" + ) + + +def _run_log( + request: AbriTriggerRequest, + flows: AbriFlows, + deal_database: DealDatabaseGateway, +) -> str: + existing_job_no = deal_database.job_no_for_deal(request.hubspot_deal_id) + if existing_job_no is not None: + return f"skipped: job {existing_job_no} already logged" + + if ( + request.place_ref is None + or request.deal_name is None + or request.confirmed_survey_date is None + ): + raise ValueError("log_job fired without place_ref/deal_name/survey date") + try: + outcome = flows.log_job( + ConfirmedSurveyBooking( + deal_id=request.hubspot_deal_id, + place_ref=PlaceRef(request.place_ref), + deal_name=request.deal_name, + confirmed_survey_date=request.confirmed_survey_date, + confirmed_survey_time=request.confirmed_survey_time, + ) + ) + except LogJobWriteBackError as error: + # The job exists in OpenHousing; a redelivery would log a duplicate. + # The orphaned job_no stays visible in the failure record. + raise NonRetriableTaskError(str(error)) from error + if isinstance(outcome, AbriRequestRejected): + raise AbriFlowRejectedError(flow="log_job", rejection=outcome) + return f"job {outcome.job_no} logged" + + +def _run_tenant_sync(request: AbriTriggerRequest, flows: AbriFlows) -> str: + if request.place_ref is None: + raise ValueError("sync_tenant_data fired without a place_ref") + outcome = flows.sync_tenant_data( + place_ref=PlaceRef(request.place_ref), + deal_id=request.hubspot_deal_id, + ) + if isinstance(outcome, AbriRequestRejected): + raise AbriFlowRejectedError(flow="sync_tenant_data", rejection=outcome) + return ( + f"{len(outcome.contact_ids)} contacts created " + f"({outcome.vulnerable_contact_count} vulnerable)" + ) diff --git a/applications/abri/handler.py b/applications/abri/handler.py new file mode 100644 index 000000000..d9a80ae63 --- /dev/null +++ b/applications/abri/handler.py @@ -0,0 +1,59 @@ +"""Lambda entry point for the Abri OpenHousing flows. + +Consumes the abri SQS queue (one trigger message per scrape, sent by the +HubSpot scraper) and dispatches the fired flows through the AbriOrchestrator. +Env-to-clients glue only; behaviour lives behind the dispatch seam. +""" + +import os +from typing import Any, Dict + +from hubspot.client import Client # type: ignore[reportMissingTypeStubs] +from sqlalchemy.pool import NullPool +from sqlmodel import Session + +from applications.abri.abri_trigger_request import AbriTriggerRequest +from applications.abri.dispatch import dispatch_abri_flows +from domain.tasks.tasks import Source +from infrastructure.abri.abri_client import AbriClient +from infrastructure.abri.config import AbriConfig +from infrastructure.hubspot.deal_contacts_client import HubspotDealContactsClient +from infrastructure.hubspot.deal_properties_client import HubspotDealPropertiesClient +from infrastructure.postgres.config import PostgresConfig +from infrastructure.postgres.engine import make_engine +from orchestration.abri_orchestrator import AbriOrchestrator +from repositories.hubspot_deals.deal_database_postgres_gateway import ( + DealDatabasePostgresGateway, +) +from utilities.aws_lambda.task_handler import task_handler +from utilities.logger import setup_logger + +logger = setup_logger() + + +@task_handler(task_source="abri_api", source=Source.HUBSPOT_DEAL) +def handler(body: dict[str, Any], context: Any) -> Dict[str, str]: + request = AbriTriggerRequest.model_validate(body) + + abri_client = AbriClient(config=AbriConfig.from_env(os.environ)) + sdk_client: Client = Client.create( # type: ignore[reportUnknownMemberType] + access_token=os.environ["HUBSPOT_API_KEY"] + ) + + # NullPool: a fresh engine per invocation, so pooling would only + # accumulate idle connections across warm Lambda containers. + engine = make_engine(PostgresConfig.from_env(dict(os.environ)), poolclass=NullPool) + with Session(engine) as session: + deal_database = DealDatabasePostgresGateway(session=session) + orchestrator = AbriOrchestrator( + abri_client=abri_client, + deal_contacts=HubspotDealContactsClient(sdk_client=sdk_client), + deal_properties=HubspotDealPropertiesClient(sdk_client=sdk_client), + deal_database=deal_database, + ) + summary = dispatch_abri_flows(request, orchestrator, deal_database) + + logger.info( + "Abri flows completed for deal %s: %s", request.hubspot_deal_id, summary + ) + return summary diff --git a/applications/abri/handler/Dockerfile b/applications/abri/handler/Dockerfile new file mode 100644 index 000000000..61aff08c3 --- /dev/null +++ b/applications/abri/handler/Dockerfile @@ -0,0 +1,18 @@ +FROM public.ecr.aws/lambda/python:3.11 + +WORKDIR /var/task + +COPY applications/abri/handler/requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +COPY utilities/ utilities/ +COPY utils/ utils/ +COPY backend/ backend/ +COPY applications/ applications/ +COPY domain/ domain/ +COPY datatypes/ datatypes/ +COPY orchestration/ orchestration/ +COPY repositories/ repositories/ +COPY infrastructure/ infrastructure/ + +CMD ["applications.abri.handler.handler"] diff --git a/applications/abri/handler/requirements.txt b/applications/abri/handler/requirements.txt new file mode 100644 index 000000000..fafca4ae3 --- /dev/null +++ b/applications/abri/handler/requirements.txt @@ -0,0 +1,8 @@ +awslambdaric +requests +sqlalchemy==2.0.36 +sqlmodel +psycopg2-binary==2.9.10 +pydantic-settings==2.6.0 +boto3==1.35.44 +hubspot-api-client diff --git a/backend/app/config.py b/backend/app/config.py index d1c5f3ace..af79b0106 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -40,6 +40,7 @@ class Settings(BaseSettings): CATEGORISATION_SQS_URL: str = "changeme" PASHUB_TO_ARA_SQS_URL: str = "changeme" MAGICPLAN_SQS_URL: str = "changeme" + ABRI_SQS_URL: str = "changeme" POSTCODE_SPLITTER_SQS_URL: str = "changeme" COMBINER_SQS_URL: str = "changeme" LANDLORD_OVERRIDES_SQS_URL: str = "changeme" diff --git a/backend/app/db/models/hubspot_deal_data.py b/backend/app/db/models/hubspot_deal_data.py index c221cbcdc..88bb5a69f 100644 --- a/backend/app/db/models/hubspot_deal_data.py +++ b/backend/app/db/models/hubspot_deal_data.py @@ -71,8 +71,10 @@ class HubspotDealData(SQLModel, table=True): confirmed_survey_time: Optional[str] = Field(default=None) surveyed_date: Optional[datetime] = Field(default=None) design_type: Optional[str] = Field(default=None) - # OpenHousing job number; HubSpot property ID is client_booking_reference. - job_no: Optional[str] = Field(default=None) + # The OpenHousing job number, named after the HubSpot property that + # carries it (matching the column the frontend repo's drizzle migration + # created). Only the domain calls it job_no. + client_booking_reference: Optional[str] = Field(default=None) survey_type: Optional[str] = Field(default=None) measures_for_pibi_ordered: Optional[str] = Field(default=None) diff --git a/deployment/terraform/lambda/abri_api/main.tf b/deployment/terraform/lambda/abri_api/main.tf new file mode 100644 index 000000000..a7bb1d6cf --- /dev/null +++ b/deployment/terraform/lambda/abri_api/main.tf @@ -0,0 +1,40 @@ +data "aws_secretsmanager_secret_version" "db_credentials" { + secret_id = "${var.stage}/assessment_model/db_credentials" +} + +locals { + db_credentials = jsondecode(data.aws_secretsmanager_secret_version.db_credentials.secret_string) +} + +module "lambda" { + source = "../../modules/lambda_with_sqs" + + name = "abri_api" + stage = var.stage + + image_uri = local.image_uri + + maximum_concurrency = var.maximum_concurrency + reserved_concurrent_executions = var.reserved_concurrent_executions + batch_size = var.batch_size + + environment = { + STAGE = var.stage + LOG_LEVEL = "info" + + # Abri relay credentials are per-stage: the dev/staging stage points at + # Abri's UAT relay endpoint; production values are pending from Abri. + ABRI_RELAY_URL = var.abri_relay_url + ABRI_RELAY_USERNAME = var.abri_relay_username + ABRI_RELAY_PASSWORD = var.abri_relay_password + ABRI_RELAY_DEFAULT_RESOURCE = var.abri_relay_default_resource + + HUBSPOT_API_KEY = var.hubspot_api_key + + POSTGRES_USERNAME = local.db_credentials.db_assessment_model_username + POSTGRES_PASSWORD = local.db_credentials.db_assessment_model_password + POSTGRES_HOST = var.db_host + POSTGRES_DATABASE = var.db_name + POSTGRES_PORT = var.db_port + } +} diff --git a/deployment/terraform/lambda/abri_api/outputs.tf b/deployment/terraform/lambda/abri_api/outputs.tf new file mode 100644 index 000000000..7bbe29cd3 --- /dev/null +++ b/deployment/terraform/lambda/abri_api/outputs.tf @@ -0,0 +1,9 @@ +output "abri_api_queue_url" { + value = module.lambda.queue_url + description = "URL of the Abri SQS queue" +} + +output "abri_api_queue_arn" { + value = module.lambda.queue_arn + description = "ARN of the Abri SQS queue" +} diff --git a/deployment/terraform/lambda/abri_api/provider.tf b/deployment/terraform/lambda/abri_api/provider.tf new file mode 100644 index 000000000..3ae50128a --- /dev/null +++ b/deployment/terraform/lambda/abri_api/provider.tf @@ -0,0 +1,16 @@ +terraform { + required_providers { + aws = { + source = "hashicorp/aws" + version = ">= 5.0" + } + } + + backend "s3" { + bucket = "abri-terraform-state" + key = "terraform.tfstate" + region = "eu-west-2" + } + + required_version = ">= 1.2.0" +} diff --git a/deployment/terraform/lambda/abri_api/variables.tf b/deployment/terraform/lambda/abri_api/variables.tf new file mode 100644 index 000000000..e431db62b --- /dev/null +++ b/deployment/terraform/lambda/abri_api/variables.tf @@ -0,0 +1,84 @@ +variable "lambda_name" { + type = string + description = "Logical name of the lambda" + default = "abri_api" +} + +variable "stage" { + description = "Deployment stage (e.g. dev, prod)" + type = string +} + +variable "ecr_repo_url" { + type = string + description = "ECR repository URL (no tag, no digest)" +} + +variable "image_digest" { + type = string + description = "Image digest (sha256:...)" +} + +variable "maximum_concurrency" { + type = number + default = null + description = "Maximum number of concurrent Lambda invocations from SQS (2-1000). null = no limit." +} + +variable "reserved_concurrent_executions" { + type = number + default = 1 +} + +variable "batch_size" { + type = number + default = 1 +} + +locals { + image_uri = "${var.ecr_repo_url}@${var.image_digest}" +} + +output "resolved_image_uri" { + value = local.image_uri +} + +variable "abri_relay_url" { + type = string + sensitive = true +} + +variable "abri_relay_username" { + type = string + sensitive = true +} + +variable "abri_relay_password" { + type = string + sensitive = true +} + +variable "abri_relay_default_resource" { + type = string + sensitive = true +} + +variable "hubspot_api_key" { + type = string + sensitive = true +} + +variable "db_host" { + type = string + sensitive = true +} + +variable "db_name" { + type = string + sensitive = true +} + +variable "db_port" { + type = string + sensitive = true +} diff --git a/deployment/terraform/lambda/hubspot_deal_etl/main.tf b/deployment/terraform/lambda/hubspot_deal_etl/main.tf index ffb5f6f53..0c06ea7c0 100644 --- a/deployment/terraform/lambda/hubspot_deal_etl/main.tf +++ b/deployment/terraform/lambda/hubspot_deal_etl/main.tf @@ -25,6 +25,15 @@ data "terraform_remote_state" "magic_plan" { } } +data "terraform_remote_state" "abri_api" { + backend = "s3" + config = { + bucket = "abri-api-terraform-state" + key = "env:/${var.stage}/terraform.tfstate" + region = "eu-west-2" + } +} + data "aws_secretsmanager_secret_version" "db_credentials" { secret_id = "${var.stage}/assessment_model/db_credentials" } @@ -59,6 +68,7 @@ module "hubspot_deal_etl" { PASHUB_TO_ARA_SQS_URL = data.terraform_remote_state.pashub_to_ara.outputs.pashub_to_ara_queue_url MAGICPLAN_SQS_URL = data.terraform_remote_state.magic_plan.outputs.magic_plan_queue_url + ABRI_SQS_URL = data.terraform_remote_state.abri.outputs.abri_api_queue_url } } @@ -100,4 +110,18 @@ module "hubspot_deal_etl_magicplan_sqs_policy" { resource "aws_iam_role_policy_attachment" "hubspot_deal_etl_magicplan_sqs_send" { role = module.hubspot_deal_etl.role_name policy_arn = module.hubspot_deal_etl_magicplan_sqs_policy.policy_arn +} + +module "hubspot_deal_etl_abri_sqs_policy" { + source = "../../modules/general_iam_policy" + + policy_name = "hubspot-deal-etl-abri-sqs-send-${var.stage}" + policy_description = "Allow HubSpot ETL Lambda to send messages to the Abri queue" + actions = ["sqs:SendMessage"] + resources = [data.terraform_remote_state.abri.outputs.abri_api_queue_arn] +} + +resource "aws_iam_role_policy_attachment" "hubspot_deal_etl_abri_sqs_send" { + role = module.hubspot_deal_etl.role_name + policy_arn = module.hubspot_deal_etl_abri_sqs_policy.policy_arn } \ No newline at end of file diff --git a/deployment/terraform/shared/main.tf b/deployment/terraform/shared/main.tf index 3ff5bda4b..b4cdafbcc 100644 --- a/deployment/terraform/shared/main.tf +++ b/deployment/terraform/shared/main.tf @@ -894,3 +894,17 @@ output "modelling_e2e_ecr_url" { value = module.modelling_e2e_registry.repository_url } + +################################################ +# Abri OpenHousing – Lambda +################################################ +module "abri_api_state_bucket" { + source = "../modules/tf_state_bucket" + bucket_name = "abri-terraform-state" +} + +module "abri_api_registry" { + source = "../modules/container_registry" + name = "abri_api" + stage = var.stage +} diff --git a/etl/hubspot/hubspotDataTodB.py b/etl/hubspot/hubspotDataTodB.py index 49719da3b..54c174c3f 100644 --- a/etl/hubspot/hubspotDataTodB.py +++ b/etl/hubspot/hubspotDataTodB.py @@ -274,7 +274,7 @@ class HubspotDataToDb: deal_data.get("confirmed_survey_date") ), "confirmed_survey_time": deal_data.get("confirmed_survey_time"), - "job_no": deal_data.get("client_booking_reference"), + "client_booking_reference": deal_data.get("client_booking_reference"), "surveyed_date": parse_hs_date(deal_data.get("surveyed_date")), "design_type": deal_data.get("design_type"), "survey_type": deal_data.get("survey_type"), @@ -383,7 +383,7 @@ class HubspotDataToDb: surveyor=deal_data.get("surveyor"), confirmed_survey_date=parse_hs_date(deal_data.get("confirmed_survey_date")), confirmed_survey_time=deal_data.get("confirmed_survey_time"), - job_no=deal_data.get("client_booking_reference"), + client_booking_reference=deal_data.get("client_booking_reference"), surveyed_date=parse_hs_date(deal_data.get("surveyed_date")), design_type=deal_data.get("design_type"), survey_type=deal_data.get("survey_type"), diff --git a/etl/hubspot/hubspot_deal_differ.py b/etl/hubspot/hubspot_deal_differ.py index 39a1956bb..e5b32ce29 100644 --- a/etl/hubspot/hubspot_deal_differ.py +++ b/etl/hubspot/hubspot_deal_differ.py @@ -1,4 +1,4 @@ -from typing import Dict, List, Optional +from typing import Any, Dict, List, Optional from backend.app.db.models.hubspot_deal_data import HubspotDealData from etl.hubspot.project_data import ProjectData @@ -21,8 +21,14 @@ class HubspotDealDiffer: "tenant refusal", "not viable", ] - ABRI_CONDITION_ASSOCIATED_PROJECT_ID = "123456" - ABRI_CONDITION_PROJECT_CODE = "Abri Condition" + # Verified against the portal (2026-07-07): Abri stock-condition deals + # carry no project association, so the project_code match is the branch + # that fires in production; the id below is the project object id + # embedded at the end of that code, kept for when associations appear. + ABRI_CONDITION_ASSOCIATED_PROJECT_ID = "1247536318670" + ABRI_CONDITION_PROJECT_CODE = ( + "[Abri Stock Condition - Privately Funded - 062026 - 1247536318670]" + ) @staticmethod def check_for_db_update_trigger( @@ -97,7 +103,7 @@ class HubspotDealDiffer: "design_type": "design_type", "surveyor": "surveyor", "confirmed_survey_time": "confirmed_survey_time", - "client_booking_reference": "job_no", + "client_booking_reference": "client_booking_reference", "survey_type": "survey_type", "measures_for_pibi_ordered": "measures_for_pibi_ordered", "property_halted_reason": "property_halted_reason", @@ -183,6 +189,46 @@ class HubspotDealDiffer: return False + @staticmethod + def check_abri_triggers_and_construct_message( + hubspot_deal_id: str, + new_deal: Dict[str, str], + new_project: Optional[ProjectData], + new_listing: Optional[Dict[str, str]], + old_deal: HubspotDealData, + ) -> Optional[Dict[str, Any]]: + flows: List[str] = [] + if HubspotDealDiffer.check_for_abri_job_amendment( + new_deal=new_deal, new_project=new_project, old_deal=old_deal + ): + flows.append("amend_job") + if HubspotDealDiffer.check_for_abri_job_logging( + new_deal=new_deal, new_project=new_project, old_deal=old_deal + ): + flows.append("log_job") + if HubspotDealDiffer.check_for_abri_tenant_data_fetch( + new_deal=new_deal, new_project=new_project, old_deal=old_deal + ): + flows.append("sync_tenant_data") + + if not flows: + return None + + confirmed_survey_date = parse_hs_date(new_deal.get("confirmed_survey_date")) + listing = new_listing or {} + return { + "hubspot_deal_id": hubspot_deal_id, + "flows": flows, + "place_ref": listing.get("owner_property_id"), + "deal_name": new_deal.get("dealname"), + "confirmed_survey_date": ( + confirmed_survey_date.date().isoformat() + if confirmed_survey_date is not None + else None + ), + "confirmed_survey_time": new_deal.get("confirmed_survey_time"), + } + @staticmethod def check_for_abri_job_logging( new_deal: Dict[str, str], diff --git a/etl/hubspot/scripts/scraper/main.py b/etl/hubspot/scripts/scraper/main.py index 589e526b8..cd78b70c5 100644 --- a/etl/hubspot/scripts/scraper/main.py +++ b/etl/hubspot/scripts/scraper/main.py @@ -110,21 +110,30 @@ def handler(body: dict[str, Any], context: Any) -> None: f"Triggering Pas Hub file fetcher for HubSpot deal ID {hubspot_deal_id}" ) _trigger_pashub_fetcher(sqs_client, hubspot_deal_id, hubspot_deal) - else: - logger.info( - f"Not Triggering PasHub file fetcher for HubSpot deal ID {hubspot_deal_id}" - ) if HubspotDealDiffer.check_for_magicplan_trigger( new_deal=hubspot_deal, old_deal=db_deal ): - logger.info( - f"Triggering MagicPlan fetcher for HubSpot deal ID {hubspot_deal_id}" - ) _trigger_magicplan_fetcher( sqs_client, hubspot_deal, listing, hubspot_deal_id ) + abri_message: Optional[Dict[str, Any]] = ( + HubspotDealDiffer.check_abri_triggers_and_construct_message( + hubspot_deal_id=hubspot_deal_id, + new_deal=hubspot_deal, + new_project=project, + new_listing=listing, + old_deal=db_deal, + ) + ) + if abri_message is not None: + logger.info( + f"Triggering Abri flows {abri_message['flows']} for HubSpot " + f"deal ID {hubspot_deal_id}" + ) + _trigger_abri_api_lambda(sqs_client, abri_message) + print("done") @@ -134,6 +143,8 @@ def _trigger_magicplan_fetcher( listing: Optional[dict[str, str]], hubspot_deal_id: str, ) -> None: + logger.info(f"Triggering MagicPlan fetcher for HubSpot deal ID {hubspot_deal_id}") + message_body = { "address": hubspot_deal.get("dealname"), "hubspot_deal_id": hubspot_deal_id, @@ -146,6 +157,14 @@ def _trigger_magicplan_fetcher( logger.info(f"Sent message to MagicPlan queue. MessageId: {response['MessageId']}") +def _trigger_abri_api_lambda(sqs_client: Any, abri_message: Dict[str, Any]) -> None: + response = sqs_client.send_message( + QueueUrl=get_settings().ABRI_SQS_URL, + MessageBody=json.dumps(abri_message), + ) + logger.info(f"Sent message to Abri queue. MessageId: {response['MessageId']}") + + def _trigger_pashub_fetcher( sqs_client: Any, deal_id: str, hubspot_deal: Dict[str, str] ) -> None: diff --git a/etl/hubspot/tests/test_abri_flow_triggers.py b/etl/hubspot/tests/test_abri_flow_triggers.py new file mode 100644 index 000000000..c6868b70b --- /dev/null +++ b/etl/hubspot/tests/test_abri_flow_triggers.py @@ -0,0 +1,232 @@ +from datetime import datetime, timezone +from typing import Any, Dict +import uuid + +from backend.app.db.models.hubspot_deal_data import HubspotDealData +from etl.hubspot.abri_flow_triggers import check_for_abri_triggers_and_construct_message + +BASE_TIME = datetime(2025, 12, 1, 12, 0, 0) + +ABRI_PROJECT_CODE = "[Abri Stock Condition - Privately Funded - 062026 - 1247536318670]" + +DEAL_ID = "9876543210" + +LISTING = {"listing_id": "55", "owner_property_id": "1007165", "national_uprn": "1"} + + +def make_old_deal(**overrides: Any) -> HubspotDealData: + return HubspotDealData( + id=overrides.get("id", uuid.uuid4()), + deal_id=DEAL_ID, + created_at=BASE_TIME, + updated_at=BASE_TIME, + **{k: v for k, v in overrides.items() if k != "id"}, + ) + + +def make_new_deal(**overrides: str) -> Dict[str, str]: + return { + "hs_object_id": DEAL_ID, + "dealname": "49 Admers Crescent", + "project_code": ABRI_PROJECT_CODE, + **overrides, + } + + +# ========================== +# SINGLE-FLOW FIRING +# ========================== + + +def test_a_first_confirmed_survey_date_builds_a_log_job_message() -> None: + # Arrange + old_deal = make_old_deal(project_code=ABRI_PROJECT_CODE, confirmed_survey_date=None) + new_deal = make_new_deal( + confirmed_survey_date="2026-06-24", confirmed_survey_time="14:30" + ) + + # Act + message = check_for_abri_triggers_and_construct_message( + hubspot_deal_id=DEAL_ID, + new_deal=new_deal, + new_project=None, + new_listing=LISTING, + old_deal=old_deal, + ) + + # Assert + assert message == { + "hubspot_deal_id": DEAL_ID, + "flows": ["log_job"], + "place_ref": "1007165", + "deal_name": "49 Admers Crescent", + "confirmed_survey_date": "2026-06-24", + "confirmed_survey_time": "14:30", + } + + +def test_a_changed_confirmed_survey_date_builds_an_amend_job_message() -> None: + # Arrange + old_deal = make_old_deal( + project_code=ABRI_PROJECT_CODE, + confirmed_survey_date=datetime(2026, 6, 18, tzinfo=timezone.utc), + ) + new_deal = make_new_deal( + confirmed_survey_date="2026-06-24", confirmed_survey_time="14:30" + ) + + # Act + message = check_for_abri_triggers_and_construct_message( + hubspot_deal_id=DEAL_ID, + new_deal=new_deal, + new_project=None, + new_listing=LISTING, + old_deal=old_deal, + ) + + # Assert + assert message is not None + assert message["flows"] == ["amend_job"] + assert message["confirmed_survey_date"] == "2026-06-24" + + +def test_a_first_expected_commencement_date_builds_a_tenant_sync_message() -> None: + # Arrange + old_deal = make_old_deal( + project_code=ABRI_PROJECT_CODE, expected_commencement_date=None + ) + new_deal = make_new_deal(expected_commencement_date="2026-07-01") + + # Act + message = check_for_abri_triggers_and_construct_message( + hubspot_deal_id=DEAL_ID, + new_deal=new_deal, + new_project=None, + new_listing=LISTING, + old_deal=old_deal, + ) + + # Assert + assert message is not None + assert message["flows"] == ["sync_tenant_data"] + assert message["place_ref"] == "1007165" + + +# ========================== +# ONE MESSAGE PER SCRAPE +# ========================== + + +def test_several_flows_firing_in_one_scrape_share_one_message() -> None: + # Arrange: date changed and commencement date first set together. + old_deal = make_old_deal( + project_code=ABRI_PROJECT_CODE, + confirmed_survey_date=datetime(2026, 6, 18, tzinfo=timezone.utc), + expected_commencement_date=None, + ) + new_deal = make_new_deal( + confirmed_survey_date="2026-06-24", + confirmed_survey_time="14:30", + expected_commencement_date="2026-07-01", + ) + + # Act + message = check_for_abri_triggers_and_construct_message( + hubspot_deal_id=DEAL_ID, + new_deal=new_deal, + new_project=None, + new_listing=LISTING, + old_deal=old_deal, + ) + + # Assert + assert message is not None + assert message["flows"] == ["amend_job", "sync_tenant_data"] + + +# ========================== +# NOTHING FIRES +# ========================== + + +def test_no_message_when_no_abri_trigger_fires() -> None: + # Arrange: something changed, but nothing Abri cares about. + old_deal = make_old_deal(project_code=ABRI_PROJECT_CODE, outcome=None) + new_deal = make_new_deal(outcome="left voicemail") + + # Act + message = check_for_abri_triggers_and_construct_message( + hubspot_deal_id=DEAL_ID, + new_deal=new_deal, + new_project=None, + new_listing=LISTING, + old_deal=old_deal, + ) + + # Assert + assert message is None + + +def test_no_message_for_a_non_abri_deal() -> None: + # Arrange: a survey date first set, but on another project's deal. + old_deal = make_old_deal(project_code="Other", confirmed_survey_date=None) + new_deal = make_new_deal(project_code="Other", confirmed_survey_date="2026-06-24") + + # Act + message = check_for_abri_triggers_and_construct_message( + hubspot_deal_id=DEAL_ID, + new_deal=new_deal, + new_project=None, + new_listing=LISTING, + old_deal=old_deal, + ) + + # Assert + assert message is None + + +def test_abandonment_alone_stays_unwired_and_builds_no_message() -> None: + # Arrange: the abandonment predicate would fire, but CancelJob does not + # exist yet, so no flow may be dispatched for it. + old_deal = make_old_deal( + project_code=ABRI_PROJECT_CODE, number_of_attempts="2", outcome=None + ) + new_deal = make_new_deal(number_of_attempts="3", outcome="no answer") + + # Act + message = check_for_abri_triggers_and_construct_message( + hubspot_deal_id=DEAL_ID, + new_deal=new_deal, + new_project=None, + new_listing=LISTING, + old_deal=old_deal, + ) + + # Assert + assert message is None + + +# ========================== +# PAYLOAD ROBUSTNESS +# ========================== + + +def test_a_missing_listing_still_builds_the_message_without_a_place_ref() -> None: + # Arrange: validation in the abri lambda is the visibility surface for + # a fired flow that lacks its place_ref; the scraper still sends. + old_deal = make_old_deal(project_code=ABRI_PROJECT_CODE, confirmed_survey_date=None) + new_deal = make_new_deal(confirmed_survey_date="2026-06-24") + + # Act + message = check_for_abri_triggers_and_construct_message( + hubspot_deal_id=DEAL_ID, + new_deal=new_deal, + new_project=None, + new_listing=None, + old_deal=old_deal, + ) + + # Assert + assert message is not None + assert message["place_ref"] is None + assert message["confirmed_survey_time"] is None diff --git a/etl/hubspot/tests/test_hubspot_data_to_db.py b/etl/hubspot/tests/test_hubspot_data_to_db.py index 60c937460..8499ccd08 100644 --- a/etl/hubspot/tests/test_hubspot_data_to_db.py +++ b/etl/hubspot/tests/test_hubspot_data_to_db.py @@ -88,7 +88,7 @@ def test_update_existing_deal__no_project__clears_project_id() -> None: def test_update_existing_deal__client_booking_reference_maps_to_job_no() -> None: - existing = HubspotDealData(deal_id="MOCK_DEAL_ID", job_no=None) + existing = HubspotDealData(deal_id="MOCK_DEAL_ID", client_booking_reference=None) deal_data = {"client_booking_reference": "AD0226519"} _make_instance()._update_existing_deal( @@ -98,11 +98,13 @@ def test_update_existing_deal__client_booking_reference_maps_to_job_no() -> None company=None, ) - assert existing.job_no == "AD0226519" + assert existing.client_booking_reference == "AD0226519" def test_update_existing_deal__client_booking_reference_cleared__nulls_job_no() -> None: - existing = HubspotDealData(deal_id="MOCK_DEAL_ID", job_no="AD0226519") + existing = HubspotDealData( + deal_id="MOCK_DEAL_ID", client_booking_reference="AD0226519" + ) deal_data = {} _make_instance()._update_existing_deal( @@ -112,7 +114,7 @@ def test_update_existing_deal__client_booking_reference_cleared__nulls_job_no() company=None, ) - assert existing.job_no is None + assert existing.client_booking_reference is None def test_build_new_deal__client_booking_reference_maps_to_job_no() -> None: @@ -124,4 +126,4 @@ def test_build_new_deal__client_booking_reference_maps_to_job_no() -> None: project=None, ) - assert new_deal.job_no == "AD0226519" + assert new_deal.client_booking_reference == "AD0226519" diff --git a/etl/hubspot/tests/test_hubspot_deal_differ.py b/etl/hubspot/tests/test_hubspot_deal_differ.py index 5e9bcfdfa..2bd15996f 100644 --- a/etl/hubspot/tests/test_hubspot_deal_differ.py +++ b/etl/hubspot/tests/test_hubspot_deal_differ.py @@ -11,6 +11,10 @@ from etl.hubspot.project_data import ProjectData BASE_TIME = datetime(2025, 12, 1, 12, 0, 0) +ABRI_PROJECT_CODE = ( + "[Abri Stock Condition - Privately Funded - 062026 - 1247536318670]" +) + def make_old_deal(**overrides: Any) -> HubspotDealData: return HubspotDealData( @@ -355,12 +359,12 @@ def test_abri_job_logging__date_first_set_on_abri_deal__returns_true() -> None: # Arrange old_deal = make_old_deal( id=deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, confirmed_survey_date=None, ) new_deal = make_new_deal( deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, confirmed_survey_date="2026-07-15", ) @@ -381,12 +385,12 @@ def test_abri_job_logging__date_already_set__returns_false() -> None: # Arrange old_deal = make_old_deal( id=deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, confirmed_survey_date=datetime(2026, 7, 1, tzinfo=timezone.utc), ) new_deal = make_new_deal( deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, confirmed_survey_date="2026-07-15", ) @@ -433,12 +437,41 @@ def test_abri_job_logging__project_code_cased_differently__returns_true() -> Non # Arrange old_deal = make_old_deal( id=deal_id, - project_code="ABRI CONDITION", + project_code=ABRI_PROJECT_CODE.upper(), confirmed_survey_date=None, ) new_deal = make_new_deal( deal_id, - project_code="ABRI CONDITION", + project_code=ABRI_PROJECT_CODE.upper(), + confirmed_survey_date="2026-07-15", + ) + + # Act + result = HubspotDealDiffer.check_for_abri_job_logging( + new_deal=new_deal, + new_project=None, + old_deal=old_deal, + ) + + # Assert + assert result is True + + +def test_abri_job_logging__real_portal_project_code__returns_true() -> None: + # The code carried by every Abri stock-condition deal in the portal + # (verified 2026-07-07); these deals have no project association, so the + # project_code fallback is the branch that must fire in production. + deal_id = uuid.uuid4() + + # Arrange + old_deal = make_old_deal( + id=deal_id, + project_code="[Abri Stock Condition - Privately Funded - 062026 - 1247536318670]", + confirmed_survey_date=None, + ) + new_deal = make_new_deal( + deal_id, + project_code="[Abri Stock Condition - Privately Funded - 062026 - 1247536318670]", confirmed_survey_date="2026-07-15", ) @@ -469,12 +502,12 @@ def test_abri_job_logging__no_parseable_new_date__returns_false( # Arrange old_deal = make_old_deal( id=deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, confirmed_survey_date=None, ) new_deal = make_new_deal( deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, **new_overrides, ) @@ -527,12 +560,12 @@ def test_abri_job_logging__non_abri_project_id_with_abri_code__returns_false() - # Arrange old_deal = make_old_deal( id=deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, confirmed_survey_date=None, ) new_deal = make_new_deal( deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, confirmed_survey_date="2026-07-15", ) new_project = ProjectData(project_id="999999", name="Southern Retrofit") @@ -559,12 +592,12 @@ def test_abri_job_amendment__date_changed_on_abri_deal__returns_true() -> None: # Arrange old_deal = make_old_deal( id=deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, confirmed_survey_date=datetime(2026, 7, 1, tzinfo=timezone.utc), ) new_deal = make_new_deal( deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, confirmed_survey_date="2026-07-15", ) @@ -611,13 +644,13 @@ def test_abri_job_amendment__date_and_time_unchanged__returns_false() -> None: # Arrange old_deal = make_old_deal( id=deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, confirmed_survey_date=datetime(2026, 7, 15, tzinfo=timezone.utc), confirmed_survey_time="09:30", ) new_deal = make_new_deal( deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, confirmed_survey_date="2026-07-15", confirmed_survey_time="09:30", ) @@ -639,13 +672,13 @@ def test_abri_job_amendment__time_changed_on_abri_deal__returns_true() -> None: # Arrange old_deal = make_old_deal( id=deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, confirmed_survey_date=datetime(2026, 7, 15, tzinfo=timezone.utc), confirmed_survey_time="09:30", ) new_deal = make_new_deal( deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, confirmed_survey_date="2026-07-15", confirmed_survey_time="14:00", ) @@ -667,12 +700,12 @@ def test_abri_job_amendment__date_first_set_on_abri_deal__returns_false() -> Non # Arrange old_deal = make_old_deal( id=deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, confirmed_survey_date=None, ) new_deal = make_new_deal( deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, confirmed_survey_date="2026-07-15", ) @@ -700,12 +733,12 @@ def test_abri_tenant_data_fetch__commencement_date_first_set_on_abri_deal__retur # Arrange old_deal = make_old_deal( id=deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, expected_commencement_date=None, ) new_deal = make_new_deal( deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, expected_commencement_date="2026-08-01", ) @@ -752,12 +785,12 @@ def test_abri_tenant_data_fetch__commencement_date_already_set__returns_false() # Arrange old_deal = make_old_deal( id=deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, expected_commencement_date=datetime(2026, 7, 1, tzinfo=timezone.utc), ) new_deal = make_new_deal( deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, expected_commencement_date="2026-08-01", ) @@ -788,12 +821,12 @@ def test_abri_tenant_data_fetch__no_parseable_new_commencement_date__returns_fal # Arrange old_deal = make_old_deal( id=deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, expected_commencement_date=None, ) new_deal = make_new_deal( deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, **new_overrides, ) @@ -846,12 +879,12 @@ def test_abri_tenant_data_fetch__non_abri_project_id_with_abri_code__returns_fal # Arrange old_deal = make_old_deal( id=deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, expected_commencement_date=None, ) new_deal = make_new_deal( deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, expected_commencement_date="2026-08-01", ) new_project = ProjectData(project_id="999999", name="Southern Retrofit") @@ -880,12 +913,12 @@ def test_abri_deal_abandonment__third_attempt_with_negative_outcome__returns_tru # Arrange old_deal = make_old_deal( id=deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, outcome=None, ) new_deal = make_new_deal( deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, number_of_attempts="3", outcome="No Answer", ) @@ -945,12 +978,12 @@ def test_abri_deal_abandonment__fewer_than_three_parseable_attempts__returns_fal # Arrange old_deal = make_old_deal( id=deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, outcome=None, ) new_deal = make_new_deal( deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, outcome="No Answer", **new_overrides, ) @@ -982,12 +1015,12 @@ def test_abri_deal_abandonment__outcome_not_negative__returns_false( # Arrange old_deal = make_old_deal( id=deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, outcome=None, ) new_deal = make_new_deal( deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, number_of_attempts="3", **new_overrides, ) @@ -1022,12 +1055,12 @@ def test_abri_deal_abandonment__each_negative_outcome__returns_true( # Arrange old_deal = make_old_deal( id=deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, outcome=None, ) new_deal = make_new_deal( deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, number_of_attempts="3", outcome=outcome, ) @@ -1049,13 +1082,13 @@ def test_abri_deal_abandonment__deal_already_abandoned__returns_false() -> None: # Arrange old_deal = make_old_deal( id=deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, number_of_attempts="3", outcome="No Answer", ) new_deal = make_new_deal( deal_id, - project_code="Abri Condition", + project_code=ABRI_PROJECT_CODE, number_of_attempts="4", outcome="No Answer", ) @@ -1234,7 +1267,7 @@ def test_db_update_trigger__client_booking_reference_changed__returns_true() -> # Arrange old_deal = make_old_deal( id=deal_id, - job_no=None, + client_booking_reference=None, ) new_deal = make_new_deal( deal_id, @@ -1260,7 +1293,7 @@ def test_db_update_trigger__client_booking_reference_unchanged__returns_false() # Arrange old_deal = make_old_deal( id=deal_id, - job_no="AD0226519", + client_booking_reference="AD0226519", ) new_deal = make_new_deal( deal_id, diff --git a/etl/hubspot/tests/test_scraper_handler.py b/etl/hubspot/tests/test_scraper_handler.py index ec0a90eae..c047153bb 100644 --- a/etl/hubspot/tests/test_scraper_handler.py +++ b/etl/hubspot/tests/test_scraper_handler.py @@ -12,6 +12,8 @@ DEAL_ID = "999" PASHUB_LINK = "https://pashub.example.com/deal/999" MAGICPLAN_QUEUE_URL = "https://sqs.eu-west-2.amazonaws.com/123/magic-plan-dev" PASHUB_QUEUE_URL = "https://sqs.test/pashub" +ABRI_API_QUEUE_URL = "https://sqs.test/abri" +ABRI_PROJECT_CODE = "[Abri Stock Condition - Privately Funded - 062026 - 1247536318670]" PROJECT = {"project_id": "proj-1", "name": "Project One"} @@ -59,6 +61,7 @@ def run_handler( mock_boto3.client.return_value = mock_sqs mock_settings.return_value.MAGICPLAN_SQS_URL = MAGICPLAN_QUEUE_URL mock_settings.return_value.PASHUB_TO_ARA_SQS_URL = PASHUB_QUEUE_URL + mock_settings.return_value.ABRI_SQS_URL = ABRI_API_QUEUE_URL handler.__wrapped__({"hubspot_deal_id": DEAL_ID}, "") @@ -126,7 +129,9 @@ def test_existing_deal_surveyed_transition__sends_magicplan_sqs() -> None: listing = {"national_uprn": UPRN} # Act - mock_sqs, _ = run_handler(hubspot_deal=hubspot_deal, db_deal=db_deal, listing=listing) + mock_sqs, _ = run_handler( + hubspot_deal=hubspot_deal, db_deal=db_deal, listing=listing + ) # Assert mock_sqs.send_message.assert_called_once_with( @@ -229,6 +234,42 @@ def test_existing_deal_pashub_link_unchanged__no_pashub_sqs() -> None: mock_sqs.send_message.assert_not_called() +# ========================================== +# EXISTING DEAL PATH - Abri trigger +# ========================================== + + +def test_existing_abri_deal_first_survey_date__sends_abri_sqs() -> None: + # Arrange + db_deal = make_db_deal(project_code=ABRI_PROJECT_CODE, confirmed_survey_date=None) + hubspot_deal = make_hubspot_deal( + project_code=ABRI_PROJECT_CODE, + confirmed_survey_date="2026-06-24", + confirmed_survey_time="14:30", + ) + listing = {"owner_property_id": "1007165", "national_uprn": UPRN} + + # Act + mock_sqs, _ = run_handler( + hubspot_deal=hubspot_deal, db_deal=db_deal, listing=listing + ) + + # Assert + mock_sqs.send_message.assert_called_once_with( + QueueUrl=ABRI_API_QUEUE_URL, + MessageBody=json.dumps( + { + "hubspot_deal_id": DEAL_ID, + "flows": ["log_job"], + "place_ref": "1007165", + "deal_name": DEAL_NAME, + "confirmed_survey_date": "2026-06-24", + "confirmed_survey_time": "14:30", + } + ), + ) + + # ==================================== # PROJECT upsert # ==================================== diff --git a/orchestration/abri_orchestrator.py b/orchestration/abri_orchestrator.py index 920eaffcf..d4b61b8ba 100644 --- a/orchestration/abri_orchestrator.py +++ b/orchestration/abri_orchestrator.py @@ -5,6 +5,8 @@ from typing import Dict, List, Optional, Protocol, Tuple, Union from domain.abri.descriptions import build_job_descriptions from domain.abri.models import ( AbriRequestRejected, + AmendJobRequest, + AppointmentAmended, LogJobRequest, PlaceRef, Tenant, @@ -123,6 +125,35 @@ LogJobOrchestrationResult = Union[LogJobSummary, AbriRequestRejected] class DealDatabaseGateway(Protocol): def record_job_no(self, deal_id: str, job_no: str) -> None: ... + def job_no_for_deal(self, deal_id: str) -> Optional[str]: ... + + +@dataclass(frozen=True) +class AppointmentChange: + """The deal fields a changed survey booking contributes to an AmendJob.""" + + deal_id: str + confirmed_survey_date: date + confirmed_survey_time: Optional[str] + + +AmendJobOrchestrationResult = Union[AppointmentAmended, AbriRequestRejected] + + +class JobNoNotYetRecordedError(Exception): + """An amendment arrived before the deal's job_no landed in the database. + + Retriable: the message should be redelivered until the log flow's + write-back lands (or it dead-letters), so a rapid log-then-amend + sequence eventually converges. + """ + + def __init__(self, deal_id: str) -> None: + self.deal_id = deal_id + super().__init__( + f"no job_no recorded yet for deal {deal_id}; cannot amend its appointment" + ) + class LogJobWriteBackError(Exception): """A job logged in OpenHousing whose job_no could not be written back. @@ -208,6 +239,21 @@ class AbriOrchestrator: ), ) + def amend_job(self, change: AppointmentChange) -> AmendJobOrchestrationResult: + job_no = self._deal_database.job_no_for_deal(change.deal_id) + if job_no is None: + raise JobNoNotYetRecordedError(deal_id=change.deal_id) + + return self._abri.amend_job( + AmendJobRequest( + job_no=job_no, + appointment_date=change.confirmed_survey_date, + appointment_time=slot_for_confirmed_time( + change.confirmed_survey_time + ), + ) + ) + def log_job(self, booking: ConfirmedSurveyBooking) -> LogJobOrchestrationResult: descriptions = build_job_descriptions(booking.deal_name) outcome = self._abri.log_job( diff --git a/repositories/hubspot_deals/__init__.py b/repositories/hubspot_deals/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/repositories/hubspot_deals/deal_database_postgres_gateway.py b/repositories/hubspot_deals/deal_database_postgres_gateway.py new file mode 100644 index 000000000..5bc960fc3 --- /dev/null +++ b/repositories/hubspot_deals/deal_database_postgres_gateway.py @@ -0,0 +1,34 @@ +"""Postgres implementation of the Abri deal-database gateway. + +First production implementation of the DealDatabaseGateway protocol used by +the AbriOrchestrator (orchestration/abri_orchestrator.py): job numbers live +on the hubspot_deal_data table, in the job_no column the scraper also +upserts from HubSpot's client_booking_reference property. +""" + +from typing import Optional + +from sqlmodel import Session, select + +from backend.app.db.models.hubspot_deal_data import HubspotDealData + + +class DealDatabasePostgresGateway: + def __init__(self, session: Session) -> None: + self._session = session + + def record_job_no(self, deal_id: str, job_no: str) -> None: + deal = self._find_deal(deal_id) + if deal is None: + raise ValueError(f"HubSpot deal {deal_id} not found") + deal.client_booking_reference = job_no + self._session.add(deal) + self._session.commit() + + def job_no_for_deal(self, deal_id: str) -> Optional[str]: + deal = self._find_deal(deal_id) + return deal.client_booking_reference if deal is not None else None + + def _find_deal(self, deal_id: str) -> Optional[HubspotDealData]: + statement = select(HubspotDealData).where(HubspotDealData.deal_id == deal_id) + return self._session.exec(statement).first() diff --git a/scripts/smoke_test_abri_log_amend.py b/scripts/smoke_test_abri_log_amend.py new file mode 100644 index 000000000..cc786f449 --- /dev/null +++ b/scripts/smoke_test_abri_log_amend.py @@ -0,0 +1,145 @@ +"""Manual smoke test for the Abri log -> amend wiring against UAT (issue #1480). + +Drives the real AbriOrchestrator log-job and amend-job flows end-to-end +against Abri's UAT relay endpoint: a job is logged for a test place +reference, its job_no is written back to a throwaway HubSpot deal, and the +appointment is then amended to the following day's other slot — proving the +full log -> write-back -> amend sequence converges. + +The deal-database gateway is in-memory: the write-back to the real deal-data +table belongs to the deployed lambda, and this script must not touch a +production database. + +Usage: + 1. Ensure these are set in the environment or backend/.env: + HUBSPOT_API_KEY, ABRI_RELAY_URL (the UAT endpoint), + ABRI_RELAY_USERNAME, ABRI_RELAY_PASSWORD, ABRI_RELAY_DEFAULT_RESOURCE. + 2. Create a throwaway test deal in the HubSpot UI and paste its id into + DEAL_ID below; paste Abri's agreed UAT test place reference into + PLACE_REF. + 3. Run: python scripts/smoke_test_abri_log_amend.py + 4. Check: the printed job_no matches the deal's client_booking_reference + in the HubSpot UI, and Abri can see the appointment on the amended + date/slot in their TM solution. + 5. Clean up: delete the test deal in the UI and ask Abri to cancel the + UAT job (there is no CancelJob route yet). +""" + +import os +from datetime import date, timedelta +from pathlib import Path +from typing import Dict, List, Optional, Tuple + +from dotenv import dotenv_values +from hubspot.client import Client # type: ignore[reportMissingTypeStubs] + +from domain.abri.models import AbriRequestRejected, PlaceRef +from infrastructure.abri.abri_client import AbriClient +from infrastructure.abri.config import AbriConfig +from infrastructure.hubspot.deal_contacts_client import HubspotDealContactsClient +from infrastructure.hubspot.deal_properties_client import HubspotDealPropertiesClient +from orchestration.abri_orchestrator import ( + AbriOrchestrator, + AppointmentChange, + ConfirmedSurveyBooking, +) + +DEAL_ID = "EDIT-ME" +PLACE_REF = PlaceRef("EDIT-ME") + +# Log for the first weekday at least a week out (a slot Abri UAT will take), +# then amend to the following day's opposite slot. +LOG_DATE = date.today() + timedelta(days=7) +LOG_TIME = "10:00" # AM slot +AMEND_DATE = LOG_DATE + timedelta(days=1) +AMEND_TIME = "14:30" # PM slot + + +def _env(name: str) -> str: + env_file = Path(__file__).parents[1] / "backend" / ".env" + value = os.getenv(name) or dotenv_values(env_file).get(name) + if not value: + raise SystemExit(f"Missing {name} (env var or backend/.env)") + return value + + +def _hubspot_sdk_client() -> Client: + return Client.create(access_token=_env("HUBSPOT_API_KEY")) # type: ignore[reportUnknownMemberType] + + +def _uat_abri_client() -> AbriClient: + return AbriClient( + config=AbriConfig( + endpoint_url=_env("ABRI_RELAY_URL"), + username=_env("ABRI_RELAY_USERNAME"), + password=_env("ABRI_RELAY_PASSWORD"), + default_resource=_env("ABRI_RELAY_DEFAULT_RESOURCE"), + ) + ) + + +class InMemoryDealDatabase: + """Stands in for the deal-data table; the real write-back is the lambda's.""" + + def __init__(self) -> None: + self.recorded: List[Tuple[str, str]] = [] + self._job_nos: Dict[str, str] = {} + + def record_job_no(self, deal_id: str, job_no: str) -> None: + self.recorded.append((deal_id, job_no)) + 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) + + +def main() -> None: + if DEAL_ID == "EDIT-ME" or PLACE_REF == PlaceRef("EDIT-ME"): + raise SystemExit("Edit DEAL_ID and PLACE_REF before running") + + sdk_client = _hubspot_sdk_client() + deal_database = InMemoryDealDatabase() + orchestrator = AbriOrchestrator( + abri_client=_uat_abri_client(), + deal_contacts=HubspotDealContactsClient(sdk_client=sdk_client), + deal_properties=HubspotDealPropertiesClient(sdk_client=sdk_client), + deal_database=deal_database, + ) + + print(f"Logging job for place_ref {PLACE_REF} on {LOG_DATE} at {LOG_TIME}...") + log_result = orchestrator.log_job( + ConfirmedSurveyBooking( + deal_id=DEAL_ID, + place_ref=PLACE_REF, + deal_name=f"Domna UAT smoke test {date.today().isoformat()}", + confirmed_survey_date=LOG_DATE, + confirmed_survey_time=LOG_TIME, + ) + ) + if isinstance(log_result, AbriRequestRejected): + raise SystemExit( + f"LogJob rejected: {log_result.code}: {log_result.message}" + ) + print(f"Job logged: job_no={log_result.job_no} (written back to deal {DEAL_ID})") + + print(f"Amending appointment to {AMEND_DATE} at {AMEND_TIME}...") + amend_result = orchestrator.amend_job( + AppointmentChange( + deal_id=DEAL_ID, + confirmed_survey_date=AMEND_DATE, + confirmed_survey_time=AMEND_TIME, + ) + ) + if isinstance(amend_result, AbriRequestRejected): + raise SystemExit( + f"AmendJob rejected: {amend_result.code}: {amend_result.message}" + ) + print( + f"Appointment amended: job_no={amend_result.job_no} " + f"date={amend_result.appointment_date} slot={amend_result.appointment_time}" + ) + print("Smoke test complete. Remember to clean up (see module docstring).") + + +if __name__ == "__main__": + main() diff --git a/scripts/smoke_test_abri_logjob_flow.py b/scripts/smoke_test_abri_logjob_flow.py new file mode 100644 index 000000000..36f0f40e8 --- /dev/null +++ b/scripts/smoke_test_abri_logjob_flow.py @@ -0,0 +1,271 @@ +"""Manual smoke test for the scraper->dispatch LogJob wiring (issue #1480). + +The closest run to end-to-end available before Abri send API credentials: +every seam is real except the Abri HTTP edge, which is stubbed to return a +canned LogJob success carrying a fake job_no. Concretely, the script + + 1. fetches the deal, listing and project from the real HubSpot portal + (exactly what the scraper lambda sees), + 2. reads the deal's row from the real (dev) deal database, + 3. runs the real differ predicates via abri_message_if_triggered, + 4. validates the real message contract (AbriTriggerRequest), + 5. runs the real dispatch through the real AbriOrchestrator, so the fake + job_no is written to the deal's client_booking_reference property in + HubSpot first, then to the same-named column in the database, + 6. reads both back and prints them. + +NOTE: the LogJob trigger fires when the CONFIRMED SURVEY DATE is first set +on an Abri deal. (Setting the *expected commencement date* fires the +tenant-data flow instead — that flow is skipped here so the stub never +mints junk contacts; smoke_test_tenant_contacts.py covers it.) + +Usage: + 1. Ensure HUBSPOT_API_KEY and the postgres_* connection values are in + backend/.env (they are read tolerantly; the strict Settings .env + parsing is bypassed). + 2. Pick a test deal on the "Abri Condition" project that the scraper has + already ingested (its row must exist in hubspot_deal_data). Paste its + id into DEAL_ID below. + 3. In the HubSpot UI, set the deal's confirmed survey date (and + optionally a time). If the dev scraper lambda syncs the change into + the database before you run this, the differ will see no change — set + RESET_DB_ROW_FIRST = True to null the row's survey date and job_no so + the trigger fires deterministically. + 4. Run: python scripts/smoke_test_abri_logjob_flow.py + 5. Check the printed read-backs, and the deal in the HubSpot UI: the + fake job_no (SMOKE...) should sit in client_booking_reference. + 6. Run it again unchanged: the dispatch should print + "skipped: job SMOKE... already logged" — the idempotency guard. + 7. Clean up: set CLEANUP = True and re-run to clear the fake job_no from + both HubSpot and the database, then unset the deal's survey date. +""" + +import os +import xml.etree.ElementTree as ET +from datetime import datetime +from pathlib import Path +from typing import Any, List, Optional, cast + +from dotenv import dotenv_values + +DEAL_ID = "EDIT-ME" + +# Null the DB row's confirmed_survey_date + job_no before evaluating the +# differ, so the trigger fires even if the dev scraper already synced it. +RESET_DB_ROW_FIRST = False + +# Cleanup mode: clear the fake job_no from HubSpot and the database. +CLEANUP = False + +FAKE_JOB_NO = "SMOKE" + datetime.now().strftime("%d%H%M") + +JOB_NO_DEAL_PROPERTY = "client_booking_reference" + + +def _bootstrap_env() -> None: + """Export backend/.env tolerantly and bypass strict Settings parsing. + + backend/.env carries keys the strict Settings model rejects; exporting + them ourselves and pointing ENVIRONMENT away from "local" lets the + legacy HubspotClient (and PostgresConfig.from_env) run unchanged. + """ + env_file = Path(__file__).parents[1] / "backend" / ".env" + for key, value in dotenv_values(env_file).items(): + if value is not None: + os.environ.setdefault(key.upper(), value) + os.environ["ENVIRONMENT"] = "smoke-test" + + +class _CannedLogJobResponse: + content = ( + f'' + ).encode() + + def raise_for_status(self) -> None: + pass + + +class _StubAbriSession: + """Stands in for requests.Session; records envelopes, returns the canned + LogJob success. Anything but a logjob request is a wiring bug.""" + + def __init__(self) -> None: + self.sent_envelopes: List[bytes] = [] + + def post(self, url: str, data: bytes) -> _CannedLogJobResponse: + request_type = ( + ET.fromstring(data).find("Body/Request").get("request_type") # type: ignore[reportOptionalMemberAccess] + ) + if request_type != "logjob": + raise AssertionError( + f"stub received a {request_type!r} request; only logjob is smoke-tested" + ) + self.sent_envelopes.append(data) + return _CannedLogJobResponse() + + +def _print_sent_parameters(envelope: bytes) -> None: + parameters = { + parameter.get("attribute"): parameter.get("attribute_value") + for parameter in ET.fromstring(envelope).findall("Body/Request/Parameters") + } + print(" Envelope sent to (stubbed) Abri relay:") + for name, value in parameters.items(): + print(f" {name} = {value}") + + +def main() -> None: + if DEAL_ID == "EDIT-ME": + raise SystemExit("Edit DEAL_ID before running") + + _bootstrap_env() + + # Imports that read settings/env must come after the bootstrap. + from hubspot.client import Client # type: ignore[reportMissingTypeStubs] + from sqlalchemy.pool import NullPool + from sqlmodel import Session, select + + from applications.abri.abri_trigger_request import AbriTriggerRequest + from applications.abri.dispatch import dispatch_abri_flows + from backend.app.db.models.hubspot_deal_data import HubspotDealData + from etl.hubspot.abri_flow_triggers import ( + check_for_abri_triggers_and_construct_message, + ) + from etl.hubspot.hubspotClient import HubspotClient + from infrastructure.abri.abri_client import AbriClient + from infrastructure.abri.config import AbriConfig + from infrastructure.hubspot.deal_contacts_client import HubspotDealContactsClient + from infrastructure.hubspot.deal_properties_client import ( + HubspotDealPropertiesClient, + ) + from infrastructure.postgres.config import PostgresConfig + from infrastructure.postgres.engine import make_engine + from orchestration.abri_orchestrator import AbriOrchestrator + from repositories.hubspot_deals.deal_database_postgres_gateway import ( + DealDatabasePostgresGateway, + ) + + engine = make_engine(PostgresConfig.from_env(dict(os.environ)), poolclass=NullPool) + sdk_client: Client = Client.create( # type: ignore[reportUnknownMemberType] + access_token=os.environ["HUBSPOT_API_KEY"] + ) + deal_properties = HubspotDealPropertiesClient(sdk_client=sdk_client) + + with Session(engine) as session: + deal_row = session.exec( + select(HubspotDealData).where(HubspotDealData.deal_id == DEAL_ID) + ).first() + if deal_row is None: + raise SystemExit( + f"Deal {DEAL_ID} has no hubspot_deal_data row — let the scraper " + "ingest it first (the gateway write-back needs the row)." + ) + + if CLEANUP: + deal_properties.write_deal_property( + deal_id=DEAL_ID, property_name=JOB_NO_DEAL_PROPERTY, value="" + ) + deal_row.client_booking_reference = None + session.add(deal_row) + session.commit() + print( + f"Cleaned up: cleared {JOB_NO_DEAL_PROPERTY} in HubSpot and " + "job_no in the database. Unset the survey date in the UI too." + ) + return + + if RESET_DB_ROW_FIRST: + deal_row.confirmed_survey_date = None + deal_row.client_booking_reference = None + session.add(deal_row) + session.commit() + session.refresh(deal_row) + print("Reset DB row: confirmed_survey_date and job_no nulled.") + + print(f"Fetching deal {DEAL_ID} from HubSpot...") + hubspot_deal, _company, listing, project = ( + HubspotClient().get_deal_and_company_and_listing_and_project(DEAL_ID) + ) + print( + f" dealname={hubspot_deal.get('dealname')!r} " + f"project_code={hubspot_deal.get('project_code')!r} " + f"confirmed_survey_date={hubspot_deal.get('confirmed_survey_date')!r} " + f"confirmed_survey_time={hubspot_deal.get('confirmed_survey_time')!r}" + ) + print( + f" DB row: confirmed_survey_date={deal_row.confirmed_survey_date} " + f"job_no={deal_row.client_booking_reference}" + ) + + message = check_for_abri_triggers_and_construct_message( + hubspot_deal_id=DEAL_ID, + new_deal=hubspot_deal, + new_project=project, + new_listing=listing, + old_deal=deal_row, + ) + if message is None: + raise SystemExit( + "No Abri trigger fired. For LogJob, set the CONFIRMED SURVEY " + "DATE on the deal (expected commencement date fires the " + "tenant-data flow, not job logging). If the dev scraper " + "already synced your change, set RESET_DB_ROW_FIRST = True." + ) + print(f"Trigger message built: flows={message['flows']}") + + request = AbriTriggerRequest.model_validate(message) + if "log_job" not in request.flows: + raise SystemExit( + f"Fired flows {request.flows} do not include log_job — set the " + "confirmed survey date (not the expected commencement date)." + ) + if request.flows != ["log_job"]: + skipped = [flow for flow in request.flows if flow != "log_job"] + print(f" Restricting to log_job for this smoke; skipping {skipped}.") + request = request.model_copy(update={"flows": ["log_job"]}) + + stub_session = _StubAbriSession() + abri_client = AbriClient( + config=AbriConfig( + endpoint_url="https://stubbed.invalid/relay", + username="smoke-test", + password="", + default_resource="SMOKE", + ) + ) + abri_client._session = cast( # pyright: ignore[reportPrivateUsage] + Any, stub_session + ) + gateway = DealDatabasePostgresGateway(session=session) + orchestrator = AbriOrchestrator( + abri_client=abri_client, + deal_contacts=HubspotDealContactsClient(sdk_client=sdk_client), + deal_properties=deal_properties, + deal_database=gateway, + ) + + print("Dispatching...") + summary = dispatch_abri_flows(request, orchestrator, gateway) + for envelope in stub_session.sent_envelopes: + _print_sent_parameters(envelope) + print(f"Dispatch summary: {summary}") + + # Read both write-backs back through the real interfaces. + deal = sdk_client.crm.deals.basic_api.get_by_id( # type: ignore[reportUnknownMemberType] + DEAL_ID, properties=[JOB_NO_DEAL_PROPERTY] + ) + hubspot_value = cast( + Optional[str], + cast(Any, deal).properties.get(JOB_NO_DEAL_PROPERTY), + ) + print(f"HubSpot {JOB_NO_DEAL_PROPERTY} is now: {hubspot_value!r}") + print(f"Database job_no is now: {gateway.job_no_for_deal(DEAL_ID)!r}") + print( + "Re-run to see the idempotency guard skip the log; " + "set CLEANUP = True to undo." + ) + + +if __name__ == "__main__": + main() diff --git a/scripts/smoke_test_tenant_contacts.py b/scripts/smoke_test_tenant_contacts.py index f9b55dc3b..8dedeedac 100644 --- a/scripts/smoke_test_tenant_contacts.py +++ b/scripts/smoke_test_tenant_contacts.py @@ -22,7 +22,7 @@ All tenant details below are fictional (Ofcom-reserved number ranges). import os from pathlib import Path -from typing import Any, List, cast +from typing import Any, List, Optional, cast from dotenv import dotenv_values from hubspot.client import Client # type: ignore[reportMissingTypeStubs] @@ -101,11 +101,14 @@ def _stubbed_abri_client() -> AbriClient: class _UnusedDealDatabase: - """Satisfies the gateway dependency; the tenant-data flow never writes it.""" + """Satisfies the gateway dependency; the tenant-data flow never touches it.""" def record_job_no(self, deal_id: str, job_no: str) -> None: raise AssertionError("tenant-data sync must not touch the deal database") + def job_no_for_deal(self, deal_id: str) -> Optional[str]: + raise AssertionError("tenant-data sync must not touch the deal database") + def _archive_contacts(sdk_client: Client) -> None: if not CONTACT_IDS_TO_ARCHIVE: diff --git a/tests/applications/abri/__init__.py b/tests/applications/abri/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/applications/abri/test_abri_trigger_request.py b/tests/applications/abri/test_abri_trigger_request.py new file mode 100644 index 000000000..9e9672a2b --- /dev/null +++ b/tests/applications/abri/test_abri_trigger_request.py @@ -0,0 +1,103 @@ +from datetime import date +from typing import Any, Dict + +import pytest +from pydantic import ValidationError + +from applications.abri.abri_trigger_request import AbriTriggerRequest + + +def _full_message() -> Dict[str, Any]: + return { + "hubspot_deal_id": "9876543210", + "flows": ["amend_job", "log_job", "sync_tenant_data"], + "place_ref": "1007165", + "deal_name": "49 Admers Crescent", + "confirmed_survey_date": "2026-06-24", + "confirmed_survey_time": "14:30", + } + + +def test_a_full_message_parses_into_a_typed_request() -> None: + # Act + request = AbriTriggerRequest.model_validate(_full_message()) + + # Assert + assert request.hubspot_deal_id == "9876543210" + assert request.flows == ["amend_job", "log_job", "sync_tenant_data"] + assert request.place_ref == "1007165" + assert request.deal_name == "49 Admers Crescent" + assert request.confirmed_survey_date == date(2026, 6, 24) + assert request.confirmed_survey_time == "14:30" + + +@pytest.mark.parametrize( + ("flow", "missing_field"), + [ + ("log_job", "place_ref"), + ("log_job", "deal_name"), + ("log_job", "confirmed_survey_date"), + ("amend_job", "confirmed_survey_date"), + ("sync_tenant_data", "place_ref"), + ], +) +def test_a_message_missing_a_field_its_flow_needs_fails_validation( + flow: str, missing_field: str +) -> None: + # Arrange + message = _full_message() + message["flows"] = [flow] + del message[missing_field] + + # Act / Assert + with pytest.raises(ValidationError, match=missing_field): + AbriTriggerRequest.model_validate(message) + + +def test_a_message_missing_only_fields_of_unfired_flows_parses() -> None: + # Arrange: tenant sync needs only the place_ref (and the deal id). + message = { + "hubspot_deal_id": "9876543210", + "flows": ["sync_tenant_data"], + "place_ref": "1007165", + } + + # Act + request = AbriTriggerRequest.model_validate(message) + + # Assert + assert request.flows == ["sync_tenant_data"] + assert request.confirmed_survey_date is None + + +def test_a_message_without_a_confirmed_time_parses_for_logging() -> None: + # Arrange: the time slot is optional — no time means an all-day slot. + message = _full_message() + message["flows"] = ["log_job"] + del message["confirmed_survey_time"] + + # Act + request = AbriTriggerRequest.model_validate(message) + + # Assert + assert request.confirmed_survey_time is None + + +def test_a_message_naming_an_unknown_flow_fails_validation() -> None: + # Arrange + message = _full_message() + message["flows"] = ["cancel_job"] + + # Act / Assert + with pytest.raises(ValidationError): + AbriTriggerRequest.model_validate(message) + + +def test_a_message_naming_no_flows_fails_validation() -> None: + # Arrange + message = _full_message() + message["flows"] = [] + + # Act / Assert + with pytest.raises(ValidationError): + AbriTriggerRequest.model_validate(message) 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)", + } diff --git a/tests/orchestration/test_abri_orchestrator_amend_job.py b/tests/orchestration/test_abri_orchestrator_amend_job.py new file mode 100644 index 000000000..f5ba20e57 --- /dev/null +++ b/tests/orchestration/test_abri_orchestrator_amend_job.py @@ -0,0 +1,178 @@ +import xml.etree.ElementTree as ET +from datetime import date +from pathlib import Path +from typing import Dict, List, Optional, Tuple +from unittest.mock import MagicMock, patch + +import pytest + +from domain.abri.models import AbriRequestRejected, AppointmentAmended +from infrastructure.abri.abri_client import AbriClient +from infrastructure.abri.config import AbriConfig +from infrastructure.hubspot.deal_contacts_client import HubspotDealContactsClient +from infrastructure.hubspot.deal_properties_client import HubspotDealPropertiesClient +from orchestration.abri_orchestrator import ( + AbriOrchestrator, + AppointmentChange, + JobNoNotYetRecordedError, +) + +FIXTURE_DIR = Path(__file__).parents[1] / "abri" +ENDPOINT_URL = "https://relay.example.test/api/DomnaRelay?code=test-function-key" +DEAL_ID = "9876543210" +JOB_NO = "AC0439951" + +CONFIG = AbriConfig( + endpoint_url=ENDPOINT_URL, + username="DomnaWeb", + password="", + default_resource="NAULKH", +) + +CHANGE = AppointmentChange( + deal_id=DEAL_ID, + confirmed_survey_date=date(2026, 6, 24), + confirmed_survey_time="14:30", +) + + +def _load_fixture(name: str) -> bytes: + return (FIXTURE_DIR / name).read_bytes() + + +class FakeDealDatabase: + """In-memory stand-in for the deal-database gateway.""" + + def __init__(self) -> None: + self.recorded_job_nos: List[Tuple[str, str]] = [] + self.job_nos: Dict[str, str] = {} + + def record_job_no(self, deal_id: str, job_no: str) -> None: + self.recorded_job_nos.append((deal_id, job_no)) + 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 mock_session() -> MagicMock: + return MagicMock() + + +@pytest.fixture() +def deal_database() -> FakeDealDatabase: + return FakeDealDatabase() + + +@pytest.fixture() +def orchestrator( + mock_session: MagicMock, + deal_database: FakeDealDatabase, +) -> AbriOrchestrator: + sdk_client = MagicMock() + with patch( + "infrastructure.abri.abri_client.requests.Session", + return_value=mock_session, + ): + abri_client = AbriClient(config=CONFIG) + return AbriOrchestrator( + abri_client=abri_client, + deal_contacts=HubspotDealContactsClient(sdk_client=sdk_client), + deal_properties=HubspotDealPropertiesClient(sdk_client=sdk_client), + deal_database=deal_database, + ) + + +# --- outbound request: the AmendOptiAppt envelope targets the recorded job --- + + +def test_amend_job_sends_an_amendoptiappt_envelope_for_the_deals_recorded_job( + orchestrator: AbriOrchestrator, + mock_session: MagicMock, + deal_database: FakeDealDatabase, +) -> None: + # Arrange + deal_database.job_nos[DEAL_ID] = JOB_NO + mock_session.post.return_value.content = _load_fixture( + "abri_relay_amendoptiappt_success_response.xml" + ) + + # Act + orchestrator.amend_job(CHANGE) + + # Assert + (url,) = mock_session.post.call_args.args + sent_body: bytes = mock_session.post.call_args.kwargs["data"] + sent_parameters = { + parameter.get("attribute"): parameter.get("attribute_value") + for parameter in ET.fromstring(sent_body).findall("Body/Request/Parameters") + } + assert url == ENDPOINT_URL + assert sent_parameters == { + "job_no": JOB_NO, + "action": "amend", + "appointment_date": "24/06/2026", + "appointment_time": "PM", + } + + +# --- happy path: the amended appointment is confirmed back --- + + +def test_amend_job_returns_the_amended_appointment( + orchestrator: AbriOrchestrator, + mock_session: MagicMock, + deal_database: FakeDealDatabase, +) -> None: + # Arrange + deal_database.job_nos[DEAL_ID] = JOB_NO + mock_session.post.return_value.content = _load_fixture( + "abri_relay_amendoptiappt_success_response.xml" + ) + + # Act + result = orchestrator.amend_job(CHANGE) + + # Assert + assert result == AppointmentAmended( + job_no=JOB_NO, + appointment_date="24/06/2026", + appointment_time="PM", + ) + + +# --- ordering race: amendment before the job_no write-back has landed --- + + +def test_amend_job_raises_a_retriable_error_when_no_job_no_has_been_recorded( + orchestrator: AbriOrchestrator, + mock_session: MagicMock, +) -> None: + # Act / Assert + with pytest.raises(JobNoNotYetRecordedError): + orchestrator.amend_job(CHANGE) + assert mock_session.post.call_count == 0 + + +# --- rejection passthrough: OpenHousing's actual reason, verbatim --- + + +def test_amend_job_returns_the_openhousing_rejection_verbatim( + orchestrator: AbriOrchestrator, + mock_session: MagicMock, + deal_database: FakeDealDatabase, +) -> None: + # Arrange + deal_database.job_nos[DEAL_ID] = JOB_NO + mock_session.post.return_value.content = _load_fixture( + "abri_relay_amendoptiappt_relayerror_response.xml" + ) + + # Act + result = orchestrator.amend_job(CHANGE) + + # Assert + assert isinstance(result, AbriRequestRejected) + assert result.code == "RelayError" + assert result.message.startswith("Job_no 1019905 is invalid") diff --git a/tests/orchestration/test_abri_orchestrator_log_job.py b/tests/orchestration/test_abri_orchestrator_log_job.py index b91739c52..c47a7fcd2 100644 --- a/tests/orchestration/test_abri_orchestrator_log_job.py +++ b/tests/orchestration/test_abri_orchestrator_log_job.py @@ -3,7 +3,7 @@ import traceback import xml.etree.ElementTree as ET from datetime import date from pathlib import Path -from typing import List, Tuple +from typing import List, Optional, Tuple from unittest.mock import MagicMock, patch import pytest @@ -57,6 +57,16 @@ class FakeDealDatabase: def record_job_no(self, deal_id: str, job_no: str) -> None: self.recorded_job_nos.append((deal_id, job_no)) + def job_no_for_deal(self, deal_id: str) -> Optional[str]: + return next( + ( + job_no + for recorded_deal_id, job_no in self.recorded_job_nos + if recorded_deal_id == deal_id + ), + None, + ) + @pytest.fixture() def mock_session() -> MagicMock: diff --git a/tests/orchestration/test_abri_orchestrator_tenant_data_sync.py b/tests/orchestration/test_abri_orchestrator_tenant_data_sync.py index e45e48686..58b82dd82 100644 --- a/tests/orchestration/test_abri_orchestrator_tenant_data_sync.py +++ b/tests/orchestration/test_abri_orchestrator_tenant_data_sync.py @@ -2,6 +2,7 @@ import json import traceback import xml.etree.ElementTree as ET from pathlib import Path +from typing import Optional from unittest.mock import MagicMock, patch import pytest @@ -66,6 +67,9 @@ class FakeDealDatabase: def record_job_no(self, deal_id: str, job_no: str) -> None: raise AssertionError("tenant-data sync must not touch the deal database") + def job_no_for_deal(self, deal_id: str) -> Optional[str]: + raise AssertionError("tenant-data sync must not touch the deal database") + @pytest.fixture() def orchestrator( diff --git a/tests/repositories/hubspot_deals/__init__.py b/tests/repositories/hubspot_deals/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/repositories/hubspot_deals/test_deal_database_postgres_gateway.py b/tests/repositories/hubspot_deals/test_deal_database_postgres_gateway.py new file mode 100644 index 000000000..62b833686 --- /dev/null +++ b/tests/repositories/hubspot_deals/test_deal_database_postgres_gateway.py @@ -0,0 +1,115 @@ +from collections.abc import Iterator + +import pytest +from sqlalchemy import Engine +from sqlmodel import Session, text + +from backend.app.db.models.hubspot_deal_data import HubspotDealData +from repositories.hubspot_deals.deal_database_postgres_gateway import ( + DealDatabasePostgresGateway, +) + +DEAL_ID = "9876543210" + + +@pytest.fixture +def session(db_engine: Engine) -> Iterator[Session]: + with Session(db_engine) as s: + yield s + + +def _insert_deal(session: Session, deal_id: str) -> HubspotDealData: + deal = HubspotDealData(deal_id=deal_id, dealname="49 Admers Crescent") + session.add(deal) + session.commit() + session.refresh(deal) + return deal + + +def test_record_job_no_persists_the_job_no_on_the_deal_row(session: Session) -> None: + # Arrange + deal = _insert_deal(session, DEAL_ID) + gateway = DealDatabasePostgresGateway(session=session) + + # Act + gateway.record_job_no(deal_id=DEAL_ID, job_no="AD0226519") + + # Assert + session.refresh(deal) + assert deal.client_booking_reference == "AD0226519" + + +def test_record_job_no_for_an_unknown_deal_raises(session: Session) -> None: + # Arrange + gateway = DealDatabasePostgresGateway(session=session) + + # Act / Assert + with pytest.raises(ValueError): + gateway.record_job_no(deal_id="0000000000", job_no="AD0226519") + + +def test_the_job_no_is_stored_in_the_client_booking_reference_column( + session: Session, +) -> None: + # The physical column matches the HubSpot property the job number is + # stored under (client_booking_reference); only the domain calls it + # job_no. The deployed table is migrated by the frontend repo, so the + # column name here must match what that migration created. + # Arrange + _insert_deal(session, DEAL_ID) + gateway = DealDatabasePostgresGateway(session=session) + + # Act + gateway.record_job_no(deal_id=DEAL_ID, job_no="AD0226519") + + # Assert + stored = ( + session.connection() + .execute( + text( + "select client_booking_reference from hubspot_deal_data " + "where deal_id = :deal_id" + ), + {"deal_id": DEAL_ID}, + ) + .one() + ) + assert stored[0] == "AD0226519" + + +def test_job_no_for_deal_returns_the_recorded_job_no(session: Session) -> None: + # Arrange + _insert_deal(session, DEAL_ID) + gateway = DealDatabasePostgresGateway(session=session) + gateway.record_job_no(deal_id=DEAL_ID, job_no="AD0226519") + + # Act + job_no = gateway.job_no_for_deal(DEAL_ID) + + # Assert + assert job_no == "AD0226519" + + +def test_job_no_for_deal_returns_none_before_any_job_is_recorded( + session: Session, +) -> None: + # Arrange + _insert_deal(session, DEAL_ID) + gateway = DealDatabasePostgresGateway(session=session) + + # Act + job_no = gateway.job_no_for_deal(DEAL_ID) + + # Assert + assert job_no is None + + +def test_job_no_for_deal_returns_none_for_an_unknown_deal(session: Session) -> None: + # Arrange + gateway = DealDatabasePostgresGateway(session=session) + + # Act + job_no = gateway.job_no_for_deal("0000000000") + + # Assert + assert job_no is None diff --git a/tests/utilities/aws_lambda/test_task_handler.py b/tests/utilities/aws_lambda/test_task_handler.py index 2bacf38e0..5ab28a2ff 100644 --- a/tests/utilities/aws_lambda/test_task_handler.py +++ b/tests/utilities/aws_lambda/test_task_handler.py @@ -8,12 +8,13 @@ import pytest from sqlalchemy import Engine, text from sqlmodel import Session +from domain.tasks.subtasks import SubTaskStatus from domain.tasks.subtasks import SubTaskFailure from domain.tasks.tasks import Source from orchestration.task_orchestrator import TaskOrchestrator from repositories.tasks.subtask_postgres_repository import SubTaskPostgresRepository from repositories.tasks.task_postgres_repository import TaskPostgresRepository -from utilities.aws_lambda.task_handler import task_handler +from utilities.aws_lambda.task_handler import NonRetriableTaskError, task_handler @dataclass @@ -48,12 +49,8 @@ def test_task_handler_records_cloudwatch_url_on_subtask( ) -> None: # arrange monkeypatch.setenv("AWS_REGION", "eu-west-2") - monkeypatch.setenv( - "AWS_LAMBDA_LOG_GROUP_NAME", "/aws/lambda/modelling-e2e" - ) - monkeypatch.setenv( - "AWS_LAMBDA_LOG_STREAM_NAME", "2026/05/20/[$LATEST]abc123" - ) + monkeypatch.setenv("AWS_LAMBDA_LOG_GROUP_NAME", "/aws/lambda/modelling-e2e") + monkeypatch.setenv("AWS_LAMBDA_LOG_STREAM_NAME", "2026/05/20/[$LATEST]abc123") @task_handler( task_source="modelling_e2e", @@ -91,7 +88,10 @@ def test_task_handler_passes_orchestrator_and_task_id_when_flag_is_true( pass_task_orchestrator=True, ) def _handler( - body: dict[str, Any], context: Any, orchestrator: TaskOrchestrator, task_id: UUID + body: dict[str, Any], + context: Any, + orchestrator: TaskOrchestrator, + task_id: UUID, ) -> None: received.append((orchestrator, task_id)) @@ -105,6 +105,54 @@ def test_task_handler_passes_orchestrator_and_task_id_when_flag_is_true( assert recv_task_id == UUID(result[0]["task_id"]) +def test_task_handler_reports_an_ordinarily_failing_record_for_redelivery( + harness: Harness, +) -> None: + # Arrange + @task_handler( + task_source="abri_api", + source=Source.HUBSPOT_DEAL, + orchestrator_cm=harness.factory, + ) + def handler(body: dict[str, Any], context: Any) -> None: + raise RuntimeError("transient failure") + + event = {"Records": [{"messageId": "msg-1", "body": '{"hubspot_deal_id": "123"}'}]} + + # Act + result = handler(event, context=None) + + # Assert + assert result["batchItemFailures"] == [{"itemIdentifier": "msg-1"}] + subtask_id = result["tasks"][0]["subtask_id"] + assert harness.subtasks.get(UUID(subtask_id)).status is SubTaskStatus.FAILED + + +def test_task_handler_does_not_requeue_a_record_failing_non_retriably( + harness: Harness, +) -> None: + # Arrange + @task_handler( + task_source="abri_api", + source=Source.HUBSPOT_DEAL, + orchestrator_cm=harness.factory, + ) + def handler(body: dict[str, Any], context: Any) -> None: + raise NonRetriableTaskError("job logged but write-back failed") + + event = {"Records": [{"messageId": "msg-1", "body": '{"hubspot_deal_id": "123"}'}]} + + # Act + result = handler(event, context=None) + + # Assert: the failure is recorded, but the message is not requeued. + assert result["batchItemFailures"] == [] + subtask_id = result["tasks"][0]["subtask_id"] + failed = harness.subtasks.get(UUID(subtask_id)) + assert failed.status is SubTaskStatus.FAILED + assert failed.outputs == {"error": "job logged but write-back failed"} + + def test_task_handler_attaches_to_a_supplied_task_and_subtask( harness: Harness, db_engine: Engine ) -> None: @@ -124,7 +172,10 @@ def test_task_handler_attaches_to_a_supplied_task_and_subtask( pass_task_orchestrator=True, ) def handler( - body: dict[str, Any], context: Any, orchestrator: TaskOrchestrator, task_id: UUID + body: dict[str, Any], + context: Any, + orchestrator: TaskOrchestrator, + task_id: UUID, ) -> None: received.append(task_id) @@ -145,9 +196,7 @@ def test_task_handler_attaches_to_a_supplied_task_and_subtask( # assert — ran under the supplied ids, sub_task completed, nothing created assert received == [task.id] - assert result["tasks"] == [ - {"task_id": str(task.id), "subtask_id": str(subtask.id)} - ] + assert result["tasks"] == [{"task_id": str(task.id), "subtask_id": str(subtask.id)}] assert harness.subtasks.get(subtask.id).status.value == "complete" with db_engine.connect() as conn: assert conn.execute(text("SELECT count(*) FROM tasks")).scalar_one() == 1 diff --git a/utilities/aws_lambda/task_handler.py b/utilities/aws_lambda/task_handler.py index de191ffb9..de420eb12 100644 --- a/utilities/aws_lambda/task_handler.py +++ b/utilities/aws_lambda/task_handler.py @@ -22,6 +22,17 @@ logger = logging.getLogger(__name__) OrchestratorCM = Callable[[], AbstractContextManager[TaskOrchestrator]] +class NonRetriableTaskError(Exception): + """A failure that must not be retried by SQS redelivery. + + The SubTask is marked failed (the failure record is the visibility + surface) but the message is not returned to the queue, so the work is + never re-run. Raise it for failures where a retry could duplicate an + irreversible side effect — e.g. a job already logged in OpenHousing + whose HubSpot write-back failed. + """ + + def task_handler( *, task_source: str, @@ -57,6 +68,7 @@ def task_handler( # app-owned task_id and its distributor-pre-created # subtask_id — run under those; create nothing. supplied = _supplied_ids(body) + source_id: Optional[str] = None if supplied is not None: task_id, subtask_id = supplied else: @@ -88,6 +100,17 @@ def task_handler( work=lambda: func(body, context), cloud_logs_url=cloud_logs_url, ) + except NonRetriableTaskError: + # Failed, but never redelivered: the SubTask's failure + # record is the only follow-up surface. + logger.exception( + "subtask failed non-retriably " + "(task_source=%s source_id=%s)", + task_source, + source_id, + ) + if "Records" not in event: + raise except SubTaskFailure as recorded: # A recorded failure (ADR-0055): run_subtask has already # failed the SubTask with the structured details — that