A manual smoke script proves log then amend end-to-end against Abri UAT 🟩

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

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()