junte/Model
1
0
Fork 0
mirror of https://github.com/Hestia-Homes/Model.git synced 2026-07-12 13:29:04 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #1479 from Hestia-Homes/feature/abri-logjob-orchestration

Browse source 
Abri LogJob orchestration: persist OpenHousing job_no to HubSpot and the deal database
This commit is contained in:
Daniel Roth 2026-07-07 09:30:31 +01:00 • committed by GitHub
commit 3963ea54e9
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
13 changed files with 813 additions and 230 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
backend/app/db/models/hubspot_deal_data.py
View file

@ -71,6 +71,8 @@ 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)
survey_type: Optional[str] = Field(default=None)
measures_for_pibi_ordered: Optional[str] = Field(default=None)

2
etl/hubspot/hubspotDataTodB.py
View file

@ -274,6 +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"),
"surveyed_date": parse_hs_date(deal_data.get("surveyed_date")),
"design_type": deal_data.get("design_type"),
"survey_type": deal_data.get("survey_type"),
@ -382,6 +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"),
surveyed_date=parse_hs_date(deal_data.get("surveyed_date")),
design_type=deal_data.get("design_type"),
survey_type=deal_data.get("survey_type"),

1
etl/hubspot/hubspot_deal_differ.py
View file

@ -97,6 +97,7 @@ class HubspotDealDiffer:
"design_type": "design_type",
"surveyor": "surveyor",
"confirmed_survey_time": "confirmed_survey_time",
"client_booking_reference": "job_no",
"survey_type": "survey_type",
"measures_for_pibi_ordered": "measures_for_pibi_ordered",
"property_halted_reason": "property_halted_reason",

40
etl/hubspot/tests/test_hubspot_data_to_db.py
View file

@ -85,3 +85,43 @@ def test_update_existing_deal__no_project__clears_project_id() -> None:
)
assert existing.project_id is None
def test_update_existing_deal__client_booking_reference_maps_to_job_no() -> None:
existing = HubspotDealData(deal_id="MOCK_DEAL_ID", job_no=None)
deal_data = {"client_booking_reference": "AD0226519"}
_make_instance()._update_existing_deal(
existing=existing,
deal_data=deal_data,
listing=None,
company=None,
)
assert existing.job_no == "AD0226519"
def test_update_existing_deal__client_booking_reference_cleared__nulls_job_no() -> None:
existing = HubspotDealData(deal_id="MOCK_DEAL_ID", job_no="AD0226519")
deal_data = {}
_make_instance()._update_existing_deal(
existing=existing,
deal_data=deal_data,
listing=None,
company=None,
)
assert existing.job_no is None
def test_build_new_deal__client_booking_reference_maps_to_job_no() -> None:
new_deal = _make_instance()._build_new_deal(
deal_id="MOCK_DEAL_ID",
deal_data={"client_booking_reference": "AD0226519"},
listing=None,
company=None,
project=None,
)
assert new_deal.job_no == "AD0226519"

52
etl/hubspot/tests/test_hubspot_deal_differ.py
View file

@ -1226,3 +1226,55 @@ def test_db_update_trigger__listing_changed__returns_true() -> None:
)
assert result is True
def test_db_update_trigger__client_booking_reference_changed__returns_true() -> None:
deal_id = uuid.uuid4()
# Arrange
old_deal = make_old_deal(
id=deal_id,
job_no=None,
)
new_deal = make_new_deal(
deal_id,
hs_object_id="1",
client_booking_reference="AD0226519",
)
# Act
result = HubspotDealDiffer.check_for_db_update_trigger(
new_deal=new_deal,
new_company=None,
new_listing=None,
old_deal=old_deal,
)
# Assert
assert result is True
def test_db_update_trigger__client_booking_reference_unchanged__returns_false() -> None:
deal_id = uuid.uuid4()
# Arrange
old_deal = make_old_deal(
id=deal_id,
job_no="AD0226519",
)
new_deal = make_new_deal(
deal_id,
hs_object_id="1",
client_booking_reference="AD0226519",
)
# Act
result = HubspotDealDiffer.check_for_db_update_trigger(
new_deal=new_deal,
new_company=None,
new_listing=None,
old_deal=old_deal,
)
# Assert
assert result is False

42
infrastructure/hubspot/deal_contacts_client.py
View file

