diff --git a/src/application/files/__init__.py b/src/application/files/__init__.py new file mode 100644 index 0000000..2dc0a36 --- /dev/null +++ b/src/application/files/__init__.py @@ -0,0 +1,19 @@ +"""Source-file upload and status use cases (ADR-0008, ADR-0009, ADR-0017).""" + +from src.application.files.errors import FilesError, FileTooLargeError, InvalidUploadError +from src.application.files.models import UploadResult, ValidatedUpload +from src.application.files.status import FileStatusResult, get_file_status +from src.application.files.upload import upload_source_file +from src.application.files.validation import validate_and_hash_upload + +__all__ = [ + "FileStatusResult", + "FileTooLargeError", + "FilesError", + "InvalidUploadError", + "UploadResult", + "ValidatedUpload", + "get_file_status", + "upload_source_file", + "validate_and_hash_upload", +] diff --git a/src/application/files/errors.py b/src/application/files/errors.py new file mode 100644 index 0000000..b58631c --- /dev/null +++ b/src/application/files/errors.py @@ -0,0 +1,17 @@ +"""Upload-validation failures (ADR-0008). No HTTP knowledge here — +`src/api/errors.py` maps these to status codes. +""" + + +class FilesError(Exception): + """Base class for file-upload failures.""" + + +class InvalidUploadError(FilesError): + """Missing domain, empty file, or content that doesn't match its + declared extension. Maps to `400`. + """ + + +class FileTooLargeError(FilesError): + """The upload exceeds `INGESTION_MAX_UPLOAD_SIZE_MB`. Maps to `413`.""" diff --git a/src/application/files/models.py b/src/application/files/models.py new file mode 100644 index 0000000..18a041e --- /dev/null +++ b/src/application/files/models.py @@ -0,0 +1,26 @@ +"""Domain models for the upload use case (ADR-0008, ADR-0009).""" + +import uuid +from dataclasses import dataclass + + +@dataclass(frozen=True) +class ValidatedUpload: + """The result of extension/content validation, before any I/O.""" + + source_type: str + content_type: str + content_sha256: str + + +@dataclass(frozen=True) +class UploadResult: + """What `upload_source_file` returns; the route maps this to `FileUploadResponse`.""" + + file_id: uuid.UUID + ingestion_job_id: uuid.UUID + status: str + chunks_indexed: int + is_new_attempt: bool + """`False` when an identical active upload already succeeded and no new + ingestion attempt was made (route returns `200`, not `201`).""" diff --git a/src/application/files/status.py b/src/application/files/status.py new file mode 100644 index 0000000..3a56f94 --- /dev/null +++ b/src/application/files/status.py @@ -0,0 +1,52 @@ +"""`GET /v1/files/{file_id}` read model (ADR-0008, ADR-0009).""" + +import uuid +from dataclasses import dataclass + +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from src.infrastructure.postgres.repositories import ingestion_jobs as jobs_repo +from src.infrastructure.postgres.repositories import source_files as source_files_repo + + +@dataclass(frozen=True) +class FileStatusResult: + file_id: uuid.UUID + source_filename: str + domain: str + status: str + ingestion_job_id: uuid.UUID | None + ingestion_status: str | None + chunks_indexed: int + + +async def get_file_status( + sessionmaker: async_sessionmaker[AsyncSession], + *, + tenant_id: uuid.UUID, + source_file_id: uuid.UUID, +) -> FileStatusResult | None: + """Returns `None` when the file doesn't exist under this tenant — the + route maps that to `404`, never `403` (ADR-0016: cross-tenant access + returns 404). + """ + async with sessionmaker() as session: + source_file = await source_files_repo.get_by_id( + session, tenant_id=tenant_id, source_file_id=source_file_id + ) + if source_file is None: + return None + + latest_job = await jobs_repo.get_latest_for_source_file( + session, tenant_id=tenant_id, source_file_id=source_file.id + ) + + return FileStatusResult( + file_id=source_file.id, + source_filename=source_file.source_filename, + domain=source_file.domain, + status=source_file.status, + ingestion_job_id=latest_job.id if latest_job else None, + ingestion_status=latest_job.status if latest_job else None, + chunks_indexed=latest_job.points_created if latest_job else 0, + ) diff --git a/src/application/files/storage_keys.py b/src/application/files/storage_keys.py new file mode 100644 index 0000000..d332de2 --- /dev/null +++ b/src/application/files/storage_keys.py @@ -0,0 +1,11 @@ +"""Object-storage key derivation (ADR-0013). + +Object keys are internal identifiers, never the caller-supplied filename. +Pure and synchronous. +""" + +import uuid + + +def source_file_object_key(tenant_id: uuid.UUID, source_file_id: uuid.UUID) -> str: + return f"tenants/{tenant_id}/source-files/{source_file_id}/original" diff --git a/src/application/files/upload.py b/src/application/files/upload.py new file mode 100644 index 0000000..b10c419 --- /dev/null +++ b/src/application/files/upload.py @@ -0,0 +1,205 @@ +"""`POST /v1/files` orchestration: the ADR-0017 three-phase upload. + +This service owns two separate short-lived sessions/transactions rather than +one request-scoped session, because the request is two units of work +(ADR-0012, ADR-0017): + + txn A (short): source_files [+ ingestion_jobs(status='running')], commit + no txn: store bytes in MinIO + txn B (short): ingestion_jobs -> succeeded/failed, append event, commit + +No Postgres session is open during the MinIO write. A storage failure between +txn A and txn B still leaves a durable, inspectable `failed` job — never a +job stuck in `running`. + +Parsing/chunking/Qdrant indexing are Phase 4/5 work, not implemented here: +this phase stores bytes only, so a successful job reports `chunks_indexed=0`. +""" + +import uuid + +import structlog +from anyio import CapacityLimiter, to_thread +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from src.application.auth.context import AuthContext +from src.application.files.errors import InvalidUploadError +from src.application.files.models import UploadResult +from src.application.files.storage_keys import source_file_object_key +from src.application.files.validation import validate_and_hash_upload +from src.application.ports.object_storage import ObjectStorage +from src.infrastructure.postgres.repositories import ingestion_jobs as jobs_repo +from src.infrastructure.postgres.repositories import source_files as source_files_repo + +logger = structlog.get_logger(__name__) + + +async def _mark_job_failed( + sessionmaker: async_sessionmaker[AsyncSession], + *, + tenant_id: uuid.UUID, + ingestion_job_id: uuid.UUID, + error_code: str, + error_message: str, +) -> None: + async with sessionmaker() as session: + job = await jobs_repo.mark_terminal( + session, + tenant_id=tenant_id, + ingestion_job_id=ingestion_job_id, + status="failed", + error_code=error_code, + error_message=error_message, + ) + if job is not None: + jobs_repo.append_event( + session, + tenant_id=tenant_id, + ingestion_job_id=ingestion_job_id, + level="error", + stage="received", + message=error_message, + ) + await session.commit() + + +async def upload_source_file( + *, + sessionmaker: async_sessionmaker[AsyncSession], + storage: ObjectStorage, + auth: AuthContext, + domain: str, + filename: str, + data: bytes, + max_upload_size_bytes: int, + chunking_strategy: str, + validation_limiter: CapacityLimiter, +) -> UploadResult: + domain = domain.strip() + if not domain: + raise InvalidUploadError("domain is required") + + validated = await to_thread.run_sync( + lambda: validate_and_hash_upload( + filename=filename, data=data, max_size_bytes=max_upload_size_bytes + ), + limiter=validation_limiter, + ) + + async with sessionmaker() as session: + existing = await source_files_repo.find_active_by_content_hash( + session, + tenant_id=auth.tenant_id, + domain=domain, + content_sha256=validated.content_sha256, + ) + + if existing is not None: + latest_job = await jobs_repo.get_latest_for_source_file( + session, tenant_id=auth.tenant_id, source_file_id=existing.id + ) + if latest_job is not None and latest_job.status == "succeeded": + logger.info( + "files.upload.duplicate", + tenant_id=str(auth.tenant_id), + file_id=str(existing.id), + ) + return UploadResult( + file_id=existing.id, + ingestion_job_id=latest_job.id, + status=latest_job.status, + chunks_indexed=latest_job.points_created, + is_new_attempt=False, + ) + source_file_id = existing.id + object_key = existing.storage_uri or source_file_object_key( + auth.tenant_id, source_file_id + ) + else: + source_file_id = uuid.uuid4() + object_key = source_file_object_key(auth.tenant_id, source_file_id) + source_files_repo.create( + session, + source_file_id=source_file_id, + tenant_id=auth.tenant_id, + domain=domain, + source_filename=filename, + source_type=validated.source_type, + content_sha256=validated.content_sha256, + byte_size=len(data), + storage_uri=object_key, + created_by_api_key_id=auth.api_key_id, + ) + # `ingestion_jobs.source_file_id` FKs to this row; flush so the + # insert below sees it, since the two mapped classes carry no + # ORM relationship for the unit of work to order by itself. + await session.flush() + + job = jobs_repo.create_running( + session, + tenant_id=auth.tenant_id, + source_file_id=source_file_id, + requested_by_api_key_id=auth.api_key_id, + chunking_strategy=chunking_strategy, + ) + jobs_repo.append_event( + session, + tenant_id=auth.tenant_id, + ingestion_job_id=job.id, + level="info", + stage="received", + message="upload accepted, storing object", + ) + await session.commit() + ingestion_job_id = job.id + + # Phase 2: no Postgres session open across this work (ADR-0017). + try: + await storage.put_object(key=object_key, data=data, content_type=validated.content_type) + except Exception as exc: + logger.warning( + "files.upload.storage_failed", + tenant_id=str(auth.tenant_id), + file_id=str(source_file_id), + ingestion_job_id=str(ingestion_job_id), + ) + await _mark_job_failed( + sessionmaker, + tenant_id=auth.tenant_id, + ingestion_job_id=ingestion_job_id, + error_code="storage_upload_failed", + error_message=f"failed to store object: {exc}", + ) + raise + + async with sessionmaker() as session: + await jobs_repo.mark_terminal( + session, + tenant_id=auth.tenant_id, + ingestion_job_id=ingestion_job_id, + status="succeeded", + points_created=0, + ) + jobs_repo.append_event( + session, + tenant_id=auth.tenant_id, + ingestion_job_id=ingestion_job_id, + level="info", + stage="completed", + message="object stored; parsing/embedding/indexing not yet implemented", + ) + await session.commit() + + logger.info( + "files.upload.succeeded", + tenant_id=str(auth.tenant_id), + file_id=str(source_file_id), + ingestion_job_id=str(ingestion_job_id), + ) + return UploadResult( + file_id=source_file_id, + ingestion_job_id=ingestion_job_id, + status="succeeded", + chunks_indexed=0, + is_new_attempt=True, + ) diff --git a/src/application/files/validation.py b/src/application/files/validation.py new file mode 100644 index 0000000..ff522d7 --- /dev/null +++ b/src/application/files/validation.py @@ -0,0 +1,55 @@ +"""Upload extension/content-type/size validation (ADR-0008). + +Pure and synchronous: no I/O. `content_sha256` computation lives here too — +hashing is blocking CPU work (ADR-0017), so the caller runs this whole +function through `anyio.to_thread.run_sync` with the ingestion +`CapacityLimiter`, the same rule applied to parsing/chunking. +""" + +import hashlib + +from src.application.files.errors import FileTooLargeError, InvalidUploadError +from src.application.files.models import ValidatedUpload +from src.application.ingestion.errors import UnsupportedSourceTypeError + +_CONTENT_TYPES = { + "csv": "text/csv", + "xlsx": "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", + "docx": "application/vnd.openxmlformats-officedocument.wordprocessingml.document", +} +_OOXML_MAGIC = b"PK\x03\x04" + + +def _source_type_from_filename(filename: str) -> str: + suffix = filename.rsplit(".", 1)[-1].lower() if "." in filename else "" + if suffix == "doc": + raise UnsupportedSourceTypeError( + "legacy .doc is not ingestible until an out-of-process conversion " + "service exists (ADR-0018)" + ) + if suffix not in _CONTENT_TYPES: + raise UnsupportedSourceTypeError(f"'.{suffix}' is not an ingestible file type") + return suffix + + +def validate_and_hash_upload(*, filename: str, data: bytes, max_size_bytes: int) -> ValidatedUpload: + source_type = _source_type_from_filename(filename) + + if not data: + raise InvalidUploadError("uploaded file is empty") + if len(data) > max_size_bytes: + raise FileTooLargeError( + f"upload is {len(data)} bytes, over the {max_size_bytes}-byte limit" + ) + + is_ooxml = data[:4] == _OOXML_MAGIC + if source_type in ("docx", "xlsx") and not is_ooxml: + raise InvalidUploadError(f"content does not match the declared .{source_type} extension") + if source_type == "csv" and is_ooxml: + raise InvalidUploadError("content does not match the declared .csv extension") + + return ValidatedUpload( + source_type=source_type, + content_type=_CONTENT_TYPES[source_type], + content_sha256=hashlib.sha256(data).hexdigest(), + ) diff --git a/src/application/ports/__init__.py b/src/application/ports/__init__.py new file mode 100644 index 0000000..d3ad233 --- /dev/null +++ b/src/application/ports/__init__.py @@ -0,0 +1,7 @@ +"""Narrow contracts for external side effects (ADR-0015). + +Ports exist for external side effects/persistence that need a swappable or +fake-able boundary — not as a blanket wrapper around every database access. +`object_storage.py` is one: MinIO is a real external system with its own +failure modes, and ADR-0016 requires a hand-written fake for it in tests. +""" diff --git a/src/application/ports/object_storage.py b/src/application/ports/object_storage.py new file mode 100644 index 0000000..2192ba5 --- /dev/null +++ b/src/application/ports/object_storage.py @@ -0,0 +1,14 @@ +"""The object-storage port (ADR-0013). + +`src/infrastructure/minio/storage.py` is the production adapter; tests use a +hand-written fake (ADR-0016). Application code depends on this Protocol, not +on the `minio` SDK. +""" + +from typing import Protocol + + +class ObjectStorage(Protocol): + async def put_object(self, *, key: str, data: bytes, content_type: str) -> None: + """Store `data` privately under `key`. Overwrites an existing object.""" + ... diff --git a/src/infrastructure/minio/storage.py b/src/infrastructure/minio/storage.py new file mode 100644 index 0000000..d887e83 --- /dev/null +++ b/src/infrastructure/minio/storage.py @@ -0,0 +1,33 @@ +"""MinIO adapter for the `ObjectStorage` port (ADR-0013, ADR-0017). + +The `minio` SDK is synchronous, so every call runs through +`anyio.to_thread.run_sync` bounded by the ingestion `CapacityLimiter` — the +same rule ADR-0017 applies to parsing/chunking. Calling the SDK directly from +`async def` would block every concurrent request in the process. +""" + +import io +from functools import partial + +from anyio import CapacityLimiter, to_thread +from minio import Minio + + +class MinioObjectStorage: + def __init__(self, client: Minio, *, bucket: str, limiter: CapacityLimiter) -> None: + self._client = client + self._bucket = bucket + self._limiter = limiter + + async def put_object(self, *, key: str, data: bytes, content_type: str) -> None: + await to_thread.run_sync( + partial( + self._client.put_object, + self._bucket, + key, + io.BytesIO(data), + length=len(data), + content_type=content_type, + ), + limiter=self._limiter, + ) diff --git a/src/infrastructure/postgres/repositories/__init__.py b/src/infrastructure/postgres/repositories/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/infrastructure/postgres/repositories/api_keys.py b/src/infrastructure/postgres/repositories/api_keys.py new file mode 100644 index 0000000..54a0516 --- /dev/null +++ b/src/infrastructure/postgres/repositories/api_keys.py @@ -0,0 +1,17 @@ +"""API-key lookups (ADR-0008, ADR-0009). + +Plain functions over an `AsyncSession` the caller owns. No function here +commits, rolls back, or closes the session (ADR-0012). Secret comparison +happens in `src/application/auth`, not here — this module only fetches rows +by their non-secret `key_prefix`. +""" + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from src.infrastructure.postgres.models.api_key import ApiKey + + +async def get_by_prefix(session: AsyncSession, key_prefix: str) -> ApiKey | None: + result = await session.execute(select(ApiKey).where(ApiKey.key_prefix == key_prefix)) + return result.scalar_one_or_none() diff --git a/src/infrastructure/postgres/repositories/ingestion_jobs.py b/src/infrastructure/postgres/repositories/ingestion_jobs.py new file mode 100644 index 0000000..022a576 --- /dev/null +++ b/src/infrastructure/postgres/repositories/ingestion_jobs.py @@ -0,0 +1,119 @@ +"""`ingestion_jobs`/`ingestion_job_events` persistence (ADR-0009, ADR-0017). + +Plain functions over an `AsyncSession` the caller owns. No function here +commits, rolls back, or closes the session (ADR-0012) — the two-transaction +shape in `src/application/files/upload.py` depends on that. Every read is +tenant-scoped by a required `tenant_id` argument. +""" + +import uuid +from datetime import UTC, datetime + +from sqlalchemy import desc, select +from sqlalchemy.ext.asyncio import AsyncSession + +from src.infrastructure.postgres.models.ingestion_job import IngestionJob +from src.infrastructure.postgres.models.ingestion_job_event import IngestionJobEvent + + +async def get_by_id( + session: AsyncSession, *, tenant_id: uuid.UUID, ingestion_job_id: uuid.UUID +) -> IngestionJob | None: + result = await session.execute( + select(IngestionJob).where( + IngestionJob.id == ingestion_job_id, IngestionJob.tenant_id == tenant_id + ) + ) + return result.scalar_one_or_none() + + +async def get_latest_for_source_file( + session: AsyncSession, *, tenant_id: uuid.UUID, source_file_id: uuid.UUID +) -> IngestionJob | None: + result = await session.execute( + select(IngestionJob) + .where( + IngestionJob.tenant_id == tenant_id, + IngestionJob.source_file_id == source_file_id, + ) + .order_by(desc(IngestionJob.created_at)) + .limit(1) + ) + return result.scalar_one_or_none() + + +def create_running( + session: AsyncSession, + *, + tenant_id: uuid.UUID, + source_file_id: uuid.UUID, + requested_by_api_key_id: uuid.UUID | None, + chunking_strategy: str, +) -> IngestionJob: + job = IngestionJob( + id=uuid.uuid4(), + tenant_id=tenant_id, + source_file_id=source_file_id, + requested_by_api_key_id=requested_by_api_key_id, + status="running", + chunking_strategy=chunking_strategy, + started_at=datetime.now(UTC), + ) + session.add(job) + return job + + +async def mark_terminal( + session: AsyncSession, + *, + tenant_id: uuid.UUID, + ingestion_job_id: uuid.UUID, + status: str, + points_created: int = 0, + points_updated: int = 0, + points_soft_deleted: int = 0, + points_skipped: int = 0, + error_code: str | None = None, + error_message: str | None = None, +) -> IngestionJob | None: + """Move a job from `running` to a terminal status. + + Never reads/writes a job whose current status is already terminal — a + terminal job must not transition back to `running` or to a different + terminal status (ADR-0017). + """ + job = await get_by_id(session, tenant_id=tenant_id, ingestion_job_id=ingestion_job_id) + if job is None or job.status != "running": + return None + job.status = status + job.completed_at = datetime.now(UTC) + job.points_created = points_created + job.points_updated = points_updated + job.points_soft_deleted = points_soft_deleted + job.points_skipped = points_skipped + job.error_code = error_code + job.error_message = error_message + return job + + +def append_event( + session: AsyncSession, + *, + tenant_id: uuid.UUID, + ingestion_job_id: uuid.UUID, + level: str, + stage: str, + message: str, + details: dict[str, object] | None = None, +) -> IngestionJobEvent: + event = IngestionJobEvent( + id=uuid.uuid4(), + tenant_id=tenant_id, + ingestion_job_id=ingestion_job_id, + level=level, + stage=stage, + message=message, + details=details or {}, + ) + session.add(event) + return event diff --git a/src/infrastructure/postgres/repositories/source_files.py b/src/infrastructure/postgres/repositories/source_files.py new file mode 100644 index 0000000..d182acc --- /dev/null +++ b/src/infrastructure/postgres/repositories/source_files.py @@ -0,0 +1,69 @@ +"""`source_files` persistence (ADR-0009). + +Plain functions over an `AsyncSession` the caller owns. No function here +commits, rolls back, or closes the session (ADR-0012). Every read is +tenant-scoped by a required `tenant_id` argument, so a missing filter is a +signature error rather than a cross-tenant leak. +""" + +import uuid + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from src.infrastructure.postgres.models.source_file import SourceFile + + +async def get_by_id( + session: AsyncSession, *, tenant_id: uuid.UUID, source_file_id: uuid.UUID +) -> SourceFile | None: + result = await session.execute( + select(SourceFile).where(SourceFile.id == source_file_id, SourceFile.tenant_id == tenant_id) + ) + return result.scalar_one_or_none() + + +async def find_active_by_content_hash( + session: AsyncSession, *, tenant_id: uuid.UUID, domain: str, content_sha256: str +) -> SourceFile | None: + result = await session.execute( + select(SourceFile).where( + SourceFile.tenant_id == tenant_id, + SourceFile.domain == domain, + SourceFile.content_sha256 == content_sha256, + SourceFile.status == "active", + ) + ) + return result.scalar_one_or_none() + + +def create( + session: AsyncSession, + *, + source_file_id: uuid.UUID, + tenant_id: uuid.UUID, + domain: str, + source_filename: str, + source_type: str, + content_sha256: str, + byte_size: int, + storage_uri: str, + created_by_api_key_id: uuid.UUID | None, +) -> SourceFile: + """`source_file_id` is caller-generated: the upload service derives the + MinIO object key from it before this row exists, so the id has to be + chosen up front rather than assigned by the database. + """ + source_file = SourceFile( + id=source_file_id, + tenant_id=tenant_id, + domain=domain, + source_filename=source_filename, + source_type=source_type, + content_sha256=content_sha256, + byte_size=byte_size, + storage_uri=storage_uri, + created_by_api_key_id=created_by_api_key_id, + ) + session.add(source_file) + return source_file diff --git a/src/infrastructure/postgres/repositories/tenants.py b/src/infrastructure/postgres/repositories/tenants.py new file mode 100644 index 0000000..3e1f738 --- /dev/null +++ b/src/infrastructure/postgres/repositories/tenants.py @@ -0,0 +1,15 @@ +"""Tenant lookups (ADR-0009). + +Plain functions over an `AsyncSession` the caller owns. No function here +commits, rolls back, or closes the session (ADR-0012). +""" + +import uuid + +from sqlalchemy.ext.asyncio import AsyncSession + +from src.infrastructure.postgres.models.tenant import Tenant + + +async def get_by_id(session: AsyncSession, tenant_id: uuid.UUID) -> Tenant | None: + return await session.get(Tenant, tenant_id) diff --git a/tests/fakes.py b/tests/fakes.py index b9e09a6..5316314 100644 --- a/tests/fakes.py +++ b/tests/fakes.py @@ -1,5 +1,17 @@ -"""Hand-written fakes for narrow application-owned ports. +"""Hand-written fakes for narrow application-owned ports (ADR-0016).""" -No application ports exist yet (Phase 1 only wires infrastructure client -lifecycle). Fakes are added here as ports are introduced in later phases. -""" +from dataclasses import dataclass, field + + +@dataclass +class FakeObjectStorage: + """In-memory `ObjectStorage`. `fail_next` simulates one upload failure.""" + + objects: dict[str, bytes] = field(default_factory=dict) + fail_next: bool = False + + async def put_object(self, *, key: str, data: bytes, content_type: str) -> None: + if self.fail_next: + self.fail_next = False + raise OSError("simulated object storage failure") + self.objects[key] = data diff --git a/tests/unit/application/files/__init__.py b/tests/unit/application/files/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/unit/application/files/test_storage_keys.py b/tests/unit/application/files/test_storage_keys.py new file mode 100644 index 0000000..9e5446c --- /dev/null +++ b/tests/unit/application/files/test_storage_keys.py @@ -0,0 +1,25 @@ +import uuid + +import pytest + +from src.application.files.storage_keys import source_file_object_key + +pytestmark = pytest.mark.unit + + +def test_source_file_object_key_is_server_derived_not_filename() -> None: + tenant_id = uuid.uuid4() + source_file_id = uuid.uuid4() + + key = source_file_object_key(tenant_id, source_file_id) + + assert key == f"tenants/{tenant_id}/source-files/{source_file_id}/original" + + +def test_source_file_object_key_differs_per_tenant() -> None: + source_file_id = uuid.uuid4() + + key_a = source_file_object_key(uuid.uuid4(), source_file_id) + key_b = source_file_object_key(uuid.uuid4(), source_file_id) + + assert key_a != key_b diff --git a/tests/unit/application/files/test_validation.py b/tests/unit/application/files/test_validation.py new file mode 100644 index 0000000..672b969 --- /dev/null +++ b/tests/unit/application/files/test_validation.py @@ -0,0 +1,58 @@ +import pytest + +from src.application.files.errors import FileTooLargeError, InvalidUploadError +from src.application.files.validation import validate_and_hash_upload +from src.application.ingestion.errors import UnsupportedSourceTypeError + +pytestmark = pytest.mark.unit + +_DOCX_HEADER = b"PK\x03\x04" + b"\x00" * 20 +_CSV_BYTES = b"name,value\nfirst,1\n" + + +def test_validate_and_hash_upload_accepts_matching_csv() -> None: + result = validate_and_hash_upload(filename="report.csv", data=_CSV_BYTES, max_size_bytes=1_000) + + assert result.source_type == "csv" + assert result.content_type == "text/csv" + assert len(result.content_sha256) == 64 + + +def test_validate_and_hash_upload_accepts_matching_docx() -> None: + result = validate_and_hash_upload( + filename="report.docx", data=_DOCX_HEADER, max_size_bytes=1_000 + ) + + assert result.source_type == "docx" + + +def test_validate_and_hash_upload_rejects_doc_as_unsupported() -> None: + with pytest.raises(UnsupportedSourceTypeError): + validate_and_hash_upload(filename="legacy.doc", data=_CSV_BYTES, max_size_bytes=1_000) + + +def test_validate_and_hash_upload_rejects_unknown_extension() -> None: + with pytest.raises(UnsupportedSourceTypeError): + validate_and_hash_upload(filename="report.pdf", data=_CSV_BYTES, max_size_bytes=1_000) + + +def test_validate_and_hash_upload_rejects_empty_file() -> None: + with pytest.raises(InvalidUploadError): + validate_and_hash_upload(filename="report.csv", data=b"", max_size_bytes=1_000) + + +def test_validate_and_hash_upload_rejects_oversized_file() -> None: + with pytest.raises(FileTooLargeError): + validate_and_hash_upload(filename="report.csv", data=_CSV_BYTES, max_size_bytes=4) + + +def test_validate_and_hash_upload_rejects_spoofed_docx_extension() -> None: + """Content is really CSV text, but the filename claims `.docx`.""" + with pytest.raises(InvalidUploadError): + validate_and_hash_upload(filename="report.docx", data=_CSV_BYTES, max_size_bytes=1_000) + + +def test_validate_and_hash_upload_rejects_spoofed_csv_extension() -> None: + """Content is really an OOXML zip, but the filename claims `.csv`.""" + with pytest.raises(InvalidUploadError): + validate_and_hash_upload(filename="report.csv", data=_DOCX_HEADER, max_size_bytes=1_000)