diff --git a/orchestration/bulk_document_download_orchestrator.py b/orchestration/bulk_document_download_orchestrator.py new file mode 100644 index 000000000..01d65a44a --- /dev/null +++ b/orchestration/bulk_document_download_orchestrator.py @@ -0,0 +1,81 @@ +"""The Bulk Document Download orchestrator (ADR-0060). + +Ties the pieces together for one run: gather a property set's uploaded files +(by `landlord_property_id`), enrich each with its display address, lay them out +with the Download Package plan, stream the chosen files into a single ZIP, +upload it, mint a presigned URL, and email the requester. Best-effort — skipped +rows are reported, not fatal; a wholly-empty selection is a recorded failure +(ADR-0060). +""" + +from __future__ import annotations + +import io +import zipfile +from dataclasses import dataclass +from datetime import datetime +from typing import Optional, Protocol, Sequence, cast + +from domain.bulk_document_download.package_plan import ( + ResolvedDocument, + SkippedDocument, + plan_download_package, +) +from domain.tasks.subtasks import SubTaskFailure +from infrastructure.postgres.uploaded_file_table import UploadedFile +from infrastructure.s3.s3_client import S3Client +from repositories.email.email_sender import EmailSender + + +class UploadedFileReader(Protocol): + def by_landlord_property_ids( + self, landlord_property_ids: Sequence[str] + ) -> list[UploadedFile]: ... + + +class AddressResolver(Protocol): + def address_for(self, landlord_property_id: str) -> str: ... + + +class DocumentBytesReader(Protocol): + def read(self, bucket: str, key: str) -> bytes: ... + + +@dataclass(frozen=True) +class DownloadPackageResult: + """What a completed run reports (lands in `sub_task.outputs`, ADR-0060).""" + + presigned_url: str + package_s3_key: str + included: int + skipped: tuple[SkippedDocument, ...] + + +class BulkDocumentDownloadOrchestrator: + def __init__( + self, + *, + uploaded_files: UploadedFileReader, + addresses: AddressResolver, + documents: DocumentBytesReader, + packages: S3Client, + email: EmailSender, + url_ttl_seconds: int, + package_key_prefix: str = "bulk-downloads", + ) -> None: + self._uploaded_files = uploaded_files + self._addresses = addresses + self._documents = documents + self._packages = packages + self._email = email + self._url_ttl_seconds = url_ttl_seconds + self._package_key_prefix = package_key_prefix + + def run( + self, + *, + landlord_property_ids: Sequence[str], + recipient_email: str, + package_name: str, + ) -> DownloadPackageResult: + raise NotImplementedError diff --git a/tests/orchestration/test_bulk_document_download_orchestrator.py b/tests/orchestration/test_bulk_document_download_orchestrator.py new file mode 100644 index 000000000..b1b48830a --- /dev/null +++ b/tests/orchestration/test_bulk_document_download_orchestrator.py @@ -0,0 +1,113 @@ +"""The Bulk Document Download orchestrator (ADR-0060): gather a property set's +files, lay them out, zip + upload, presign, and email the requester.""" + +from __future__ import annotations + +import io +import zipfile +from collections.abc import Iterator +from datetime import datetime, timezone + +import pytest +from moto import mock_aws + +from infrastructure.postgres.uploaded_file_table import FileTypeEnum, UploadedFile +from infrastructure.s3.s3_client import S3Client +from orchestration.bulk_document_download_orchestrator import ( + BulkDocumentDownloadOrchestrator, +) +from tests.infrastructure import make_boto_client + +DATA_BUCKET = "data-bucket" + + +class _FakeUploadedFiles: + def __init__(self, files: list[UploadedFile]) -> None: + self._files = files + + def by_landlord_property_ids( + self, landlord_property_ids: object + ) -> list[UploadedFile]: + wanted = set(landlord_property_ids) # type: ignore[call-overload] + return [f for f in self._files if f.landlord_property_id in wanted] + + +class _FakeAddresses: + def __init__(self, mapping: dict[str, str]) -> None: + self._mapping = mapping + + def address_for(self, landlord_property_id: str) -> str: + return self._mapping[landlord_property_id] + + +class _FakeDocuments: + def __init__(self, blobs: dict[tuple[str, str], bytes]) -> None: + self._blobs = blobs + + def read(self, bucket: str, key: str) -> bytes: + return self._blobs[(bucket, key)] + + +class _RecordingEmail: + def __init__(self) -> None: + self.sent: list[tuple[str, str, str]] = [] + + def send(self, *, to: str, subject: str, body: str) -> None: + self.sent.append((to, subject, body)) + + +def _uploaded_file( + landlord_property_id: str, + s3_file_bucket: str, + s3_file_key: str, + file_type: FileTypeEnum = FileTypeEnum.SITE_NOTE, +) -> UploadedFile: + return UploadedFile( + s3_file_bucket=s3_file_bucket, + s3_file_key=s3_file_key, + s3_upload_timestamp=datetime(2026, 1, 1, tzinfo=timezone.utc), + landlord_property_id=landlord_property_id, + file_type=file_type.value, + ) + + +@pytest.fixture +def packages() -> Iterator[S3Client]: + with mock_aws(): + boto_client = make_boto_client("s3") + boto_client.create_bucket(Bucket=DATA_BUCKET) + yield S3Client(boto_client, DATA_BUCKET) + + +def test_builds_uploads_and_emails_a_download_package(packages: S3Client) -> None: + # Arrange — one property with one site note. + files = [_uploaded_file("LP1", "src-bucket", "surveys/a.pdf")] + email = _RecordingEmail() + orchestrator = BulkDocumentDownloadOrchestrator( + uploaded_files=_FakeUploadedFiles(files), + addresses=_FakeAddresses({"LP1": "12 Oak Street"}), + documents=_FakeDocuments({("src-bucket", "surveys/a.pdf"): b"PDF-BYTES"}), + packages=packages, + email=email, + url_ttl_seconds=3600, + ) + + # Act + result = orchestrator.run( + landlord_property_ids=["LP1"], + recipient_email="user@example.com", + package_name="portfolio-42", + ) + + # Assert — one file packaged, at the planned path with the source bytes. + assert result.included == 1 + archive = packages.get_object(result.package_s3_key) + with zipfile.ZipFile(io.BytesIO(archive)) as zf: + assert zf.read("12 Oak Street (LP1)/site_note.pdf") == b"PDF-BYTES" + + # ...the presigned URL points at that object, and the requester is emailed it. + assert result.package_s3_key in result.presigned_url + assert len(email.sent) == 1 + to, _subject, body = email.sent[0] + assert to == "user@example.com" + assert result.presigned_url in body