Model/etl/hubspot/tests/test_scraper_handler.py

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"


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],
) -> 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_cls.return_value.get_deal_and_company_and_listing.return_value = (
            hubspot_deal,
            None,
            listing,
        )
        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


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