From de2a6d1d7913f278014ddde948ef0d2d3f9296d1 Mon Sep 17 00:00:00 2001 From: Daniel Roth Date: Mon, 6 Jul 2026 14:19:05 +0000 Subject: [PATCH] Add manual smoke script for the Abri tenant-contact HubSpot writes Co-Authored-By: Claude Fable 5 --- scripts/smoke_test_tenant_contacts.py | 140 ++++++++++++++++++++++++++ 1 file changed, 140 insertions(+) create mode 100644 scripts/smoke_test_tenant_contacts.py diff --git a/scripts/smoke_test_tenant_contacts.py b/scripts/smoke_test_tenant_contacts.py new file mode 100644 index 000000000..0b7073bd9 --- /dev/null +++ b/scripts/smoke_test_tenant_contacts.py @@ -0,0 +1,140 @@ +"""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 +through the real parse, phone-priority selection and vulnerability mapping, +and the resulting contact writes hit the real HubSpot portal. + +Usage: + 1. Ensure HUBSPOT_API_KEY is set in backend/.env. + 2. Create a throwaway test deal in the HubSpot UI and paste its id into + DEAL_ID below. + 3. Run: python scripts/smoke_test_tenant_contacts.py + 4. Check the deal in the UI: two contacts (Jane Doe, Joe Bloggs), with + Jane carrying mobilephone, phone, secondary_phone_number, + is_vulnerable=true and a two-line vulnerability_description, and Joe + carrying only a name and is_vulnerable=false. + 5. Clean up: paste the printed contact ids into CONTACT_IDS_TO_ARCHIVE, + set ARCHIVE = True and re-run, then delete the test deal in the UI. + +All tenant details below are fictional (Ofcom-reserved number ranges). +""" + +import os +from pathlib import Path +from typing import Any, List, cast + +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 orchestration.abri_tenant_data_sync_orchestrator import ( + AbriTenantDataSyncOrchestrator, +) + +DEAL_ID = "EDIT-ME" + +# Cleanup mode: set ARCHIVE = True and list the contact ids printed by a +# previous run to archive them instead of creating new ones. +ARCHIVE = False +CONTACT_IDS_TO_ARCHIVE: List[str] = [] + +PLACE_REF = PlaceRef("SMOKE1") + +# Jane Doe exercises best-of-kind phone selection, the runner-up pick into +# secondary_phone_number, and the vulnerability fields; Joe Bloggs exercises +# the always-written is_vulnerable=false with every other property omitted. +SMOKE_TENANCY_XML = b""" + SMOKETEST0001900000107700900123 + 07700900456 + 01632960123 + 01632960789 + Blindness + Hard of hearing + + 9000002 + +""" + + +class _CannedResponse: + content = SMOKE_TENANCY_XML + + def raise_for_status(self) -> None: + pass + + +class _StubAbriSession: + """Stands in for requests.Session; returns the canned tenancy.""" + + def post(self, url: str, data: bytes) -> _CannedResponse: + return _CannedResponse() + + +def _hubspot_sdk_client() -> Client: + # Read the key directly rather than via backend.app.config.get_settings, + # whose strict Settings model rejects unrelated extra keys in backend/.env. + env_file = Path(__file__).parents[1] / "backend" / ".env" + access_token = os.getenv("HUBSPOT_API_KEY") or dotenv_values(env_file).get( + "HUBSPOT_API_KEY" + ) + if not access_token: + raise SystemExit("Missing HUBSPOT_API_KEY (env var or backend/.env)") + return Client.create(access_token=access_token) # type: ignore[reportUnknownMemberType] + + +def _stubbed_abri_client() -> AbriClient: + client = AbriClient( + config=AbriConfig( + endpoint_url="https://stubbed.invalid/relay", + username="smoke-test", + password="", + default_resource="", + ) + ) + client._session = cast(Any, _StubAbriSession()) # pyright: ignore[reportPrivateUsage] + return client + + +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") + for contact_id in CONTACT_IDS_TO_ARCHIVE: + sdk_client.crm.contacts.basic_api.archive(contact_id=contact_id) # type: ignore[reportUnknownMemberType] + print(f"Archived contact {contact_id}") + + +def main() -> None: + sdk_client = _hubspot_sdk_client() + + if ARCHIVE: + _archive_contacts(sdk_client) + return + + if DEAL_ID == "EDIT-ME": + raise SystemExit("Set DEAL_ID to a throwaway test deal id first") + + orchestrator = AbriTenantDataSyncOrchestrator( + abri_client=_stubbed_abri_client(), + hubspot=HubspotDealContactsClient(sdk_client=sdk_client), + ) + + result = orchestrator.run(PLACE_REF, deal_id=DEAL_ID) + + if isinstance(result, AbriRequestRejected): + raise SystemExit(f"Unexpected Abri rejection from stub: {result}") + + print(f"Tenancy reference: {result.tenancy_reference}") + print(f"Contacts created: {list(result.contact_ids)}") + print(f"Vulnerable contact count: {result.vulnerable_contact_count}") + print( + "\nTo clean up: paste the ids above into CONTACT_IDS_TO_ARCHIVE, " + "set ARCHIVE = True and re-run." + ) + + +if __name__ == "__main__": + main()