import json import uuid from typing import Any, Dict, Optional from unittest.mock import MagicMock, patch from backend.app.db.models.hubspot_deal_data import HubspotDealData from etl.hubspot.scripts.scraper.main import handler DEAL_NAME = "123 Main Street" UPRN = "12345678" 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" PROJECT = {"project_id": "proj-1", "name": "Project One"} def make_hubspot_deal(**kwargs: Any) -> Dict[str, Any]: return { "hs_object_id": DEAL_ID, "dealname": DEAL_NAME, "pashub_link": None, **kwargs, } def make_db_deal(**kwargs: Any) -> HubspotDealData: return HubspotDealData( id=uuid.uuid4(), deal_id=DEAL_ID, **kwargs, ) def run_handler( hubspot_deal: Dict[str, Any], db_deal: Optional[HubspotDealData], listing: Optional[dict], project: Optional[dict] = None, ) -> tuple[MagicMock, MagicMock]: mock_sqs = MagicMock() mock_sqs.send_message.return_value = {"MessageId": "test-id"} with ( patch("etl.hubspot.scripts.scraper.main.HubspotDataToDb") as mock_db_cls, patch("etl.hubspot.scripts.scraper.main.HubspotClient") as mock_hs_cls, patch("etl.hubspot.scripts.scraper.main.boto3") as mock_boto3, patch("etl.hubspot.scripts.scraper.main.get_settings") as mock_settings, ): mock_db_cls.return_value.find_deal_with_deal_id.return_value = db_deal mock_db_cls.return_value.upsert_deal.return_value = None mock_hs = mock_hs_cls.return_value mock_hs.get_deal_and_company_and_listing_and_project.return_value = ( hubspot_deal, None, listing, project, ) 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 handler.__wrapped__({"hubspot_deal_id": DEAL_ID}, "") return mock_sqs, mock_db_cls.return_value # ==================================== # NEW DEAL PATH - MagicPlan trigger # ==================================== def test_new_deal_surveyed__sends_magicplan_sqs() -> None: # Arrange hubspot_deal = make_hubspot_deal(outcome="surveyed") listing = {"national_uprn": UPRN} # Act mock_sqs, _ = run_handler(hubspot_deal=hubspot_deal, db_deal=None, listing=listing) # Assert mock_sqs.send_message.assert_called_once_with( QueueUrl=MAGICPLAN_QUEUE_URL, MessageBody=json.dumps( {"address": DEAL_NAME, "hubspot_deal_id": DEAL_ID, "uprn": UPRN} ), ) def test_new_deal_not_surveyed__no_magicplan_sqs() -> None: # Arrange hubspot_deal = make_hubspot_deal(outcome="assessed") # Act mock_sqs, _ = run_handler(hubspot_deal=hubspot_deal, db_deal=None, listing=None) # Assert mock_sqs.send_message.assert_not_called() def test_new_deal_surveyed_no_listing__magicplan_sqs_uprn_is_null() -> None: # Arrange hubspot_deal = make_hubspot_deal(outcome="surveyed") # Act mock_sqs, _ = run_handler(hubspot_deal=hubspot_deal, db_deal=None, listing=None) # Assert mock_sqs.send_message.assert_called_once_with( QueueUrl=MAGICPLAN_QUEUE_URL, MessageBody=json.dumps( {"address": DEAL_NAME, "hubspot_deal_id": DEAL_ID, "uprn": None} ), ) # ========================================== # EXISTING DEAL PATH - MagicPlan trigger # ========================================== def test_existing_deal_surveyed_transition__sends_magicplan_sqs() -> None: # Arrange db_deal = make_db_deal(outcome="assessed") hubspot_deal = make_hubspot_deal(outcome="surveyed") listing = {"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=MAGICPLAN_QUEUE_URL, MessageBody=json.dumps( {"address": DEAL_NAME, "hubspot_deal_id": DEAL_ID, "uprn": UPRN} ), ) def test_existing_deal_already_surveyed__no_magicplan_sqs() -> None: # Arrange db_deal = make_db_deal(outcome="surveyed", dealname="Old Name") hubspot_deal = make_hubspot_deal(outcome="surveyed", dealname="New Name") # Act mock_sqs, _ = run_handler(hubspot_deal=hubspot_deal, db_deal=db_deal, listing=None) # Assert mock_sqs.send_message.assert_not_called() # ==================================== # NEW DEAL PATH - PasHub trigger # ==================================== def test_new_deal_with_pashub_link__sends_pashub_sqs() -> None: # Arrange hubspot_deal = make_hubspot_deal(pashub_link=PASHUB_LINK) # Act mock_sqs, _ = run_handler(hubspot_deal=hubspot_deal, db_deal=None, listing=None) # Assert mock_sqs.send_message.assert_called_once_with( QueueUrl=PASHUB_QUEUE_URL, MessageBody=json.dumps( { "pashub_link": PASHUB_LINK, "address": None, "hubspot_deal_id": DEAL_ID, "sharepoint_link": None, "uprn": None, "landlord_property_id": None, "deal_stage": None, } ), ) def test_new_deal_no_pashub_link__no_pashub_sqs() -> None: # Arrange hubspot_deal = make_hubspot_deal() # Act mock_sqs, _ = run_handler(hubspot_deal=hubspot_deal, db_deal=None, listing=None) # Assert mock_sqs.send_message.assert_not_called() # ========================================== # EXISTING DEAL PATH - PasHub trigger # ========================================== def test_existing_deal_pashub_link_added__sends_pashub_sqs() -> None: # Arrange db_deal = make_db_deal(pashub_link=None) hubspot_deal = make_hubspot_deal(pashub_link=PASHUB_LINK) # Act mock_sqs, _ = run_handler(hubspot_deal=hubspot_deal, db_deal=db_deal, listing=None) # Assert mock_sqs.send_message.assert_called_once_with( QueueUrl=PASHUB_QUEUE_URL, MessageBody=json.dumps( { "pashub_link": PASHUB_LINK, "address": None, "hubspot_deal_id": DEAL_ID, "sharepoint_link": None, "uprn": None, "landlord_property_id": None, "deal_stage": None, } ), ) def test_existing_deal_pashub_link_unchanged__no_pashub_sqs() -> None: # Arrange db_deal = make_db_deal(pashub_link=PASHUB_LINK, dealname="Old Name") hubspot_deal = make_hubspot_deal(pashub_link=PASHUB_LINK, dealname="New Name") # Act mock_sqs, _ = run_handler(hubspot_deal=hubspot_deal, db_deal=db_deal, listing=None) # Assert mock_sqs.send_message.assert_not_called() # ==================================== # PROJECT upsert # ==================================== def test_new_deal_with_project__upserts_project() -> None: # Arrange hubspot_deal = make_hubspot_deal() # Act _, mock_db = run_handler( hubspot_deal=hubspot_deal, db_deal=None, listing=None, project=PROJECT ) # Assert mock_db.upsert_project.assert_called_once_with(PROJECT) def test_new_deal_no_project__no_project_upsert() -> None: # Arrange hubspot_deal = make_hubspot_deal() # Act _, mock_db = run_handler( hubspot_deal=hubspot_deal, db_deal=None, listing=None, project=None ) # Assert mock_db.upsert_project.assert_not_called() def test_existing_deal_changed_with_project__upserts_project() -> None: # Arrange db_deal = make_db_deal(outcome="assessed") hubspot_deal = make_hubspot_deal(outcome="surveyed") # Act _, mock_db = run_handler( hubspot_deal=hubspot_deal, db_deal=db_deal, listing=None, project=PROJECT ) # Assert mock_db.upsert_project.assert_called_once_with(PROJECT) def test_existing_deal_unchanged__no_project_upsert() -> None: # Arrange: db deal matches hubspot deal, so the differ reports no change db_deal = make_db_deal(dealname=DEAL_NAME) hubspot_deal = make_hubspot_deal() # Act _, mock_db = run_handler( hubspot_deal=hubspot_deal, db_deal=db_deal, listing=None, project=PROJECT ) # Assert mock_db.upsert_project.assert_not_called() mock_db.upsert_deal.assert_not_called()