mirror of
https://github.com/Hestia-Homes/Model.git
synced 2026-06-08 11:17:27 +00:00
108 lines
3 KiB
Python
108 lines
3 KiB
Python
from typing import Generator
|
|
from unittest.mock import MagicMock, call, patch
|
|
|
|
import pytest
|
|
|
|
from backend.app.db.models.uploaded_file import FileTypeEnum
|
|
from backend.ecmk_fetcher.upload import upload_file_to_s3_and_record
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_uploaded_file() -> MagicMock:
|
|
obj = MagicMock()
|
|
obj.id = 42
|
|
return obj
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_session() -> MagicMock:
|
|
return MagicMock()
|
|
|
|
|
|
@pytest.fixture
|
|
def patched_deps(
|
|
mock_uploaded_file: MagicMock, mock_session: MagicMock
|
|
) -> Generator[dict[str, MagicMock], None, None]:
|
|
with (
|
|
patch(
|
|
"backend.ecmk_fetcher.upload.upload_file_to_s3"
|
|
) as mock_s3,
|
|
patch(
|
|
"backend.ecmk_fetcher.upload.db_session"
|
|
) as mock_db_ctx,
|
|
patch(
|
|
"backend.ecmk_fetcher.upload.UploadedFile",
|
|
return_value=mock_uploaded_file,
|
|
) as mock_model,
|
|
):
|
|
mock_db_ctx.return_value.__enter__.return_value = mock_session
|
|
mock_db_ctx.return_value.__exit__.return_value = False
|
|
yield {
|
|
"s3": mock_s3,
|
|
"db_ctx": mock_db_ctx,
|
|
"model": mock_model,
|
|
"session": mock_session,
|
|
"uploaded_file": mock_uploaded_file,
|
|
}
|
|
|
|
|
|
def test_returns_uploaded_file_id_as_int(
|
|
patched_deps: dict[str, MagicMock],
|
|
) -> None:
|
|
result = upload_file_to_s3_and_record(
|
|
bucket="test-bucket",
|
|
file_path="/tmp/report.pdf",
|
|
hubspot_listing_id="hs-001",
|
|
file_type=FileTypeEnum.ECMK_RD_SAP_SITE_NOTE,
|
|
)
|
|
|
|
assert isinstance(result, int)
|
|
assert result == 42
|
|
|
|
|
|
def test_uploads_to_s3_with_key_derived_from_listing_id_and_filename(
|
|
patched_deps: dict[str, MagicMock],
|
|
) -> None:
|
|
upload_file_to_s3_and_record(
|
|
bucket="my-bucket",
|
|
file_path="/some/path/site_note.pdf",
|
|
hubspot_listing_id="hs-999",
|
|
file_type=FileTypeEnum.ECMK_RD_SAP_SITE_NOTE,
|
|
)
|
|
|
|
patched_deps["s3"].assert_called_once_with(
|
|
"/some/path/site_note.pdf",
|
|
"my-bucket",
|
|
"documents/hubspot_listing_id/hs-999/site_note.pdf",
|
|
)
|
|
|
|
|
|
def test_adds_uploaded_file_record_to_session(
|
|
patched_deps: dict[str, MagicMock],
|
|
) -> None:
|
|
upload_file_to_s3_and_record(
|
|
bucket="test-bucket",
|
|
file_path="/tmp/report.pdf",
|
|
hubspot_listing_id="hs-001",
|
|
file_type=FileTypeEnum.ECMK_RD_SAP_SITE_NOTE,
|
|
)
|
|
|
|
patched_deps["session"].add.assert_called_once_with(
|
|
patched_deps["uploaded_file"]
|
|
)
|
|
patched_deps["session"].flush.assert_called_once()
|
|
|
|
|
|
def test_site_note_type_does_not_trigger_pdf_parsing(
|
|
patched_deps: dict[str, MagicMock],
|
|
) -> None:
|
|
# If parsing branch still existed, this would blow up without a
|
|
# parse_site_notes_pdf mock — test passes only when branch is absent.
|
|
result = upload_file_to_s3_and_record(
|
|
bucket="test-bucket",
|
|
file_path="/tmp/site_note.pdf",
|
|
hubspot_listing_id="hs-002",
|
|
file_type=FileTypeEnum.ECMK_RD_SAP_SITE_NOTE,
|
|
)
|
|
|
|
assert result == 42
|