Add manual smoke script for the Abri tenant-contact HubSpot writes

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

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"""<Root>
+    <Tenancy end_date="" start_date="15/12/2008">SMOKETEST0001<Tenant forenames="Jane" main-contact="yes" person_title="Mrs" surname="Doe">9000001<Mobile priority="10">07700900123</Mobile>
+            <Mobile priority="20">07700900456</Mobile>
+            <Telephone priority="5">01632960123</Telephone>
+            <Telephone priority="30">01632960789</Telephone>
+            <Vulnerability>Blindness</Vulnerability>
+            <Vulnerability>Hard of hearing</Vulnerability>
+        </Tenant>
+        <Tenant forenames="Joe" main-contact="no" person_title="Mr" surname="Bloggs">9000002</Tenant>
+    </Tenancy>
+</Root>"""
+
+
+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()