mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
254 lines
8.9 KiB
Python
254 lines
8.9 KiB
Python
from typing import Optional
|
|
from unittest.mock import MagicMock, call, patch
|
|
|
|
|
|
from backend.pashub_fetcher.pashub_client import PashubClient
|
|
from backend.pashub_fetcher.pashub_service import PashubService
|
|
from backend.pashub_fetcher.pashub_to_ara_trigger_request import (
|
|
PashubToAraTriggerRequest,
|
|
)
|
|
from utils.sharepoint.domna_sharepoint_client import DomnaSharepointClient
|
|
|
|
|
|
FAKE_JOB_LINK = "https://pashub.net/jobs/job-id-123/details"
|
|
|
|
|
|
def make_request(
|
|
pashub_link: str = FAKE_JOB_LINK,
|
|
uprn: Optional[str] = None,
|
|
hubspot_deal_id: Optional[str] = None,
|
|
sharepoint_link: Optional[str] = None,
|
|
) -> PashubToAraTriggerRequest:
|
|
return PashubToAraTriggerRequest(
|
|
pashub_link=pashub_link,
|
|
uprn=uprn,
|
|
hubspot_deal_id=hubspot_deal_id,
|
|
sharepoint_link=sharepoint_link,
|
|
)
|
|
|
|
|
|
def make_service(
|
|
pashub_client: Optional[PashubClient] = None,
|
|
sharepoint_client: Optional[DomnaSharepointClient] = None,
|
|
s3_bucket: str = "test-bucket",
|
|
) -> PashubService:
|
|
return PashubService(
|
|
pashub_client=pashub_client or MagicMock(spec=PashubClient),
|
|
sharepoint_client=sharepoint_client or MagicMock(spec=DomnaSharepointClient),
|
|
s3_bucket=s3_bucket,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# run(): returns file paths
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_run_returns_file_paths() -> None:
|
|
mock_client = MagicMock(spec=PashubClient)
|
|
mock_client.get_uprn_by_job_id.return_value = None
|
|
mock_client.get_core_evidence_files_by_job_id.return_value = [
|
|
"/tmp/a.pdf",
|
|
"/tmp/b.pdf",
|
|
]
|
|
|
|
service = make_service(pashub_client=mock_client)
|
|
|
|
with patch("backend.pashub_fetcher.pashub_service.os.remove"):
|
|
result = service.run(make_request())
|
|
|
|
assert result == ["/tmp/a.pdf", "/tmp/b.pdf"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# run(): skips upload when neither uprn nor hubspot_deal_id
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_run_skips_upload_when_no_uprn_and_no_deal_id() -> None:
|
|
mock_client = MagicMock(spec=PashubClient)
|
|
mock_client.get_uprn_by_job_id.return_value = None
|
|
mock_client.get_core_evidence_files_by_job_id.return_value = ["/tmp/a.pdf"]
|
|
|
|
service = make_service(pashub_client=mock_client)
|
|
|
|
with (
|
|
patch("backend.pashub_fetcher.pashub_service.upload_file_to_s3") as mock_s3,
|
|
patch("backend.pashub_fetcher.pashub_service.os.remove"),
|
|
):
|
|
service.run(make_request(uprn=None, hubspot_deal_id=None))
|
|
|
|
mock_s3.assert_not_called()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# run(): UPRN present → uploads each file to S3 with correct bucket/key
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_run_uploads_files_to_s3_using_uprn_path() -> None:
|
|
mock_client = MagicMock(spec=PashubClient)
|
|
mock_client.get_uprn_by_job_id.return_value = None
|
|
mock_client.get_core_evidence_files_by_job_id.return_value = [
|
|
"/tmp/SiteNote_001.pdf",
|
|
"/tmp/Photopack_002.pdf",
|
|
]
|
|
|
|
service = make_service(pashub_client=mock_client, s3_bucket="my-bucket")
|
|
|
|
with (
|
|
patch("backend.pashub_fetcher.pashub_service.upload_file_to_s3") as mock_s3,
|
|
patch("backend.pashub_fetcher.pashub_service.db_session"),
|
|
patch("backend.pashub_fetcher.pashub_service.os.remove"),
|
|
):
|
|
service.run(make_request(uprn="12345"))
|
|
|
|
mock_s3.assert_has_calls(
|
|
[
|
|
call(
|
|
"/tmp/SiteNote_001.pdf",
|
|
"my-bucket",
|
|
"documents/uprn/12345/SiteNote_001.pdf",
|
|
),
|
|
call(
|
|
"/tmp/Photopack_002.pdf",
|
|
"my-bucket",
|
|
"documents/uprn/12345/Photopack_002.pdf",
|
|
),
|
|
],
|
|
any_order=False,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# run(): UPRN present → UploadedFile records added to DB session
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_run_persists_uploaded_file_records_to_db() -> None:
|
|
mock_client = MagicMock(spec=PashubClient)
|
|
mock_client.get_uprn_by_job_id.return_value = None
|
|
mock_client.get_core_evidence_files_by_job_id.return_value = [
|
|
"/tmp/SiteNote_001.pdf"
|
|
]
|
|
|
|
fake_session = MagicMock()
|
|
service = make_service(pashub_client=mock_client)
|
|
|
|
with (
|
|
patch("backend.pashub_fetcher.pashub_service.upload_file_to_s3"),
|
|
patch("backend.pashub_fetcher.pashub_service.db_session") as mock_db,
|
|
patch("backend.pashub_fetcher.pashub_service.os.remove"),
|
|
):
|
|
mock_db.return_value.__enter__.return_value = fake_session
|
|
service.run(make_request(uprn="12345"))
|
|
|
|
fake_session.add_all.assert_called_once()
|
|
added: list = fake_session.add_all.call_args[0][0]
|
|
assert len(added) == 1
|
|
assert added[0].s3_file_bucket == "test-bucket"
|
|
assert added[0].uprn == 12345
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# run(): hubspot_deal_id only → uses deal_id S3 path prefix
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_run_uses_hubspot_deal_id_path_when_no_uprn() -> None:
|
|
mock_client = MagicMock(spec=PashubClient)
|
|
mock_client.get_uprn_by_job_id.return_value = None
|
|
mock_client.get_core_evidence_files_by_job_id.return_value = [
|
|
"/tmp/SiteNote_001.pdf"
|
|
]
|
|
|
|
service = make_service(pashub_client=mock_client, s3_bucket="my-bucket")
|
|
|
|
with (
|
|
patch("backend.pashub_fetcher.pashub_service.upload_file_to_s3") as mock_s3,
|
|
patch("backend.pashub_fetcher.pashub_service.db_session"),
|
|
patch("backend.pashub_fetcher.pashub_service.os.remove"),
|
|
):
|
|
service.run(make_request(uprn=None, hubspot_deal_id="deal-abc"))
|
|
|
|
mock_s3.assert_called_once_with(
|
|
"/tmp/SiteNote_001.pdf",
|
|
"my-bucket",
|
|
"documents/hubspot_deal_id/deal-abc/SiteNote_001.pdf",
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# run(): RD_SAP_SITE_NOTE file → site notes parsed and saved to DB
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_run_parses_and_saves_site_notes_for_rd_sap_site_note_file() -> None:
|
|
mock_client = MagicMock(spec=PashubClient)
|
|
mock_client.get_uprn_by_job_id.return_value = None
|
|
mock_client.get_core_evidence_files_by_job_id.return_value = [
|
|
"/tmp/RdSAP_SiteNote_001.pdf"
|
|
]
|
|
|
|
fake_epc_data = MagicMock()
|
|
fake_session = MagicMock()
|
|
fake_uploaded_file_id = 99
|
|
|
|
service = make_service(pashub_client=mock_client)
|
|
|
|
with (
|
|
patch("backend.pashub_fetcher.pashub_service.upload_file_to_s3"),
|
|
patch(
|
|
"backend.pashub_fetcher.pashub_service.parse_site_notes_pdf",
|
|
return_value=fake_epc_data,
|
|
) as mock_parse,
|
|
patch(
|
|
"backend.pashub_fetcher.pashub_service.save_epc_property_data"
|
|
) as mock_save,
|
|
patch("backend.pashub_fetcher.pashub_service.db_session") as mock_db,
|
|
patch("backend.pashub_fetcher.pashub_service.os.remove"),
|
|
):
|
|
fake_session.add_all = MagicMock(
|
|
side_effect=lambda files: setattr(files[0], "id", fake_uploaded_file_id)
|
|
)
|
|
mock_db.return_value.__enter__.return_value = fake_session
|
|
service.run(make_request(uprn="12345"))
|
|
|
|
mock_parse.assert_called_once_with("/tmp/RdSAP_SiteNote_001.pdf")
|
|
mock_save.assert_called_once_with(
|
|
fake_session, fake_epc_data, uploaded_file_id=fake_uploaded_file_id
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# run(): site notes parse failure → warning logged, run returns normally
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_run_warns_and_continues_when_site_notes_parsing_fails() -> None:
|
|
mock_client = MagicMock(spec=PashubClient)
|
|
mock_client.get_uprn_by_job_id.return_value = None
|
|
mock_client.get_core_evidence_files_by_job_id.return_value = [
|
|
"/tmp/RdSAP_SiteNote_001.pdf"
|
|
]
|
|
|
|
service = make_service(pashub_client=mock_client)
|
|
|
|
with (
|
|
patch("backend.pashub_fetcher.pashub_service.upload_file_to_s3"),
|
|
patch(
|
|
"backend.pashub_fetcher.pashub_service.parse_site_notes_pdf",
|
|
side_effect=ValueError("corrupt pdf"),
|
|
),
|
|
patch(
|
|
"backend.pashub_fetcher.pashub_service.save_epc_property_data"
|
|
) as mock_save,
|
|
patch("backend.pashub_fetcher.pashub_service.db_session"),
|
|
patch("backend.pashub_fetcher.pashub_service.logger") as mock_logger,
|
|
patch("backend.pashub_fetcher.pashub_service.os.remove"),
|
|
):
|
|
result = service.run(make_request(uprn="12345"))
|
|
|
|
assert result == ["/tmp/RdSAP_SiteNote_001.pdf"]
|
|
mock_logger.warning.assert_called()
|
|
mock_save.assert_not_called()
|