diff --git a/tests/integration/minio/conftest.py b/tests/integration/minio/conftest.py new file mode 100644 index 0000000..e5354e5 --- /dev/null +++ b/tests/integration/minio/conftest.py @@ -0,0 +1,39 @@ +from collections.abc import Iterator + +import pytest +from testcontainers.community.minio import MinioContainer + +from src.config import MinioSettings +from src.infrastructure.minio.client import create_client + +_BUCKET = "test-source-files" + + +@pytest.fixture(scope="session") +def minio_container() -> Iterator[MinioContainer]: + with MinioContainer() as container: + yield container + + +@pytest.fixture(scope="session") +def minio_settings(minio_container: MinioContainer) -> MinioSettings: + config = minio_container.get_config() + # Pinned to IPv4 for the same reason as postgres_url (see + # tests/integration/postgres/conftest.py): `localhost` resolves to `::1` + # first, but Docker only publishes the mapped port on IPv4, so the + # connection hangs instead of failing. + endpoint = config["endpoint"].replace("localhost:", "127.0.0.1:") + return MinioSettings( + endpoint=endpoint, + access_key=config["access_key"], + secret_key=config["secret_key"], + secure=False, + bucket=_BUCKET, + ) + + +@pytest.fixture(scope="session", autouse=True) +def _ensure_bucket(minio_settings: MinioSettings) -> None: + client = create_client(minio_settings) + if not client.bucket_exists(minio_settings.bucket): + client.make_bucket(minio_settings.bucket) diff --git a/tests/integration/minio/test_storage.py b/tests/integration/minio/test_storage.py new file mode 100644 index 0000000..3426305 --- /dev/null +++ b/tests/integration/minio/test_storage.py @@ -0,0 +1,39 @@ +import uuid + +import pytest +from anyio import CapacityLimiter + +from src.application.files.storage_keys import source_file_object_key +from src.config import MinioSettings +from src.infrastructure.minio.client import create_client +from src.infrastructure.minio.storage import MinioObjectStorage + +pytestmark = [pytest.mark.integration, pytest.mark.minio, pytest.mark.asyncio] + + +async def test_put_object_stores_bytes_under_server_derived_key( + minio_settings: MinioSettings, +) -> None: + client = create_client(minio_settings) + storage = MinioObjectStorage(client, bucket=minio_settings.bucket, limiter=CapacityLimiter(2)) + key = source_file_object_key(uuid.uuid4(), uuid.uuid4()) + data = b"tenant-scoped, id-addressed bytes" + + await storage.put_object(key=key, data=data, content_type="text/csv") + + stored = client.get_object(minio_settings.bucket, key).read() + assert stored == data + + +async def test_put_object_overwrites_existing_object_at_same_key( + minio_settings: MinioSettings, +) -> None: + client = create_client(minio_settings) + storage = MinioObjectStorage(client, bucket=minio_settings.bucket, limiter=CapacityLimiter(2)) + key = source_file_object_key(uuid.uuid4(), uuid.uuid4()) + + await storage.put_object(key=key, data=b"first", content_type="text/csv") + await storage.put_object(key=key, data=b"second", content_type="text/csv") + + stored = client.get_object(minio_settings.bucket, key).read() + assert stored == b"second" diff --git a/tests/integration/postgres/test_auth_service.py b/tests/integration/postgres/test_auth_service.py new file mode 100644 index 0000000..d4e9319 --- /dev/null +++ b/tests/integration/postgres/test_auth_service.py @@ -0,0 +1,65 @@ +import pytest +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from src.application.auth.errors import InvalidApiKeyError, TenantInactiveError +from src.application.auth.service import resolve_auth_context +from tests.support.factories import create_api_key, create_tenant + +pytestmark = [ + pytest.mark.integration, + pytest.mark.postgres, + pytest.mark.asyncio(loop_scope="session"), +] + + +async def test_resolve_auth_context_accepts_valid_key( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + tenant = await create_tenant(db_session) + _, full_key = await create_api_key(db_session, tenant=tenant, scopes=["files:write"]) + await db_session.commit() + + auth = await resolve_auth_context(db_sessionmaker, full_key) + + assert auth.tenant_id == tenant.id + assert auth.has_scope("files:write") + + +async def test_resolve_auth_context_rejects_wrong_secret( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + tenant = await create_tenant(db_session) + api_key, _ = await create_api_key(db_session, tenant=tenant) + await db_session.commit() + + with pytest.raises(InvalidApiKeyError): + await resolve_auth_context(db_sessionmaker, f"sk_{api_key.key_prefix}_wrong-secret") + + +async def test_resolve_auth_context_rejects_unknown_prefix( + db_sessionmaker: async_sessionmaker[AsyncSession], +) -> None: + with pytest.raises(InvalidApiKeyError): + await resolve_auth_context(db_sessionmaker, "sk_doesnotexist_secret") + + +async def test_resolve_auth_context_rejects_revoked_key( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + tenant = await create_tenant(db_session) + _, full_key = await create_api_key(db_session, tenant=tenant, status="revoked") + await db_session.commit() + + with pytest.raises(InvalidApiKeyError): + await resolve_auth_context(db_sessionmaker, full_key) + + +async def test_resolve_auth_context_rejects_suspended_tenant( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + tenant = await create_tenant(db_session, status="suspended") + _, full_key = await create_api_key(db_session, tenant=tenant) + await db_session.commit() + + with pytest.raises(TenantInactiveError): + await resolve_auth_context(db_sessionmaker, full_key) diff --git a/tests/integration/postgres/test_repositories.py b/tests/integration/postgres/test_repositories.py new file mode 100644 index 0000000..04332eb --- /dev/null +++ b/tests/integration/postgres/test_repositories.py @@ -0,0 +1,113 @@ +import uuid + +import pytest +from sqlalchemy.ext.asyncio import AsyncSession + +from src.infrastructure.postgres.repositories import ingestion_jobs as jobs_repo +from src.infrastructure.postgres.repositories import source_files as source_files_repo +from tests.support.factories import create_tenant + +pytestmark = [ + pytest.mark.integration, + pytest.mark.postgres, + pytest.mark.asyncio(loop_scope="session"), +] + + +async def test_source_files_get_by_id_is_tenant_scoped(db_session: AsyncSession) -> None: + owner = await create_tenant(db_session) + other = await create_tenant(db_session) + source_file_id = uuid.uuid4() + source_files_repo.create( + db_session, + source_file_id=source_file_id, + tenant_id=owner.id, + domain="general", + source_filename="report.csv", + source_type="csv", + content_sha256="a" * 64, + byte_size=10, + storage_uri="tenants/x/source-files/y/original", + created_by_api_key_id=None, + ) + await db_session.commit() + + found_for_owner = await source_files_repo.get_by_id( + db_session, tenant_id=owner.id, source_file_id=source_file_id + ) + found_for_other = await source_files_repo.get_by_id( + db_session, tenant_id=other.id, source_file_id=source_file_id + ) + + assert found_for_owner is not None + assert found_for_other is None + + +async def test_source_files_find_active_by_content_hash_matches_tenant_domain_hash( + db_session: AsyncSession, +) -> None: + tenant = await create_tenant(db_session) + source_files_repo.create( + db_session, + source_file_id=uuid.uuid4(), + tenant_id=tenant.id, + domain="general", + source_filename="report.csv", + source_type="csv", + content_sha256="b" * 64, + byte_size=10, + storage_uri="tenants/x/source-files/y/original", + created_by_api_key_id=None, + ) + await db_session.commit() + + found = await source_files_repo.find_active_by_content_hash( + db_session, tenant_id=tenant.id, domain="general", content_sha256="b" * 64 + ) + not_found_other_domain = await source_files_repo.find_active_by_content_hash( + db_session, tenant_id=tenant.id, domain="other", content_sha256="b" * 64 + ) + + assert found is not None + assert not_found_other_domain is None + + +async def test_ingestion_jobs_mark_terminal_rejects_non_running_job( + db_session: AsyncSession, +) -> None: + """A job already in a terminal state must not be re-marked (ADR-0017).""" + tenant = await create_tenant(db_session) + source_file_id = uuid.uuid4() + source_files_repo.create( + db_session, + source_file_id=source_file_id, + tenant_id=tenant.id, + domain="general", + source_filename="report.csv", + source_type="csv", + content_sha256="c" * 64, + byte_size=10, + storage_uri="tenants/x/source-files/y/original", + created_by_api_key_id=None, + ) + await db_session.flush() + job = jobs_repo.create_running( + db_session, + tenant_id=tenant.id, + source_file_id=source_file_id, + requested_by_api_key_id=None, + chunking_strategy="fixed_size", + ) + await db_session.commit() + + first = await jobs_repo.mark_terminal( + db_session, tenant_id=tenant.id, ingestion_job_id=job.id, status="succeeded" + ) + await db_session.commit() + second = await jobs_repo.mark_terminal( + db_session, tenant_id=tenant.id, ingestion_job_id=job.id, status="failed" + ) + + assert first is not None + assert first.status == "succeeded" + assert second is None diff --git a/tests/integration/postgres/test_upload_service.py b/tests/integration/postgres/test_upload_service.py new file mode 100644 index 0000000..f18d1d6 --- /dev/null +++ b/tests/integration/postgres/test_upload_service.py @@ -0,0 +1,194 @@ +"""ADR-0017's two-transaction upload shape, exercised against real Postgres. + +Object storage is faked (`FakeObjectStorage`) -- it's a port, not the thing +under test here. `MinioObjectStorage` itself is covered in +`tests/integration/minio/`. +""" + +import pytest +from anyio import CapacityLimiter +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from src.application.auth.context import AuthContext +from src.application.files.upload import upload_source_file +from src.infrastructure.postgres.models.ingestion_job import IngestionJob +from tests.fakes import FakeObjectStorage +from tests.support.factories import create_api_key, create_tenant + +pytestmark = [ + pytest.mark.integration, + pytest.mark.postgres, + pytest.mark.asyncio(loop_scope="session"), +] + +_CSV_BYTES = b"name,value\nfirst,1\n" + + +async def _auth_for(db_session: AsyncSession) -> AuthContext: + tenant = await create_tenant(db_session) + api_key, _ = await create_api_key(db_session, tenant=tenant) + await db_session.commit() + return AuthContext( + tenant_id=tenant.id, + tenant_slug=tenant.slug, + api_key_id=api_key.id, + scopes=frozenset({"files:write"}), + actor_type="backend", + ) + + +async def test_upload_source_file_commits_running_job_before_storage_write( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + """After phase 1 commits, a *separate* session must already see the + running job -- proving txn A committed and released before phase 2 work, + per ADR-0017. + """ + auth = await _auth_for(db_session) + storage = FakeObjectStorage() + + result = await upload_source_file( + sessionmaker=db_sessionmaker, + storage=storage, + auth=auth, + domain="general", + filename="report.csv", + data=_CSV_BYTES, + max_upload_size_bytes=1_000_000, + chunking_strategy="fixed_size", + validation_limiter=CapacityLimiter(2), + ) + + assert result.status == "succeeded" + assert result.chunks_indexed == 0 + assert result.is_new_attempt + + async with db_sessionmaker() as verify_session: + job = await verify_session.get(IngestionJob, result.ingestion_job_id) + assert job is not None + assert job.status == "succeeded" + assert storage.objects # bytes were actually written + + +async def test_upload_source_file_duplicate_hash_returns_existing_without_reingesting( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + auth = await _auth_for(db_session) + storage = FakeObjectStorage() + + first = await upload_source_file( + sessionmaker=db_sessionmaker, + storage=storage, + auth=auth, + domain="general", + filename="report.csv", + data=_CSV_BYTES, + max_upload_size_bytes=1_000_000, + chunking_strategy="fixed_size", + validation_limiter=CapacityLimiter(2), + ) + second = await upload_source_file( + sessionmaker=db_sessionmaker, + storage=storage, + auth=auth, + domain="general", + filename="report.csv", + data=_CSV_BYTES, + max_upload_size_bytes=1_000_000, + chunking_strategy="fixed_size", + validation_limiter=CapacityLimiter(2), + ) + + assert second.file_id == first.file_id + assert second.ingestion_job_id == first.ingestion_job_id + assert not second.is_new_attempt + assert len(storage.objects) == 1 # no second write + + +async def test_upload_source_file_retries_after_failed_job( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + """A duplicate upload whose last attempt failed must get a fresh job, + not be treated as already-succeeded (ADR-0017: safe to run more than + once). + """ + auth = await _auth_for(db_session) + failing_storage = FakeObjectStorage(fail_next=True) + + with pytest.raises(OSError): + await upload_source_file( + sessionmaker=db_sessionmaker, + storage=failing_storage, + auth=auth, + domain="general", + filename="report.csv", + data=_CSV_BYTES, + max_upload_size_bytes=1_000_000, + chunking_strategy="fixed_size", + validation_limiter=CapacityLimiter(2), + ) + + async with db_sessionmaker() as verify_session: + jobs = ( + ( + await verify_session.execute( + select(IngestionJob).where(IngestionJob.tenant_id == auth.tenant_id) + ) + ) + .scalars() + .all() + ) + assert len(jobs) == 1 + assert jobs[0].status == "failed" + + retry = await upload_source_file( + sessionmaker=db_sessionmaker, + storage=FakeObjectStorage(), + auth=auth, + domain="general", + filename="report.csv", + data=_CSV_BYTES, + max_upload_size_bytes=1_000_000, + chunking_strategy="fixed_size", + validation_limiter=CapacityLimiter(2), + ) + + assert retry.status == "succeeded" + assert retry.is_new_attempt + assert retry.file_id == jobs[0].source_file_id + + +async def test_upload_source_file_is_tenant_isolated_for_identical_content( + db_session: AsyncSession, db_sessionmaker: async_sessionmaker[AsyncSession] +) -> None: + """Two different tenants uploading byte-identical files must not collide + on the (tenant_id, domain, content_sha256) idempotency key. + """ + auth_a = await _auth_for(db_session) + auth_b = await _auth_for(db_session) + + result_a = await upload_source_file( + sessionmaker=db_sessionmaker, + storage=FakeObjectStorage(), + auth=auth_a, + domain="general", + filename="report.csv", + data=_CSV_BYTES, + max_upload_size_bytes=1_000_000, + chunking_strategy="fixed_size", + validation_limiter=CapacityLimiter(2), + ) + result_b = await upload_source_file( + sessionmaker=db_sessionmaker, + storage=FakeObjectStorage(), + auth=auth_b, + domain="general", + filename="report.csv", + data=_CSV_BYTES, + max_upload_size_bytes=1_000_000, + chunking_strategy="fixed_size", + validation_limiter=CapacityLimiter(2), + ) + + assert result_a.file_id != result_b.file_id