@ -1,46 +1,15 @@
import json
from typing import Any, Callable, Dict, Optional, cast
from typing import Any, Callable, Dict
from hubspot.client import Client # type: ignore[reportMissingTypeStubs]
from hubspot.crm.associations.v4 import AssociationSpec # type: ignore[reportMissingTypeStubs]
from hubspot.crm.contacts import SimplePublicObjectInputForCreate # type: ignore[reportMissingTypeStubs]
from infrastructure.hubspot.errors import HubspotRequestError
from infrastructure.hubspot.retry import call_with_retry
from infrastructure.hubspot.scrubbed_calls import scrubbed_call_with_retry
# HubSpot-defined deal -> contact association ("deal_to_contact" in v3 terms).
DEAL_TO_CONTACT_ASSOCIATION_TYPE_ID = 3
def _scrubbed(error: Exception) -> HubspotRequestError:
"""Reduce an SDK error to non-PII diagnostics, dropping the body.
Each HubSpot sub-module ships its own ApiException class with no shared
base beyond Exception, so the fields are read by duck-typing.
"""
status = getattr(error, "status", None)
status_code = status if isinstance(status, int) else None
category: Optional[str] = None
correlation_id: Optional[str] = None
try:
parsed: Any = json.loads(getattr(error, "body", None) or "")
if isinstance(parsed, dict):
details = cast(Dict[str, Any], parsed)
raw_category = details.get("category")
raw_correlation_id = details.get("correlationId")
category = raw_category if isinstance(raw_category, str) else None
correlation_id = (
raw_correlation_id if isinstance(raw_correlation_id, str) else None
)
except ValueError:
pass
return HubspotRequestError(
status_code=status_code, category=category, correlation_id=correlation_id
)
class HubspotDealContactsClient:
"""Focused HubSpot client for the tenant-data sync flow.
@ -81,9 +50,4 @@ class HubspotDealContactsClient:
@staticmethod
def _call(fn: Callable[[], Any]) -> Any:
"""Run one SDK call under the shared retry policy, scrubbed on failure."""
try:
return call_with_retry(fn)
except Exception as error:
# `from None` keeps the body-carrying SDK error out of tracebacks.
raise _scrubbed(error) from None
return scrubbed_call_with_retry(fn)

28
infrastructure/hubspot/deal_properties_client.py Normal file
View file

@ -0,0 +1,28 @@
from hubspot.client import Client # type: ignore[reportMissingTypeStubs]
from hubspot.crm.deals import SimplePublicObjectInput # type: ignore[reportMissingTypeStubs]
from infrastructure.hubspot.scrubbed_calls import scrubbed_call_with_retry
class HubspotDealPropertiesClient:
"""Focused HubSpot client for writing single deal properties.
Deliberately separate from the legacy HubspotClient (frozen for the
refactor tracked in issue #1473): the SDK client is constructor-injected,
and SDK errors are re-raised scrubbed of response bodies.
"""
def __init__(self, sdk_client: Client) -> None:
self._client: Client = sdk_client
def write_deal_property(
self, deal_id: str, property_name: str, value: str
) -> None:
scrubbed_call_with_retry(
lambda: self._client.crm.deals.basic_api.update( # type: ignore[reportUnknownMemberType]
deal_id,
simple_public_object_input=SimplePublicObjectInput(
properties={property_name: value}
),
)
)

45
infrastructure/hubspot/scrubbed_calls.py Normal file
View file

@ -0,0 +1,45 @@
import json
from typing import Any, Callable, Dict, Optional, TypeVar, cast
from infrastructure.hubspot.errors import HubspotRequestError
from infrastructure.hubspot.retry import call_with_retry
T = TypeVar("T")
def scrubbed(error: Exception) -> HubspotRequestError:
"""Reduce an SDK error to non-PII diagnostics, dropping the body.
Each HubSpot sub-module ships its own ApiException class with no shared
base beyond Exception, so the fields are read by duck-typing.
"""
status = getattr(error, "status", None)
status_code = status if isinstance(status, int) else None
category: Optional[str] = None
correlation_id: Optional[str] = None
try:
parsed: Any = json.loads(getattr(error, "body", None) or "")
if isinstance(parsed, dict):
details = cast(Dict[str, Any], parsed)
raw_category = details.get("category")
raw_correlation_id = details.get("correlationId")
category = raw_category if isinstance(raw_category, str) else None
correlation_id = (
raw_correlation_id if isinstance(raw_correlation_id, str) else None
)
except ValueError:
pass
return HubspotRequestError(
status_code=status_code, category=category, correlation_id=correlation_id
)
def scrubbed_call_with_retry(fn: Callable[[], T]) -> T:
"""Run one SDK call under the shared retry policy, scrubbed on failure."""
try:
return call_with_retry(fn)
except Exception as error:
# `from None` keeps the body-carrying SDK error out of tracebacks.
raise scrubbed(error) from None

248
orchestration/abri_orchestrator.py Normal file
View file

@ -0,0 +1,248 @@
from dataclasses import dataclass
from datetime import date
from typing import Dict, List, Optional, Protocol, Tuple, Union
from domain.abri.descriptions import build_job_descriptions
from domain.abri.models import (
AbriRequestRejected,
LogJobRequest,
PlaceRef,
Tenant,
)
from domain.abri.slots import slot_for_confirmed_time
from infrastructure.abri.abri_client import AbriClient
from infrastructure.hubspot.deal_contacts_client import HubspotDealContactsClient
from infrastructure.hubspot.deal_properties_client import HubspotDealPropertiesClient
from infrastructure.hubspot.errors import HubspotRequestError
# The HubSpot deal property that carries the OpenHousing job number; the
# property ID is fixed externally, the domain and database call it job_no.
JOB_NO_DEAL_PROPERTY = "client_booking_reference"
def _vulnerability_description(tenant: Tenant) -> Optional[str]:
"""The tenant's distinct vulnerabilities, verbatim and newline-joined.
Case-insensitively distinct within the tenant, keeping the first-seen
casing and document order.
"""
distinct: List[str] = []
seen: set[str] = set()
for vulnerability in tenant.vulnerabilities:
if vulnerability.lower() in seen:
continue
seen.add(vulnerability.lower())
distinct.append(vulnerability)
return "\n".join(distinct) if distinct else None
def _contact_properties(tenant: Tenant) -> Dict[str, str]:
candidates: Dict[str, Optional[str]] = {
"firstname": tenant.forenames,
"lastname": tenant.surname,
"mobilephone": tenant.mobile,
"phone": tenant.telephone,
"secondary_phone_number": tenant.secondary_number,
"vulnerability_description": _vulnerability_description(tenant),
}
properties = {name: value for name, value in candidates.items() if value}
# Always written, so "not vulnerable" is distinguishable from "not synced".
properties["is_vulnerable"] = "true" if tenant.vulnerabilities else "false"
return properties
@dataclass(frozen=True)
class TenantDataSyncSummary:
"""PII-free record of what a sync run did: safe to log and persist."""
tenancy_reference: str
contact_ids: Tuple[str, ...]
vulnerable_contact_count: int
TenantDataSyncResult = Union[TenantDataSyncSummary, AbriRequestRejected]
class TenantDataSyncError(Exception):
"""A sync that stopped at the first failed HubSpot write.
Carries a PII-free progress report so the operator knows the deal's
half-state before cleaning up and re-running; contacts created but not
associated are orphaned in HubSpot and need manual cleanup.
"""
def __init__(
self,
message: str,
deal_id: str,
tenant_index: int,
contact_ids_created: Tuple[str, ...],
contact_ids_associated: Tuple[str, ...],
) -> None:
self.deal_id = deal_id
self.tenant_index = tenant_index
self.contact_ids_created = contact_ids_created
self.contact_ids_associated = contact_ids_associated
super().__init__(
f"{message} (deal: {deal_id}, tenant index: {tenant_index}, "
f"contacts created: {list(contact_ids_created)}, "
f"associated: {list(contact_ids_associated)})"
)
@property
def orphaned_contact_ids(self) -> Tuple[str, ...]:
return tuple(
contact_id
for contact_id in self.contact_ids_created
if contact_id not in self.contact_ids_associated
)
@dataclass(frozen=True)
class ConfirmedSurveyBooking:
"""The deal fields a confirmed survey booking contributes to a LogJob."""
deal_id: str
place_ref: PlaceRef
deal_name: str
confirmed_survey_date: date
confirmed_survey_time: Optional[str]
@dataclass(frozen=True)
class LogJobSummary:
"""PII-free record of a logged job: safe to log and persist."""
deal_id: str
job_no: str
LogJobOrchestrationResult = Union[LogJobSummary, AbriRequestRejected]
class DealDatabaseGateway(Protocol):
def record_job_no(self, deal_id: str, job_no: str) -> None: ...
class LogJobWriteBackError(Exception):
"""A job logged in OpenHousing whose job_no could not be written back.
Carries the orphaned job_no so an operator can paste it into HubSpot
manually; the run must never retry log_job, which would log a duplicate
job in OpenHousing.
"""
def __init__(self, message: str, deal_id: str, job_no: str) -> None:
self.deal_id = deal_id
self.job_no = job_no
super().__init__(f"{message} (deal: {deal_id}, orphaned job_no: {job_no})")
def client_ref_for_deal(deal_id: str) -> str:
"""The client_ref sent to OpenHousing: the HubSpot deal ID, for now.
UNCONFIRMED with Abri: whether client_ref must be unique per job (a deal
re-logging after a cancellation would repeat the bare deal ID) is an open
question on issue #1478. Swap the derivation here once Abri answers.
"""
return deal_id
class AbriOrchestrator:
"""Runs the Abri OpenHousing flows for a HubSpot deal, one method each.
Tenant data flows straight from Abri to HubSpot; nothing tenant-related
is persisted in Domna's database or logs. For job logging, HubSpot is
written first (it is the source of truth); the database second, so a
failed database write self-heals on the next scrape.
"""
def __init__(
self,
abri_client: AbriClient,
deal_contacts: HubspotDealContactsClient,
deal_properties: HubspotDealPropertiesClient,
deal_database: DealDatabaseGateway,
) -> None:
self._abri = abri_client
self._deal_contacts = deal_contacts
self._deal_properties = deal_properties
self._deal_database = deal_database
def sync_tenant_data(
self, place_ref: PlaceRef, deal_id: str
) -> TenantDataSyncResult:
tenancy = self._abri.get_tenant_data(place_ref)
if isinstance(tenancy, AbriRequestRejected):
return tenancy
contact_ids: List[str] = []
associated_ids: List[str] = []
for tenant_index, tenant in enumerate(tenancy.tenants):
try:
contact_id = self._deal_contacts.create_contact(
_contact_properties(tenant)
)
contact_ids.append(contact_id)
self._deal_contacts.associate_contact_to_deal(
deal_id=deal_id, contact_id=contact_id
)
associated_ids.append(contact_id)
except HubspotRequestError as error:
# Fail fast at the first failed write; `error` is already
# scrubbed, so chaining it keeps tracebacks PII-free.
raise TenantDataSyncError(
message=str(error),
deal_id=deal_id,
tenant_index=tenant_index,
contact_ids_created=tuple(contact_ids),
contact_ids_associated=tuple(associated_ids),
) from error
return TenantDataSyncSummary(
tenancy_reference=tenancy.tenancy_reference,
contact_ids=tuple(contact_ids),
vulnerable_contact_count=sum(
1 for tenant in tenancy.tenants if tenant.vulnerabilities
),
)
def log_job(self, booking: ConfirmedSurveyBooking) -> LogJobOrchestrationResult:
descriptions = build_job_descriptions(booking.deal_name)
outcome = self._abri.log_job(
LogJobRequest(
client_ref=client_ref_for_deal(booking.deal_id),
place_ref=booking.place_ref,
appointment_date=booking.confirmed_survey_date,
appointment_time=slot_for_confirmed_time(
booking.confirmed_survey_time
),
short_description=descriptions.short_description,
long_description=descriptions.long_description,
)
)
if isinstance(outcome, AbriRequestRejected):
return outcome
try:
self._deal_properties.write_deal_property(
deal_id=booking.deal_id,
property_name=JOB_NO_DEAL_PROPERTY,
value=outcome.job_no,
)
except HubspotRequestError as error:
# The job now exists in OpenHousing; never re-run log_job (it
# would log a duplicate). `error` is already scrubbed, so
# chaining it keeps tracebacks free of the response body.
raise LogJobWriteBackError(
message=str(error),
deal_id=booking.deal_id,
job_no=outcome.job_no,
) from error
self._deal_database.record_job_no(
deal_id=booking.deal_id, job_no=outcome.job_no
)
return LogJobSummary(deal_id=booking.deal_id, job_no=outcome.job_no)

136
orchestration/abri_tenant_data_sync_orchestrator.py
View file

@ -1,136 +0,0 @@
from dataclasses import dataclass
from typing import Dict, List, Optional, Tuple, Union
from domain.abri.models import AbriRequestRejected, PlaceRef, Tenant
from infrastructure.hubspot.deal_contacts_client import HubspotDealContactsClient
from infrastructure.hubspot.errors import HubspotRequestError
from infrastructure.abri.abri_client import AbriClient
def _vulnerability_description(tenant: Tenant) -> Optional[str]:
"""The tenant's distinct vulnerabilities, verbatim and newline-joined.
Case-insensitively distinct within the tenant, keeping the first-seen
casing and document order.
"""
distinct: List[str] = []
seen: set[str] = set()
for vulnerability in tenant.vulnerabilities:
if vulnerability.lower() in seen:
continue
seen.add(vulnerability.lower())
distinct.append(vulnerability)
return "\n".join(distinct) if distinct else None
def _contact_properties(tenant: Tenant) -> Dict[str, str]:
candidates: Dict[str, Optional[str]] = {
"firstname": tenant.forenames,
"lastname": tenant.surname,
"mobilephone": tenant.mobile,
"phone": tenant.telephone,
"secondary_phone_number": tenant.secondary_number,
"vulnerability_description": _vulnerability_description(tenant),
}
properties = {name: value for name, value in candidates.items() if value}
# Always written, so "not vulnerable" is distinguishable from "not synced".
properties["is_vulnerable"] = "true" if tenant.vulnerabilities else "false"
return properties
@dataclass(frozen=True)
class TenantDataSyncSummary:
"""PII-free record of what a sync run did: safe to log and persist."""
tenancy_reference: str
contact_ids: Tuple[str, ...]
vulnerable_contact_count: int
TenantDataSyncResult = Union[TenantDataSyncSummary, AbriRequestRejected]
class TenantDataSyncError(Exception):
"""A sync that stopped at the first failed HubSpot write.
Carries a PII-free progress report so the operator knows the deal's
half-state before cleaning up and re-running; contacts created but not
associated are orphaned in HubSpot and need manual cleanup.
"""
def __init__(
self,
message: str,
deal_id: str,
tenant_index: int,
contact_ids_created: Tuple[str, ...],
contact_ids_associated: Tuple[str, ...],
) -> None:
self.deal_id = deal_id
self.tenant_index = tenant_index
self.contact_ids_created = contact_ids_created
self.contact_ids_associated = contact_ids_associated
super().__init__(
f"{message} (deal: {deal_id}, tenant index: {tenant_index}, "
f"contacts created: {list(contact_ids_created)}, "
f"associated: {list(contact_ids_associated)})"
)
@property
def orphaned_contact_ids(self) -> Tuple[str, ...]:
return tuple(
contact_id
for contact_id in self.contact_ids_created
if contact_id not in self.contact_ids_associated
)
class AbriTenantDataSyncOrchestrator:
"""Syncs an Abri tenancy's signatories to a HubSpot deal.
Tenant data flows straight from Abri to HubSpot; nothing is persisted
in Domna's database or logs.
"""
def __init__(
self,
abri_client: AbriClient,
hubspot: HubspotDealContactsClient,
) -> None:
self._abri = abri_client
self._hubspot = hubspot
def run(self, place_ref: PlaceRef, deal_id: str) -> TenantDataSyncResult:
tenancy = self._abri.get_tenant_data(place_ref)
if isinstance(tenancy, AbriRequestRejected):
return tenancy
contact_ids: List[str] = []
associated_ids: List[str] = []
for tenant_index, tenant in enumerate(tenancy.tenants):
try:
contact_id = self._hubspot.create_contact(_contact_properties(tenant))
contact_ids.append(contact_id)
self._hubspot.associate_contact_to_deal(
deal_id=deal_id, contact_id=contact_id
)
associated_ids.append(contact_id)
except HubspotRequestError as error:
# Fail fast at the first failed write; `error` is already
# scrubbed, so chaining it keeps tracebacks PII-free.
raise TenantDataSyncError(
message=str(error),
deal_id=deal_id,
tenant_index=tenant_index,
contact_ids_created=tuple(contact_ids),
contact_ids_associated=tuple(associated_ids),
) from error
return TenantDataSyncSummary(
tenancy_reference=tenancy.tenancy_reference,
contact_ids=tuple(contact_ids),
vulnerable_contact_count=sum(
1 for tenant in tenancy.tenants if tenant.vulnerabilities
),
)

24
scripts/smoke_test_tenant_contacts.py
View file

@ -1,7 +1,7 @@
"""Manual smoke test for the Abri tenant-data -> HubSpot sync (issue #1467).
Drives the real TenantDataSyncOrchestrator end-to-end with only the Abri HTTP
edge stubbed (no Abri API key needed): the canned tenancy XML below flows
Drives the real AbriOrchestrator tenant-data flow end-to-end with only the
Abri HTTP edge stubbed (no Abri API key needed): the canned tenancy XML below flows
through the real parse, phone-priority selection and vulnerability mapping,
and the resulting contact writes hit the real HubSpot portal.
@ -31,9 +31,8 @@ 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 orchestration.abri_tenant_data_sync_orchestrator import (
AbriTenantDataSyncOrchestrator,
)
from infrastructure.hubspot.deal_properties_client import HubspotDealPropertiesClient
from orchestration.abri_orchestrator import AbriOrchestrator
DEAL_ID = "485125892321"
@ -101,6 +100,13 @@ def _stubbed_abri_client() -> AbriClient:
return client
class _UnusedDealDatabase:
"""Satisfies the gateway dependency; the tenant-data flow never writes 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 _archive_contacts(sdk_client: Client) -> None:
if not CONTACT_IDS_TO_ARCHIVE:
raise SystemExit("ARCHIVE is set but CONTACT_IDS_TO_ARCHIVE is empty")
@ -119,12 +125,14 @@ def main() -> None:
if DEAL_ID == "EDIT-ME":
raise SystemExit("Set DEAL_ID to a throwaway test deal id first")
orchestrator = AbriTenantDataSyncOrchestrator(
orchestrator = AbriOrchestrator(
abri_client=_stubbed_abri_client(),
hubspot=HubspotDealContactsClient(sdk_client=sdk_client),
deal_contacts=HubspotDealContactsClient(sdk_client=sdk_client),
deal_properties=HubspotDealPropertiesClient(sdk_client=sdk_client),
deal_database=_UnusedDealDatabase(),
)
result = orchestrator.run(PLACE_REF, deal_id=DEAL_ID)
result = orchestrator.sync_tenant_data(PLACE_REF, deal_id=DEAL_ID)
if isinstance(result, AbriRequestRejected):
raise SystemExit(f"Unexpected Abri rejection from stub: {result}")

319
tests/orchestration/test_abri_orchestrator_log_job.py Normal file
View file

@ -0,0 +1,319 @@
import json
import traceback
import xml.etree.ElementTree as ET
from datetime import date
from pathlib import Path
from typing import List, Tuple
from unittest.mock import MagicMock, patch
import pytest
import requests
from hubspot.crm.deals import ApiException as DealsApiException # type: ignore[reportMissingTypeStubs]
from hubspot.crm.deals import SimplePublicObjectInput # 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.abri.errors import AbriResponseParseError, AbriTransportError
from infrastructure.hubspot.deal_contacts_client import HubspotDealContactsClient
from infrastructure.hubspot.deal_properties_client import HubspotDealPropertiesClient
from orchestration.abri_orchestrator import (
AbriOrchestrator,
ConfirmedSurveyBooking,
LogJobSummary,
LogJobWriteBackError,
)
FIXTURE_DIR = Path(__file__).parents[1] / "abri"
ENDPOINT_URL = "https://relay.example.test/api/DomnaRelay?code=test-function-key"
DEAL_ID = "9876543210"
CONFIG = AbriConfig(
endpoint_url=ENDPOINT_URL,
username="DomnaWeb",
password="",
default_resource="NAULKH",
)
BOOKING = ConfirmedSurveyBooking(
deal_id=DEAL_ID,
place_ref=PlaceRef("1007165"),
deal_name="49 Admers Crescent",
confirmed_survey_date=date(2026, 6, 18),
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]] = []
def record_job_no(self, deal_id: str, job_no: str) -> None:
self.recorded_job_nos.append((deal_id, job_no))
@pytest.fixture()
def mock_session() -> MagicMock:
return MagicMock()
@pytest.fixture()
def sdk_client() -> MagicMock:
return MagicMock()
@pytest.fixture()
def deal_database() -> FakeDealDatabase:
return FakeDealDatabase()
@pytest.fixture()
def orchestrator(
mock_session: MagicMock,
sdk_client: MagicMock,
deal_database: FakeDealDatabase,
) -> AbriOrchestrator:
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 LogJob envelope is built from the deal's fields ---
def test_log_job_sends_a_logjob_envelope_built_from_the_deals_fields(
orchestrator: AbriOrchestrator, mock_session: MagicMock
) -> None:
# Arrange
mock_session.post.return_value.content = _load_fixture(
"abri_relay_logjob_success_response.xml"
)
# Act
orchestrator.log_job(BOOKING)
# 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 == {
"place_ref": "1007165",
"std_job_code": "SCSEXT",
"client": "HSG",
"short_description": "Domna condition survey - 49 Admers Crescent",
"long_description": (
"Domna condition survey visit at 49 Admers Crescent, "
"booked by Domna operations."
),
"client_ref": DEAL_ID,
"appointment_date": "18/06/2026",
"appointment_time": "PM",
"resource": "NAULKH",
"resource_group": "Surveyors",
}
# --- happy path: the returned job_no lands in HubSpot, then the database ---
def test_log_job_writes_the_job_no_to_the_hubspot_deal_and_then_the_database(
orchestrator: AbriOrchestrator,
mock_session: MagicMock,
sdk_client: MagicMock,
deal_database: FakeDealDatabase,
) -> None:
# Arrange
mock_session.post.return_value.content = _load_fixture(
"abri_relay_logjob_success_response.xml"
)
# Act
result = orchestrator.log_job(BOOKING)
# Assert
assert result == LogJobSummary(deal_id=DEAL_ID, job_no="AD0226519")
sdk_client.crm.deals.basic_api.update.assert_called_once_with(
DEAL_ID,
simple_public_object_input=SimplePublicObjectInput(
properties={"client_booking_reference": "AD0226519"}
),
)
assert deal_database.recorded_job_nos == [(DEAL_ID, "AD0226519")]
# --- rejection passthrough: OpenHousing's actual reason, nothing recorded ---
def test_log_job_returns_the_openhousing_rejection_verbatim_and_records_nothing(
orchestrator: AbriOrchestrator,
mock_session: MagicMock,
sdk_client: MagicMock,
deal_database: FakeDealDatabase,
) -> None:
# Arrange
mock_session.post.return_value.content = _load_fixture(
"abri_relay_logjob_relayerror_response.xml"
)
# Act
result = orchestrator.log_job(BOOKING)
# Assert
assert isinstance(result, AbriRequestRejected)
assert result.code == "RelayError"
assert result.message.startswith("No property was found with UPRN/place_ref")
assert sdk_client.mock_calls == []
assert deal_database.recorded_job_nos == []
# --- ambiguous responses are retriable failures, never success or rejection ---
@pytest.mark.parametrize(
"body",
[
b"not xml at all <<<",
b"<Root></Root>", # no Jobs/Job_logged element
b"<Root><Jobs><Job_logged>AD0226519</Job_logged></Jobs></Root>", # no attrs
b"<unexpected />",
],
)
def test_log_job_raises_a_retriable_parse_error_on_an_unexpectedly_shaped_response(
orchestrator: AbriOrchestrator,
mock_session: MagicMock,
sdk_client: MagicMock,
deal_database: FakeDealDatabase,
body: bytes,
) -> None:
# Arrange
mock_session.post.return_value.content = body
# Act / Assert
with pytest.raises(AbriResponseParseError):
orchestrator.log_job(BOOKING)
assert sdk_client.mock_calls == []
assert deal_database.recorded_job_nos == []
def test_log_job_raises_a_transport_error_when_the_relay_is_unreachable(
orchestrator: AbriOrchestrator,
mock_session: MagicMock,
sdk_client: MagicMock,
deal_database: FakeDealDatabase,
) -> None:
# Arrange
mock_session.post.side_effect = requests.ConnectionError("connection refused")
# Act / Assert
with pytest.raises(AbriTransportError):
orchestrator.log_job(BOOKING)
assert sdk_client.mock_calls == []
assert deal_database.recorded_job_nos == []
# --- partial failure: job logged in OpenHousing, HubSpot write failed ---
def _body_echoing_api_error() -> DealsApiException:
"""A HubSpot validation error whose body echoes the submitted values."""
api_error = DealsApiException(status=400, reason="Bad Request")
api_error.body = json.dumps(
{
"status": "error",
"message": 'Invalid value "AD0226519" for client_booking_reference',
"correlationId": "corr-abc-123",
"category": "VALIDATION_ERROR",
}
)
api_error.headers = {}
return api_error
def test_log_job_raises_an_error_carrying_the_orphaned_job_no_when_the_hubspot_write_fails(
orchestrator: AbriOrchestrator,
mock_session: MagicMock,
sdk_client: MagicMock,
deal_database: FakeDealDatabase,
) -> None:
# Arrange
mock_session.post.return_value.content = _load_fixture(
"abri_relay_logjob_success_response.xml"
)
sdk_client.crm.deals.basic_api.update.side_effect = _body_echoing_api_error()
# Act
with pytest.raises(LogJobWriteBackError) as exc_info:
orchestrator.log_job(BOOKING)
# Assert: the orphaned job_no is reported for manual paste-in, the
# database write is skipped, and log_job is never retried.
error = exc_info.value
assert error.deal_id == DEAL_ID
assert error.job_no == "AD0226519"
assert "AD0226519" in str(error)
assert deal_database.recorded_job_nos == []
assert mock_session.post.call_count == 1
def test_log_job_raises_a_write_back_error_scrubbed_of_the_hubspot_response_body(
orchestrator: AbriOrchestrator,
mock_session: MagicMock,
sdk_client: MagicMock,
) -> None:
# Arrange
mock_session.post.return_value.content = _load_fixture(
"abri_relay_logjob_success_response.xml"
)
sdk_client.crm.deals.basic_api.update.side_effect = _body_echoing_api_error()
# Act
with pytest.raises(LogJobWriteBackError) as exc_info:
orchestrator.log_job(BOOKING)
# Assert: non-PII diagnostics survive; the echoed body does not
rendered_traceback = "".join(traceback.format_exception(exc_info.value))
assert "400" in str(exc_info.value)
assert "VALIDATION_ERROR" in str(exc_info.value)
assert "corr-abc-123" in str(exc_info.value)
assert "Invalid value" not in rendered_traceback
def test_log_job_retries_a_rate_limited_hubspot_write_before_succeeding(
orchestrator: AbriOrchestrator,
mock_session: MagicMock,
sdk_client: MagicMock,
deal_database: FakeDealDatabase,
) -> None:
# Arrange
mock_session.post.return_value.content = _load_fixture(
"abri_relay_logjob_success_response.xml"
)
rate_limited = DealsApiException(status=429, reason="Too Many Requests")
rate_limited.headers = {"x-hubspot-ratelimit-interval-milliseconds": "0"}
sdk_client.crm.deals.basic_api.update.side_effect = [rate_limited, MagicMock()]
# Act
result = orchestrator.log_job(BOOKING)
# Assert
assert result == LogJobSummary(deal_id=DEAL_ID, job_no="AD0226519")
assert sdk_client.crm.deals.basic_api.update.call_count == 2
assert deal_database.recorded_job_nos == [(DEAL_ID, "AD0226519")]

104
tests/orchestration/test_abri_tenant_data_sync_orchestrator.py → tests/orchestration/test_abri_orchestrator_tenant_data_sync.py
View file

@ -12,12 +12,13 @@ from hubspot.crm.contacts import SimplePublicObjectInputForCreate # type: ignor
from domain.abri.models import AbriRequestRejected, PlaceRef
from infrastructure.hubspot.deal_contacts_client import HubspotDealContactsClient
from infrastructure.hubspot.deal_properties_client import HubspotDealPropertiesClient
from infrastructure.abri.abri_client import AbriClient
from infrastructure.abri.config import AbriConfig
from infrastructure.abri.errors import AbriResponseParseError, AbriTransportError
from orchestration.abri_tenant_data_sync_orchestrator import (
from orchestration.abri_orchestrator import (
TenantDataSyncError,
AbriTenantDataSyncOrchestrator,
AbriOrchestrator,
TenantDataSyncSummary,
)
@ -59,32 +60,41 @@ def sdk_client() -> MagicMock:
return sdk
class FakeDealDatabase:
"""In-memory stand-in for the deal-database gateway; unused by this flow."""
def record_job_no(self, deal_id: str, job_no: str) -> None:
raise AssertionError("tenant-data sync must not touch the deal database")
@pytest.fixture()
def orchestrator(
mock_session: MagicMock, sdk_client: MagicMock
) -> AbriTenantDataSyncOrchestrator:
) -> AbriOrchestrator:
with patch(
"infrastructure.abri.abri_client.requests.Session",
return_value=mock_session,
):
abri_client = AbriClient(config=CONFIG)
return AbriTenantDataSyncOrchestrator(
return AbriOrchestrator(
abri_client=abri_client,
hubspot=HubspotDealContactsClient(sdk_client=sdk_client),
deal_contacts=HubspotDealContactsClient(sdk_client=sdk_client),
deal_properties=HubspotDealPropertiesClient(sdk_client=sdk_client),
deal_database=FakeDealDatabase(),
)
# --- outbound request: the spec's recorded envelope ---
def test_run_sends_the_recorded_getscstenantdata_envelope_to_the_relay_endpoint(
orchestrator: AbriTenantDataSyncOrchestrator, mock_session: MagicMock
def test_sync_tenant_data_sends_the_recorded_getscstenantdata_envelope_to_the_relay_endpoint(
orchestrator: AbriOrchestrator, mock_session: MagicMock
) -> None:
# Arrange
mock_session.post.return_value.content = _tenancy_response("")
# Act
orchestrator.run(PLACE_REF, deal_id=DEAL_ID)
orchestrator.sync_tenant_data(PLACE_REF, deal_id=DEAL_ID)
# Assert
(url,) = mock_session.post.call_args.args
@ -99,8 +109,8 @@ def test_run_sends_the_recorded_getscstenantdata_envelope_to_the_relay_endpoint(
# --- happy path: signatories become associated deal contacts ---
def test_run_creates_an_associated_deal_contact_per_signatory_and_returns_a_summary(
orchestrator: AbriTenantDataSyncOrchestrator,
def test_sync_tenant_data_creates_an_associated_deal_contact_per_signatory_and_returns_a_summary(
orchestrator: AbriOrchestrator,
mock_session: MagicMock,
sdk_client: MagicMock,
) -> None:
@ -112,7 +122,7 @@ def test_run_creates_an_associated_deal_contact_per_signatory_and_returns_a_summ
)
# Act
result = orchestrator.run(PLACE_REF, deal_id=DEAL_ID)
result = orchestrator.sync_tenant_data(PLACE_REF, deal_id=DEAL_ID)
# Assert
assert result == TenantDataSyncSummary(
@ -144,8 +154,8 @@ def test_run_creates_an_associated_deal_contact_per_signatory_and_returns_a_summ
)
def test_run_creates_a_separate_contact_for_each_signatory(
orchestrator: AbriTenantDataSyncOrchestrator,
def test_sync_tenant_data_creates_a_separate_contact_for_each_signatory(
orchestrator: AbriOrchestrator,
mock_session: MagicMock,
sdk_client: MagicMock,
) -> None:
@ -159,7 +169,7 @@ def test_run_creates_a_separate_contact_for_each_signatory(
]
# Act
result = orchestrator.run(PLACE_REF, deal_id=DEAL_ID)
result = orchestrator.sync_tenant_data(PLACE_REF, deal_id=DEAL_ID)
# Assert
assert result == TenantDataSyncSummary(
@ -193,8 +203,8 @@ def test_run_creates_a_separate_contact_for_each_signatory(
assert associated_contact_ids == ["60123", "60124"]
def test_run_picks_the_best_priority_number_of_each_kind_for_a_contact(
orchestrator: AbriTenantDataSyncOrchestrator,
def test_sync_tenant_data_picks_the_best_priority_number_of_each_kind_for_a_contact(
orchestrator: AbriOrchestrator,
mock_session: MagicMock,
sdk_client: MagicMock,
) -> None:
@ -213,7 +223,7 @@ def test_run_picks_the_best_priority_number_of_each_kind_for_a_contact(
)
# Act
orchestrator.run(PLACE_REF, deal_id=DEAL_ID)
orchestrator.sync_tenant_data(PLACE_REF, deal_id=DEAL_ID)
# Assert
created_properties = sdk_client.crm.contacts.basic_api.create.call_args.kwargs[
@ -223,8 +233,8 @@ def test_run_picks_the_best_priority_number_of_each_kind_for_a_contact(
assert created_properties["phone"] == "01000000007"
def test_run_fills_secondary_phone_number_with_the_best_leftover_number(
orchestrator: AbriTenantDataSyncOrchestrator,
def test_sync_tenant_data_fills_secondary_phone_number_with_the_best_leftover_number(
orchestrator: AbriOrchestrator,
mock_session: MagicMock,
sdk_client: MagicMock,
) -> None:
@ -240,7 +250,7 @@ def test_run_fills_secondary_phone_number_with_the_best_leftover_number(
)
# Act
orchestrator.run(PLACE_REF, deal_id=DEAL_ID)
orchestrator.sync_tenant_data(PLACE_REF, deal_id=DEAL_ID)
# Assert
sdk_client.crm.contacts.basic_api.create.assert_called_once_with(
@ -260,8 +270,8 @@ def test_run_fills_secondary_phone_number_with_the_best_leftover_number(
# --- vulnerabilities: recorded on the tenant's own contact ---
def test_run_records_vulnerabilities_on_the_tenants_own_contact(
orchestrator: AbriTenantDataSyncOrchestrator,
def test_sync_tenant_data_records_vulnerabilities_on_the_tenants_own_contact(
orchestrator: AbriOrchestrator,
mock_session: MagicMock,
sdk_client: MagicMock,
) -> None:
@ -282,7 +292,7 @@ def test_run_records_vulnerabilities_on_the_tenants_own_contact(
]
# Act
result = orchestrator.run(PLACE_REF, deal_id=DEAL_ID)
result = orchestrator.sync_tenant_data(PLACE_REF, deal_id=DEAL_ID)
# Assert
assert result == TenantDataSyncSummary(
@ -309,8 +319,8 @@ def test_run_records_vulnerabilities_on_the_tenants_own_contact(
]
def test_run_still_creates_a_contact_for_an_all_blank_signatory(
orchestrator: AbriTenantDataSyncOrchestrator,
def test_sync_tenant_data_still_creates_a_contact_for_an_all_blank_signatory(
orchestrator: AbriOrchestrator,
mock_session: MagicMock,
sdk_client: MagicMock,
) -> None:
@ -321,7 +331,7 @@ def test_run_still_creates_a_contact_for_an_all_blank_signatory(
)
# Act
result = orchestrator.run(PLACE_REF, deal_id=DEAL_ID)
result = orchestrator.sync_tenant_data(PLACE_REF, deal_id=DEAL_ID)
# Assert
assert result == TenantDataSyncSummary(
@ -339,8 +349,8 @@ def test_run_still_creates_a_contact_for_an_all_blank_signatory(
# --- unusual but valid outcomes ---
def test_run_treats_a_tenancy_with_no_signatories_as_a_valid_zero_contact_outcome(
orchestrator: AbriTenantDataSyncOrchestrator,
def test_sync_tenant_data_treats_a_tenancy_with_no_signatories_as_a_valid_zero_contact_outcome(
orchestrator: AbriOrchestrator,
mock_session: MagicMock,
sdk_client: MagicMock,
) -> None:
@ -348,7 +358,7 @@ def test_run_treats_a_tenancy_with_no_signatories_as_a_valid_zero_contact_outcom
mock_session.post.return_value.content = _tenancy_response("")
# Act
result = orchestrator.run(PLACE_REF, deal_id=DEAL_ID)
result = orchestrator.sync_tenant_data(PLACE_REF, deal_id=DEAL_ID)
# Assert
assert result == TenantDataSyncSummary(
@ -362,8 +372,8 @@ def test_run_treats_a_tenancy_with_no_signatories_as_a_valid_zero_contact_outcom
# --- rejection passthrough ---
def test_run_returns_the_abri_rejection_verbatim_and_makes_no_hubspot_calls(
orchestrator: AbriTenantDataSyncOrchestrator,
def test_sync_tenant_data_returns_the_abri_rejection_verbatim_and_makes_no_hubspot_calls(
orchestrator: AbriOrchestrator,
mock_session: MagicMock,
sdk_client: MagicMock,
) -> None:
@ -373,7 +383,7 @@ def test_run_returns_the_abri_rejection_verbatim_and_makes_no_hubspot_calls(
)
# Act
result = orchestrator.run(PLACE_REF, deal_id=DEAL_ID)
result = orchestrator.sync_tenant_data(PLACE_REF, deal_id=DEAL_ID)
# Assert
assert isinstance(result, AbriRequestRejected)
@ -395,8 +405,8 @@ def test_run_returns_the_abri_rejection_verbatim_and_makes_no_hubspot_calls(
b"<unexpected />",
],
)
def test_run_raises_a_retriable_parse_error_on_an_unexpectedly_shaped_response(
orchestrator: AbriTenantDataSyncOrchestrator,
def test_sync_tenant_data_raises_a_retriable_parse_error_on_an_unexpectedly_shaped_response(
orchestrator: AbriOrchestrator,
mock_session: MagicMock,
sdk_client: MagicMock,
body: bytes,
@ -406,12 +416,12 @@ def test_run_raises_a_retriable_parse_error_on_an_unexpectedly_shaped_response(
# Act / Assert
with pytest.raises(AbriResponseParseError):
orchestrator.run(PLACE_REF, deal_id=DEAL_ID)
orchestrator.sync_tenant_data(PLACE_REF, deal_id=DEAL_ID)
assert sdk_client.mock_calls == []
def test_run_raises_a_transport_error_when_the_relay_is_unreachable(
orchestrator: AbriTenantDataSyncOrchestrator,
def test_sync_tenant_data_raises_a_transport_error_when_the_relay_is_unreachable(
orchestrator: AbriOrchestrator,
mock_session: MagicMock,
sdk_client: MagicMock,
) -> None:
@ -420,7 +430,7 @@ def test_run_raises_a_transport_error_when_the_relay_is_unreachable(
# Act / Assert
with pytest.raises(AbriTransportError):
orchestrator.run(PLACE_REF, deal_id=DEAL_ID)
orchestrator.sync_tenant_data(PLACE_REF, deal_id=DEAL_ID)
assert sdk_client.mock_calls == []
@ -442,8 +452,8 @@ def _pii_echoing_api_error() -> ContactsApiException:
return api_error
def test_run_raises_an_error_scrubbed_of_the_hubspot_response_body(
orchestrator: AbriTenantDataSyncOrchestrator,
def test_sync_tenant_data_raises_an_error_scrubbed_of_the_hubspot_response_body(
orchestrator: AbriOrchestrator,
mock_session: MagicMock,
sdk_client: MagicMock,
) -> None:
@ -456,7 +466,7 @@ def test_run_raises_an_error_scrubbed_of_the_hubspot_response_body(
# Act
with pytest.raises(Exception) as exc_info:
orchestrator.run(PLACE_REF, deal_id=DEAL_ID)
orchestrator.sync_tenant_data(PLACE_REF, deal_id=DEAL_ID)
# Assert: non-PII diagnostics survive; the echoed body does not
rendered_traceback = "".join(traceback.format_exception(exc_info.value))
@ -468,8 +478,8 @@ def test_run_raises_an_error_scrubbed_of_the_hubspot_response_body(
assert "07700900123" not in rendered_traceback
def test_run_fails_fast_when_association_fails_and_reports_the_orphaned_contact(
orchestrator: AbriTenantDataSyncOrchestrator,
def test_sync_tenant_data_fails_fast_when_association_fails_and_reports_the_orphaned_contact(
orchestrator: AbriOrchestrator,
mock_session: MagicMock,
sdk_client: MagicMock,
) -> None:
@ -483,7 +493,7 @@ def test_run_fails_fast_when_association_fails_and_reports_the_orphaned_contact(
# Act
with pytest.raises(TenantDataSyncError) as exc_info:
orchestrator.run(PLACE_REF, deal_id=DEAL_ID)
orchestrator.sync_tenant_data(PLACE_REF, deal_id=DEAL_ID)
# Assert: the flow stopped at the first failed write, with progress reported
error = exc_info.value
@ -499,8 +509,8 @@ def test_run_fails_fast_when_association_fails_and_reports_the_orphaned_contact(
assert "07700900456" not in rendered_traceback
def test_run_retries_a_rate_limited_hubspot_call_before_succeeding(
orchestrator: AbriTenantDataSyncOrchestrator,
def test_sync_tenant_data_retries_a_rate_limited_hubspot_call_before_succeeding(
orchestrator: AbriOrchestrator,
mock_session: MagicMock,
sdk_client: MagicMock,
) -> None:
@ -517,7 +527,7 @@ def test_run_retries_a_rate_limited_hubspot_call_before_succeeding(
]
# Act
result = orchestrator.run(PLACE_REF, deal_id=DEAL_ID)
result = orchestrator.sync_tenant_data(PLACE_REF, deal_id=DEAL_ID)
# Assert
assert result == TenantDataSyncSummary(