diff --git a/.gitignore b/.gitignore index 5593ad3..1656658 100644 --- a/.gitignore +++ b/.gitignore @@ -9,3 +9,9 @@ db.txt .venv multiai-c9380-firebase-adminsdk-fbsvc-cb6e5ce41b.json + +# Local Test & Debug Files +.coverage +dummy.json +test_output.log +*.log diff --git a/app/router/staff/__init__.py b/app/router/staff/__init__.py index e17b02c..35cbe0e 100644 --- a/app/router/staff/__init__.py +++ b/app/router/staff/__init__.py @@ -4,7 +4,7 @@ from app.router.staff.notifications import router as staff_notifications_router from app.router.staff.uploads import router as staff_uploads_router -router = APIRouter(prefix="/stuff", tags=["stuff"]) +router = APIRouter(prefix="/staff", tags=["staff"]) router.include_router(staff_drive_router) router.include_router(staff_notifications_router) router.include_router(staff_uploads_router) diff --git a/app/worker/photo_worker/main.py b/app/worker/photo_worker/main.py index 15e0586..59a116b 100644 --- a/app/worker/photo_worker/main.py +++ b/app/worker/photo_worker/main.py @@ -179,9 +179,7 @@ async def _handle_group_photo(self, event: PhotoProcessEvent, faces: list[Detect ) if approvals_created == 0: - logger.info("No users matched in group photo %s, auto-approving as public", event.photo_id) - await self._photo_querier.update_photo_status(id=event.photo_id, status="approved") - await self._photo_querier.update_photo_visibility(id=event.photo_id, visibility="public") + logger.info("No users matched in group photo %s, leaving as pending", event.photo_id) async def _create_job(self, event: PhotoProcessEvent) -> models.ProcessingJob | None: diff --git a/db/generated/photo_approvals.py b/db/generated/photo_approvals.py index 56ba3a3..13a5096 100644 --- a/db/generated/photo_approvals.py +++ b/db/generated/photo_approvals.py @@ -11,6 +11,25 @@ from db.generated import models +EXPIRE_STALE_APPROVALS = """-- name: expire_stale_approvals \\:many +WITH stale_photos AS ( + SELECT id FROM photos + WHERE status = 'pending' + AND created_at < now() - make_interval(days => :p1::int) +), +_update_approvals AS ( + UPDATE photo_approvals + SET decision = 'approved', decided_at = now() + WHERE photo_id IN (SELECT id FROM stale_photos) + AND decision = 'pending' +) +UPDATE photos +SET status = 'approved' +WHERE id IN (SELECT id FROM stale_photos) +RETURNING id +""" + + CREATE_PHOTO_APPROVAL = """-- name: create_photo_approval \\:one INSERT INTO photo_approvals ( photo_id, @@ -68,6 +87,11 @@ class AsyncQuerier: def __init__(self, conn: sqlalchemy.ext.asyncio.AsyncConnection): self._conn = conn + async def expire_stale_approvals(self, *, timeout_days: int) -> AsyncIterator[uuid.UUID]: + result = await self._conn.stream(sqlalchemy.text(EXPIRE_STALE_APPROVALS), {"p1": timeout_days}) + async for row in result: + yield row[0] + async def create_photo_approval(self, *, photo_id: uuid.UUID, user_id: uuid.UUID, decision: str) -> Optional[models.PhotoApproval]: row = (await self._conn.execute(sqlalchemy.text(CREATE_PHOTO_APPROVAL), {"p1": photo_id, "p2": user_id, "p3": decision})).first() if row is None: diff --git a/pyproject.toml b/pyproject.toml index 2b0f8b2..2fc41df 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -63,7 +63,9 @@ warn_redundant_casts = true follow_imports = "silent" files = ["app", "db", "migrations"] exclude = [ - "db/generated" + "db/generated", + "tests", + "scripts" ] [dependency-groups] dev = [ diff --git a/scripts/check_ai_results.py b/scripts/check_ai_results.py new file mode 100644 index 0000000..fa17943 --- /dev/null +++ b/scripts/check_ai_results.py @@ -0,0 +1,23 @@ +import asyncio +from sqlalchemy.ext.asyncio import create_async_engine +from app.core.config import settings + +async def main(): + url = f"postgresql+asyncpg://{settings.POSTGRES_USER}:{settings.POSTGRES_PASSWORD}@{settings.POSTGRES_HOST}:{settings.POSTGRES_PORT}/{settings.POSTGRES_DB}" + engine = create_async_engine(url) + + async with engine.connect() as conn: + import sqlalchemy + + # Check processing jobs + jobs = (await conn.execute(sqlalchemy.text("SELECT status, count(*) FROM processing_jobs GROUP BY status"))).fetchall() + print("=== PROCESSING JOBS STATUS ===") + for j in jobs: + print(f"Status: {j[0]}, Count: {j[1]}") + + # Check photo faces + faces = (await conn.execute(sqlalchemy.text("SELECT count(*) FROM photo_faces"))).scalar() + print("\n=== VISAGES DETECTES ===") + print(f"Nombre total de visages isolΓ©s et enregistrΓ©s par l'IA : {faces}") + +asyncio.run(main()) diff --git a/scripts/check_scopes.py b/scripts/check_scopes.py new file mode 100644 index 0000000..1835786 --- /dev/null +++ b/scripts/check_scopes.py @@ -0,0 +1,17 @@ +import asyncio +from sqlalchemy.ext.asyncio import create_async_engine +from app.core.config import settings + +async def main(): + url = f"postgresql+asyncpg://{settings.POSTGRES_USER}:{settings.POSTGRES_PASSWORD}@{settings.POSTGRES_HOST}:{settings.POSTGRES_PORT}/{settings.POSTGRES_DB}" + engine = create_async_engine(url) + + async with engine.connect() as conn: + import sqlalchemy + row = (await conn.execute(sqlalchemy.text("SELECT google_email, scopes FROM staff_drive_connections LIMIT 1"))).fetchone() + if row: + print(f"Email: {row[0]}, Scopes: {row[1]}") + else: + print("No connection found") + +asyncio.run(main()) diff --git a/scripts/generate_drive_url.py b/scripts/generate_drive_url.py new file mode 100644 index 0000000..91a5802 --- /dev/null +++ b/scripts/generate_drive_url.py @@ -0,0 +1,44 @@ +import asyncio +from sqlalchemy.ext.asyncio import create_async_engine +from app.core.config import settings +from db.generated import stuff_user as staff_queries +from app.infra.redis import RedisClient +from app.service.staff_drive import StaffDriveService +from db.generated import staff_drive_connections as drive_queries + +async def main(): + url = f"postgresql+asyncpg://{settings.POSTGRES_USER}:{settings.POSTGRES_PASSWORD}@{settings.POSTGRES_HOST}:{settings.POSTGRES_PORT}/{settings.POSTGRES_DB}" + engine = create_async_engine(url) + + # Init redis + RedisClient.init(host=settings.REDIS_HOST, port=settings.REDIS_PORT, password=settings.REDIS_PASSWORD or "") + redis = RedisClient.get_instance() + + async with engine.connect() as conn: + q = staff_queries.AsyncQuerier(conn) + import sqlalchemy + row = (await conn.execute(sqlalchemy.text("SELECT id FROM staff_users LIMIT 1"))).fetchone() + + if not row: + print("No staff user found! Creating one...") + res = await q.create_staff_user(email="testadmin@example.com", hashed_password="pw", display_name="Admin", role="admin") + staff_user_id = res.id + else: + staff_user_id = row[0] + + class DummyUser: + pass + staff_user = DummyUser() + staff_user.id = staff_user_id + + drive_service = StaffDriveService( + staff_user_querier=q, + drive_connection_querier=drive_queries.AsyncQuerier(conn), + redis=redis + ) + url, state = await drive_service.create_connect_url(staff_user) + print("========================") + print("GOOGLE_AUTH_URL:", url) + print("========================") + +asyncio.run(main()) diff --git a/scripts/list_drive.py b/scripts/list_drive.py new file mode 100644 index 0000000..d4940bf --- /dev/null +++ b/scripts/list_drive.py @@ -0,0 +1,50 @@ +import asyncio +from sqlalchemy.ext.asyncio import create_async_engine +from app.core.config import settings +from db.generated import stuff_user as staff_queries +from app.infra.redis import RedisClient +from app.infra.google_drive import GoogleDriveClient +from app.service.staff_drive import StaffDriveService +from db.generated import staff_drive_connections as drive_queries + +async def main(): + url = f"postgresql+asyncpg://{settings.POSTGRES_USER}:{settings.POSTGRES_PASSWORD}@{settings.POSTGRES_HOST}:{settings.POSTGRES_PORT}/{settings.POSTGRES_DB}" + engine = create_async_engine(url) + + # Init redis + try: + RedisClient.init(host=settings.REDIS_HOST, port=settings.REDIS_PORT, password=settings.REDIS_PASSWORD or "") + except RuntimeError: + pass + redis = RedisClient.get_instance() + + async with engine.connect() as conn: + q = staff_queries.AsyncQuerier(conn) + import sqlalchemy + row = (await conn.execute(sqlalchemy.text("SELECT id FROM staff_users LIMIT 1"))).fetchone() + staff_user_id = row[0] + + class DummyUser: + pass + staff_user = DummyUser() + staff_user.id = staff_user_id + + drive_service = StaffDriveService( + staff_user_querier=q, + drive_connection_querier=drive_queries.AsyncQuerier(conn), + redis=redis + ) + + access_token = await drive_service.get_access_token_for_staff_user(staff_user_id) + + # List root folders + print("=== DOSSIERS ET FICHIERS A LA RACINE DU DRIVE ===") + items = await GoogleDriveClient.list_folder_contents(access_token=access_token) + for i in items[:20]: + type_str = "πŸ“ DOSSIER" if i.mime_type == "application/vnd.google-apps.folder" else "πŸ“„ FICHIER" + print(f"{type_str} | ID: {i.id} | NOM: {i.name}") + + if not items: + print("Aucun fichier trouvΓ©.") + +asyncio.run(main()) diff --git a/scripts/test_routes.py b/scripts/test_routes.py new file mode 100644 index 0000000..f793df6 --- /dev/null +++ b/scripts/test_routes.py @@ -0,0 +1,8 @@ +from fastapi.testclient import TestClient +from app.main import app + +with TestClient(app) as client: + resp = client.get("/user/photos") + print(f"/user/photos -> {resp.status_code} {resp.text}") + resp2 = client.get("/user/photos/") + print(f"/user/photos/ -> {resp2.status_code} {resp2.text}") diff --git a/scripts/trigger_import.py b/scripts/trigger_import.py new file mode 100644 index 0000000..f309878 --- /dev/null +++ b/scripts/trigger_import.py @@ -0,0 +1,73 @@ +import asyncio +from sqlalchemy.ext.asyncio import create_async_engine +from app.core.config import settings +from db.generated import stuff_user as staff_queries +from app.infra.redis import RedisClient +from app.infra.google_drive import GoogleDriveClient +from app.infra.minio import init_minio_client +from app.service.staff_drive import StaffDriveService, SelectedDriveFile +from db.generated import staff_drive_connections as drive_queries + +async def main(): + url = f"postgresql+asyncpg://{settings.POSTGRES_USER}:{settings.POSTGRES_PASSWORD}@{settings.POSTGRES_HOST}:{settings.POSTGRES_PORT}/{settings.POSTGRES_DB}" + engine = create_async_engine(url) + + try: + RedisClient.init(host=settings.REDIS_HOST, port=settings.REDIS_PORT, password=settings.REDIS_PASSWORD or "") + except RuntimeError: + pass + redis = RedisClient.get_instance() + + # Init minio + await init_minio_client( + minio_host=settings.MINIO_HOST, + minio_port=settings.MINIO_API_PORT, + minio_root_user=settings.MINIO_ROOT_USER, + minio_root_password=settings.MINIO_ROOT_PASSWORD + ) + + async with engine.connect() as conn: + q = staff_queries.AsyncQuerier(conn) + import sqlalchemy + row = (await conn.execute(sqlalchemy.text("SELECT id FROM staff_users LIMIT 1"))).fetchone() + staff_user_id = row[0] + + class DummyUser: + pass + staff_user = DummyUser() + staff_user.id = staff_user_id + + drive_service = StaffDriveService( + staff_user_querier=q, + drive_connection_querier=drive_queries.AsyncQuerier(conn), + redis=redis + ) + + access_token = await drive_service.get_access_token_for_staff_user(staff_user_id) + + print("Fetching images from Drive...") + items = await GoogleDriveClient.list_folder_contents(access_token=access_token) + + # Filter images + image_items = [i for i in items if i.mime_type.startswith("image/")] + + if not image_items: + print("No images found to import.") + return + + print(f"Found {len(image_items)} images. Initiating import...") + selections = [ + SelectedDriveFile(id=f.id, name=f.name, mime_type=f.mime_type) + for f in image_items[:20] # Let's import up to 20 for this test + ] + + results = await drive_service.import_images_from_drive( + staff_user=staff_user, + selected_files=selections, + ) + + print(f"Successfully imported {len(results)} images to MinIO!") + for r in results: + print(f" -> {r.original_file_name} stored as {r.minio_object_name}") + +asyncio.run(main()) diff --git a/scripts/trigger_photo_worker.py b/scripts/trigger_photo_worker.py new file mode 100644 index 0000000..b1eba58 --- /dev/null +++ b/scripts/trigger_photo_worker.py @@ -0,0 +1,80 @@ +import asyncio +import json +import uuid +from sqlalchemy.ext.asyncio import create_async_engine +from app.core.config import settings +from app.infra.redis import RedisClient +from app.infra.minio import init_minio_client +from app.infra.nats import NatsClient, NatsSubjects + +async def main(): + url = f"postgresql+asyncpg://{settings.POSTGRES_USER}:{settings.POSTGRES_PASSWORD}@{settings.POSTGRES_HOST}:{settings.POSTGRES_PORT}/{settings.POSTGRES_DB}" + engine = create_async_engine(url) + + try: + RedisClient.init(host=settings.REDIS_HOST, port=settings.REDIS_PORT, password=settings.REDIS_PASSWORD or "") + except RuntimeError: + pass + + await init_minio_client( + minio_host=settings.MINIO_HOST, + minio_port=settings.MINIO_API_PORT, + minio_root_user=settings.MINIO_ROOT_USER, + minio_root_password=settings.MINIO_ROOT_PASSWORD + ) + + await NatsClient.connect( + host=settings.NATS_HOST, + port=settings.NATS_PORT, + user=settings.NATS_USER, + password=settings.NATS_PASSWORD + ) + + async with engine.connect() as conn: + # staff_queries not used here but kept for context + import sqlalchemy + row = (await conn.execute(sqlalchemy.text("SELECT id FROM staff_users LIMIT 1"))).fetchone() + staff_user_id = row[0] + + event_row = (await conn.execute(sqlalchemy.text("SELECT id FROM events LIMIT 1"))).fetchone() + if not event_row: + print("Creating dummy event...") + ev_id = uuid.uuid4() + await conn.execute(sqlalchemy.text("INSERT INTO events (id, title, date, location) VALUES (:id, 'Test Event', now(), 'Test Location')"), {"id": ev_id}) + event_id = ev_id + else: + event_id = event_row[0] + + from app.infra.minio import ImageBucket + bucket = ImageBucket(f"staff-drive/{staff_user_id}") + + objects = bucket.client.list_objects(bucket.bucket_name, prefix=bucket.file_prefix + "/", recursive=True) + count = 0 + async for obj in objects: + storage_key = obj.object_name + print(f"Injecting photo {storage_key} into AI pipeline...") + + # Create photo in DB + new_id = uuid.uuid4() + await conn.execute( + sqlalchemy.text("INSERT INTO photos (id, event_id, storage_key, visibility) VALUES (:id, :event_id, :storage_key, 'public')"), + {"id": new_id, "event_id": event_id, "storage_key": storage_key} + ) + + # Publish event + await NatsClient.publish( + NatsSubjects.PHOTO_PROCESS, + json.dumps({ + "photo_id": str(new_id), + "image_ref": storage_key, + "event_id": str(event_id) + }).encode("utf-8") + ) + count += 1 + if count >= 20: + break + + await conn.commit() + print(f"Successfully injected {count} photos to the AI worker!") + +asyncio.run(main()) diff --git a/scripts/trigger_upload_request.py b/scripts/trigger_upload_request.py new file mode 100644 index 0000000..639579e --- /dev/null +++ b/scripts/trigger_upload_request.py @@ -0,0 +1,117 @@ +import asyncio +from sqlalchemy.ext.asyncio import create_async_engine +from app.core.config import settings +from db.generated import stuff_user as staff_queries +from db.generated import upload_request_groups as group_queries +from db.generated import upload_requests as request_queries +from db.generated import upload_request_photos as request_photo_queries +from db.generated import photos as photo_queries +from db.generated import staff_drive_connections as drive_queries +from db.generated import staff_notifications as notif_queries +from db.generated import audit as audit_queries +from app.service.upload_requests import UploadRequestsService +from app.service.staged_upload_storage import StagedUploadStorageService +from app.service.staff_drive import StaffDriveService +from app.service.staff_notifications import StaffNotificationsService +from app.service.audit import AuditService +from app.schema.request.staff.uploads import CreateUploadRequestPhotoRequest +from app.infra.redis import RedisClient +from app.infra.minio import init_minio_client +from app.infra.nats import NatsClient +import uuid + +async def main(): + url = f"postgresql+asyncpg://{settings.POSTGRES_USER}:{settings.POSTGRES_PASSWORD}@{settings.POSTGRES_HOST}:{settings.POSTGRES_PORT}/{settings.POSTGRES_DB}" + engine = create_async_engine(url) + + try: + RedisClient.init(host=settings.REDIS_HOST, port=settings.REDIS_PORT, password=settings.REDIS_PASSWORD or "") + except RuntimeError: + pass + redis = RedisClient.get_instance() + + await init_minio_client( + minio_host=settings.MINIO_HOST, + minio_port=settings.MINIO_API_PORT, + minio_root_user=settings.MINIO_ROOT_USER, + minio_root_password=settings.MINIO_ROOT_PASSWORD + ) + + await NatsClient.connect( + host=settings.NATS_HOST, + port=settings.NATS_PORT, + user=settings.NATS_USER, + password=settings.NATS_PASSWORD + ) + + async with engine.connect() as conn: + q = staff_queries.AsyncQuerier(conn) + import sqlalchemy + row = (await conn.execute(sqlalchemy.text("SELECT id FROM staff_users LIMIT 1"))).fetchone() + staff_user_id = row[0] + + class DummyUser: + pass + staff_user = DummyUser() + staff_user.id = staff_user_id + staff_user.role = "multi_team_lead" # Important for approval! + + # Need an event + event_row = (await conn.execute(sqlalchemy.text("SELECT id FROM events LIMIT 1"))).fetchone() + if not event_row: + print("Creating dummy event...") + ev_id = uuid.uuid4() + await conn.execute(sqlalchemy.text("INSERT INTO events (id, title, date, location) VALUES (:id, 'Test Event', now(), 'Test Location')"), {"id": ev_id}) + event_id = ev_id + else: + event_id = event_row[0] + + upload_service = UploadRequestsService( + upload_request_group_querier=group_queries.AsyncQuerier(conn), + upload_request_querier=request_queries.AsyncQuerier(conn), + upload_request_photo_querier=request_photo_queries.AsyncQuerier(conn), + photo_querier=photo_queries.AsyncQuerier(conn), + staged_upload_storage=StagedUploadStorageService(), + staff_drive_service=StaffDriveService(staff_user_querier=q, drive_connection_querier=drive_queries.AsyncQuerier(conn), redis=redis), + staff_notifications_service=StaffNotificationsService(notif_queries.AsyncQuerier(conn)), + audit_service=AuditService(audit_queries.AsyncQuerier(conn), None), + ) + + # Get staged photos from minio bucket for this staff user + from app.infra.minio import ImageBucket + bucket = ImageBucket(f"staff-drive/{staff_user_id}") + + objects = bucket.client.list_objects(bucket.bucket_name, prefix=bucket.file_prefix + "/", recursive=True) + photo_inputs = [] + async for obj in objects: + name = obj.object_name.split("/")[-1] + photo_inputs.append(CreateUploadRequestPhotoRequest( + staged_object_name=name, + original_file_name=name, + )) + if len(photo_inputs) >= 20: + break + + if not photo_inputs: + print("No staged photos found in minio.") + return + + print(f"Submitting {len(photo_inputs)} photos to upload request...") + + req_details = await upload_service.create_upload( + event_id=event_id, + folder_id="drive-import", + photos=photo_inputs, + visibility="public", + day_number=1, + requested_by=staff_user, + ) + + print(f"Created upload request! ID: {req_details.id}") + print("Approving upload request to trigger AI pipeline...") + + await upload_service.approve_request(request_id=req_details.id, approved_by=staff_user) + + print("Done! Photos should now be processed by AI.") + +asyncio.run(main()) diff --git a/tests/e2e/conftest.py b/tests/e2e/conftest.py new file mode 100644 index 0000000..8428d2b --- /dev/null +++ b/tests/e2e/conftest.py @@ -0,0 +1,120 @@ +import asyncio +import os +import uuid +from collections.abc import AsyncGenerator +from pathlib import Path + +import pytest +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncConnection + +from app.core.config import settings +from app.infra.database import engine +from app.infra.minio import init_minio_client +from app.infra.nats import NatsClient + +FIXTURE_DIR = Path(__file__).parent.parent / "fixtures" / "images" + +# ── guard: only run when explicitly requested ───────────────────────── +def pytest_configure(config: pytest.Config) -> None: + config.addinivalue_line("markers", "e2e: mark test as an end-to-end test") + +@pytest.fixture(autouse=True) +async def setup_infra() -> AsyncGenerator[None, None]: + if os.getenv("MULTAI_RUN_E2E") != "1": + pytest.skip("set MULTAI_RUN_E2E=1 to run live e2e tests") + + await init_minio_client( + minio_host=settings.MINIO_HOST, + minio_port=settings.MINIO_API_PORT, + minio_root_user=settings.MINIO_ROOT_USER, + minio_root_password=settings.MINIO_ROOT_PASSWORD, + ) + await NatsClient.connect() + yield + await NatsClient.close() + NatsClient._nc = None # type: ignore[attr-defined] + await engine.dispose() + + +# ── shared helpers ──────────────────────────────────────────────────── + +async def _seed_event_and_photo( + conn: AsyncConnection, + *, + photo_id: uuid.UUID, + storage_key: str, +) -> uuid.UUID: + """Insert a staff user, a new event, and a photo. Returns event_id.""" + event_id = uuid.uuid4() + await conn.execute( # type: ignore[union-attr] + text( + """ + INSERT INTO staff_users (id, email, password, role) + VALUES ('00000000-0000-0000-0000-000000000001'::uuid, + 'e2e@test.com', 'hash', 'admin') + ON CONFLICT (id) DO NOTHING + """ + ) + ) + event_code = f"E2E-{str(uuid.uuid4())[:6].upper()}" + await conn.execute( # type: ignore[union-attr] + text( + """ + INSERT INTO events (id, name, event_code, event_date, status, created_by) + VALUES (:id, 'E2E Test Event', :code, NOW(), 'draft', + '00000000-0000-0000-0000-000000000001'::uuid) + """ + ), + {"id": event_id, "code": event_code}, + ) + await conn.execute( # type: ignore[union-attr] + text( + """ + INSERT INTO photos (id, event_id, storage_key, visibility, status) + VALUES (:id, :event_id, :key, 'private', 'pending') + """ + ), + {"id": photo_id, "event_id": event_id, "key": storage_key}, + ) + return event_id + + +async def _wait_for_job(photo_id: uuid.UUID, timeout_s: int = 60) -> str: + """Poll processing_jobs until terminal status. Returns 'completed', 'failed', or 'timeout'.""" + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout_s + async with engine.connect() as conn: + while loop.time() < deadline: + row = ( + await conn.execute( + text( + "SELECT status FROM processing_jobs " + "WHERE photo_id = :pid AND job_type = 'face_detection'" + ), + {"pid": photo_id}, + ) + ).fetchone() + if row and row[0] in ("completed", "failed"): + return str(row[0]) + await asyncio.sleep(1.0) + return "timeout" + + +async def _cleanup( + conn: AsyncConnection, + *, + photo_id: uuid.UUID, + event_id: uuid.UUID, + user_id: str | None = None, +) -> None: + """Delete all rows created during a test, in FK-safe order.""" + if user_id: + await conn.execute(text("DELETE FROM notifications WHERE user_id = :uid"), {"uid": user_id}) # type: ignore[union-attr] + await conn.execute(text("DELETE FROM face_matches WHERE user_id = :uid"), {"uid": user_id}) # type: ignore[union-attr] + await conn.execute(text("DELETE FROM users WHERE id = :uid"), {"uid": user_id}) # type: ignore[union-attr] + await conn.execute(text("DELETE FROM face_matches fm USING photo_faces pf WHERE pf.id = fm.photo_face_id AND pf.photo_id = :pid"), {"pid": photo_id}) # type: ignore[union-attr] + await conn.execute(text("DELETE FROM photo_faces WHERE photo_id = :pid"), {"pid": photo_id}) # type: ignore[union-attr] + await conn.execute(text("DELETE FROM processing_jobs WHERE photo_id = :pid"), {"pid": photo_id}) # type: ignore[union-attr] + await conn.execute(text("DELETE FROM photos WHERE id = :pid"), {"pid": photo_id}) # type: ignore[union-attr] + await conn.execute(text("DELETE FROM events WHERE id = :eid"), {"eid": event_id}) # type: ignore[union-attr] diff --git a/tests/e2e/test_photo_ai_edge_cases.py b/tests/e2e/test_photo_ai_edge_cases.py new file mode 100644 index 0000000..ed3e076 --- /dev/null +++ b/tests/e2e/test_photo_ai_edge_cases.py @@ -0,0 +1,212 @@ +import json +import uuid + +from sqlalchemy import text + +from app.infra.database import engine +from app.infra.minio import Bucket, IMAGES_BUCKET_NAME +from app.infra.nats import NatsClient, NatsSubjects + +from tests.e2e.conftest import _seed_event_and_photo, _wait_for_job, _cleanup, FIXTURE_DIR + +# ── tests ───────────────────────────────────────────────────────────── + + +async def test_photo_ai_pipeline_detects_0_faces() -> None: + """Photo with no faces β†’ auto-approved, visibility set to public.""" + photo_id = uuid.uuid4() + storage_key = f"e2e_noface_{photo_id}.jpg" + image_path = FIXTURE_DIR / "noface.jpg" + assert image_path.exists(), f"Fixture not found: {image_path}" + + bucket = Bucket(IMAGES_BUCKET_NAME, "") + await bucket.put_bytes( + object_name=storage_key, + data=image_path.read_bytes(), + content_type="image/jpeg", + ) + async with engine.begin() as conn: + event_id = await _seed_event_and_photo( + conn, photo_id=photo_id, storage_key=storage_key + ) + + payload = {"photo_id": str(photo_id), "image_ref": storage_key, "event_id": str(event_id)} + await NatsClient.publish(NatsSubjects.PHOTO_PROCESS, json.dumps(payload).encode("utf-8")) + + try: + final_status = await _wait_for_job(photo_id) + assert final_status == "completed", f"Job ended with: {final_status}" + + async with engine.connect() as conn: + row = ( + await conn.execute( + text("SELECT status, visibility FROM photos WHERE id = :pid"), + {"pid": photo_id}, + ) + ).fetchone() + assert row is not None + assert row[0] == "approved", f"Expected 'approved', got {row[0]}" + assert row[1] == "public", f"Expected visibility 'public', got {row[1]}" + finally: + async with engine.begin() as conn: + await _cleanup(conn, photo_id=photo_id, event_id=event_id) + try: + await bucket.delete(storage_key) + except Exception: + pass + + +async def test_photo_ai_pipeline_detects_multiple_faces() -> None: + """Group photo (multiple faces) β†’ status stays 'pending' awaiting approval.""" + photo_id = uuid.uuid4() + storage_key = f"e2e_group_{photo_id}.jpg" + image_path = FIXTURE_DIR / "group.jpg" + assert image_path.exists(), f"Fixture not found: {image_path}" + + bucket = Bucket(IMAGES_BUCKET_NAME, "") + await bucket.put_bytes( + object_name=storage_key, + data=image_path.read_bytes(), + content_type="image/jpeg", + ) + async with engine.begin() as conn: + event_id = await _seed_event_and_photo( + conn, photo_id=photo_id, storage_key=storage_key + ) + + payload = {"photo_id": str(photo_id), "image_ref": storage_key, "event_id": str(event_id)} + await NatsClient.publish(NatsSubjects.PHOTO_PROCESS, json.dumps(payload).encode("utf-8")) + + try: + final_status = await _wait_for_job(photo_id) + assert final_status == "completed", f"Job ended with: {final_status}" + + async with engine.connect() as conn: + photo_status = ( + await conn.execute( + text("SELECT status FROM photos WHERE id = :pid"), + {"pid": photo_id}, + ) + ).scalar() + # Group photo β†’ pending because multiple unverified faces require human approval + assert photo_status == "pending", f"Expected 'pending', got {photo_status}" + finally: + async with engine.begin() as conn: + await _cleanup(conn, photo_id=photo_id, event_id=event_id) + try: + await bucket.delete(storage_key) + except Exception: + pass + + +async def test_photo_ai_pipeline_matched_user() -> None: + """Single face matching an enrolled user β†’ 'approved' + notification created.""" + from app.service.face_embedding import FaceEmbeddingService, FaceImagePayload + + photo_id = uuid.uuid4() + storage_key = f"e2e_matched_{photo_id}.jpg" + image_path = FIXTURE_DIR / "face.jpg" + assert image_path.exists(), f"Fixture not found: {image_path}" + + image_bytes = image_path.read_bytes() + + # Pre-compute the embedding so we can plant a matching user + face_service = FaceEmbeddingService() + payload_face = FaceImagePayload( + filename="face.jpg", content_type="image/jpeg", bytes=image_bytes + ) + faces = await face_service.detect_faces(payload_face) + assert len(faces) == 1, f"Expected exactly 1 face in face.jpg, got {len(faces)}" + embedding_literal = "[" + ", ".join(str(x) for x in faces[0].embedding) + "]" + + matched_user_id = "00000000-0000-0000-0000-000000000002" + + bucket = Bucket(IMAGES_BUCKET_NAME, "") + await bucket.put_bytes( + object_name=storage_key, data=image_bytes, content_type="image/jpeg" + ) + + async with engine.begin() as conn: + event_id = await _seed_event_and_photo( + conn, photo_id=photo_id, storage_key=storage_key + ) + # Clean up any previous failed run artifacts for this user + await conn.execute( + text("DELETE FROM notifications WHERE user_id = :uid"), + {"uid": matched_user_id}, + ) + await conn.execute( + text("DELETE FROM face_matches WHERE user_id = :uid"), + {"uid": matched_user_id}, + ) + await conn.execute( + text("DELETE FROM users WHERE id = :uid"), {"uid": matched_user_id} + ) + # Insert a user with the exact same embedding + await conn.execute( + text( + """ + INSERT INTO users (id, email, hashed_password, face_embedding) + VALUES (:uid, 'matched@test.com', 'hash', :emb) + """ + ), + {"uid": matched_user_id, "emb": embedding_literal}, + ) + + payload = {"photo_id": str(photo_id), "image_ref": storage_key, "event_id": str(event_id)} + await NatsClient.publish(NatsSubjects.PHOTO_PROCESS, json.dumps(payload).encode("utf-8")) + + try: + final_status = await _wait_for_job(photo_id) + assert final_status == "completed", f"Job ended with: {final_status}" + + async with engine.connect() as conn: + photo_status = ( + await conn.execute( + text("SELECT status FROM photos WHERE id = :pid"), + {"pid": photo_id}, + ) + ).scalar() + assert photo_status == "approved", ( + f"Expected 'approved', got {photo_status}" + ) + + matched_db_user = ( + await conn.execute( + text( + """ + SELECT fm.user_id + FROM face_matches fm + JOIN photo_faces pf ON pf.id = fm.photo_face_id + WHERE pf.photo_id = :pid + """ + ), + {"pid": photo_id}, + ) + ).scalar() + assert str(matched_db_user) == matched_user_id, ( + f"Expected matched user {matched_user_id}, got {matched_db_user}" + ) + + notif_count = ( + await conn.execute( + text( + "SELECT count(*) FROM notifications " + "WHERE user_id = :uid AND type = 'face_match'" + ), + {"uid": matched_user_id}, + ) + ).scalar() + assert notif_count == 1, f"Expected 1 notification, got {notif_count}" + finally: + async with engine.begin() as conn: + await _cleanup( + conn, + photo_id=photo_id, + event_id=event_id, + user_id=matched_user_id, + ) + try: + await bucket.delete(storage_key) + except Exception: + pass diff --git a/tests/e2e/test_photo_ai_load.py b/tests/e2e/test_photo_ai_load.py new file mode 100644 index 0000000..0f9437b --- /dev/null +++ b/tests/e2e/test_photo_ai_load.py @@ -0,0 +1,219 @@ +import asyncio +import json +import os +import random +import uuid +from collections.abc import AsyncGenerator +from pathlib import Path + +import pytest +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncConnection + +from app.core.config import settings +from app.infra.database import engine +from app.infra.minio import Bucket, IMAGES_BUCKET_NAME, init_minio_client +from app.infra.nats import NatsClient, NatsSubjects + +# ── guard: only run when explicitly requested ───────────────────────── +pytestmark = [ + pytest.mark.e2e, + pytest.mark.asyncio, + pytest.mark.skipif( + os.getenv("MULTAI_RUN_E2E") != "1", + reason="set MULTAI_RUN_E2E=1 to run live e2e tests", + ), +] + +FIXTURE_DIR = Path(__file__).parent.parent / "fixtures" / "images" + + +@pytest.fixture(scope="function") +async def setup_infra() -> AsyncGenerator[None, None]: + await init_minio_client( + minio_host=settings.MINIO_HOST, + minio_port=settings.MINIO_API_PORT, + minio_root_user=settings.MINIO_ROOT_USER, + minio_root_password=settings.MINIO_ROOT_PASSWORD, + ) + await NatsClient.connect() + yield + await NatsClient.close() + NatsClient._nc = None # type: ignore[attr-defined] + await engine.dispose() + + +async def _setup_event(conn: AsyncConnection) -> uuid.UUID: + event_id = uuid.uuid4() + await conn.execute( # type: ignore[union-attr] + text( + """ + INSERT INTO staff_users (id, email, password, role) + VALUES ('00000000-0000-0000-0000-000000000001'::uuid, + 'e2e_load@test.com', 'hash', 'admin') + ON CONFLICT (id) DO NOTHING + """ + ) + ) + event_code = f"LOAD-{str(uuid.uuid4())[:6].upper()}" + await conn.execute( # type: ignore[union-attr] + text( + """ + INSERT INTO events (id, name, event_code, event_date, status, created_by) + VALUES (:id, 'E2E Load Event', :code, NOW(), 'draft', + '00000000-0000-0000-0000-000000000001'::uuid) + """ + ), + {"id": event_id, "code": event_code}, + ) + return event_id + + +async def _wait_for_jobs( + photo_ids: list[uuid.UUID], timeout: int = 180 +) -> dict[str, int]: + """Return a statusβ†’count dict once all jobs have a terminal status.""" + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + async with engine.connect() as conn: + while loop.time() < deadline: + rows = ( + await conn.execute( + text( + """ + SELECT status, count(*) + FROM processing_jobs + WHERE photo_id = ANY(:ids) + AND job_type = 'face_detection' + GROUP BY status + """ + ), + {"ids": photo_ids}, + ) + ).fetchall() + status_counts: dict[str, int] = {row[0]: int(row[1]) for row in rows} + total = sum(status_counts.values()) + if total == len(photo_ids): + completed = status_counts.get("completed", 0) + failed = status_counts.get("failed", 0) + if completed + failed == len(photo_ids): + return status_counts + await asyncio.sleep(2.0) + return {"timeout": 1} + + +async def test_photo_ai_load_20_photos(setup_infra: None) -> None: # noqa: ARG001 + """Process 20 photos concurrently; expect 0 failures within 3 minutes.""" + image_files = ["face.jpg", "group.jpg", "noface.jpg"] + image_contents: dict[str, bytes] = {} + for img in image_files: + path = FIXTURE_DIR / img + assert path.exists(), f"Fixture not found: {path}" + image_contents[img] = path.read_bytes() + + num_photos = 20 + photo_tasks: list[dict[str, object]] = [] + event_id: uuid.UUID | None = None + + async with engine.begin() as conn: + event_id = await _setup_event(conn) + for _ in range(num_photos): + photo_id = uuid.uuid4() + selected_img = random.choice(image_files) + storage_key = f"load-test/{event_id}/{photo_id}.jpg" + photo_tasks.append( + {"photo_id": photo_id, "storage_key": storage_key, "content": image_contents[selected_img]} + ) + await conn.execute( + text( + """ + INSERT INTO photos (id, event_id, storage_key, visibility, status) + VALUES (:id, :event_id, :key, 'private', 'pending') + """ + ), + {"id": photo_id, "event_id": event_id, "key": storage_key}, + ) + + assert event_id is not None + bucket = Bucket(IMAGES_BUCKET_NAME, "") + + try: + # 1. Upload all photos to MinIO concurrently + await asyncio.gather( + *[ + bucket.put_bytes( + object_name=p["storage_key"], # type: ignore[arg-type] + data=p["content"], # type: ignore[arg-type] + content_type="image/jpeg", + ) + for p in photo_tasks + ] + ) + + # 2. Publish 20 NATS messages concurrently + await asyncio.gather( + *[ + NatsClient.publish( + NatsSubjects.PHOTO_PROCESS.value, + json.dumps( + { + "photo_id": str(p["photo_id"]), + "image_ref": p["storage_key"], + "event_id": str(event_id), + } + ).encode("utf-8"), + ) + for p in photo_tasks + ] + ) + + # 3. Wait for all jobs + photo_ids: list[uuid.UUID] = [ + p["photo_id"] for p in photo_tasks # type: ignore[misc] + ] + status_counts = await _wait_for_jobs(photo_ids, timeout=180) + + assert "timeout" not in status_counts, ( + "Load test timed out waiting for jobs to complete" + ) + failed = status_counts.get("failed", 0) + completed = status_counts.get("completed", 0) + assert failed == 0, f"Expected 0 failed jobs, got {failed}" + assert completed == num_photos, ( + f"Expected {num_photos} completed jobs, got {completed}" + ) + finally: + # Always clean up DB and MinIO regardless of test outcome + async with engine.begin() as conn: + photo_ids_list = [p["photo_id"] for p in photo_tasks] + if photo_ids_list: + await conn.execute( + text( + "DELETE FROM face_matches fm " + "USING photo_faces pf " + "WHERE pf.id = fm.photo_face_id " + "AND pf.photo_id = ANY(:ids)" + ), + {"ids": photo_ids_list}, + ) + await conn.execute( + text("DELETE FROM photo_faces WHERE photo_id = ANY(:ids)"), + {"ids": photo_ids_list}, + ) + await conn.execute( + text("DELETE FROM processing_jobs WHERE photo_id = ANY(:ids)"), + {"ids": photo_ids_list}, + ) + await conn.execute( + text("DELETE FROM photos WHERE event_id = :eid"), + {"eid": event_id}, + ) + await conn.execute( + text("DELETE FROM events WHERE id = :eid"), {"eid": event_id} + ) + # Clean up MinIO objects + for p in photo_tasks: + try: + await bucket.delete(p["storage_key"]) # type: ignore[arg-type] + except Exception: + pass diff --git a/tests/e2e/test_photo_ai_pipeline_e2e.py b/tests/e2e/test_photo_ai_pipeline_e2e.py new file mode 100644 index 0000000..0a7b3b5 --- /dev/null +++ b/tests/e2e/test_photo_ai_pipeline_e2e.py @@ -0,0 +1,106 @@ +import json +import uuid + +from sqlalchemy import text + +from app.infra.database import engine +from app.infra.minio import Bucket, IMAGES_BUCKET_NAME +from app.infra.nats import NatsClient, NatsSubjects + +from tests.e2e.conftest import _seed_event_and_photo, _wait_for_job, _cleanup, FIXTURE_DIR + + +async def test_photo_ai_pipeline_detects_single_face() -> None: + """Single face in photo with no enrolled users β†’ auto-approved.""" + photo_id = uuid.uuid4() + storage_key = f"e2e_test_{photo_id}.jpg" + image_path = FIXTURE_DIR / "face.jpg" + + assert image_path.exists(), f"Test fixture not found: {image_path}" + + bucket = Bucket(IMAGES_BUCKET_NAME, "") + + # 1. Seed MinIO + DB + await bucket.put_bytes( + object_name=storage_key, + data=image_path.read_bytes(), + content_type="image/jpeg", + ) + async with engine.begin() as conn: + event_id = await _seed_event_and_photo( + conn, photo_id=photo_id, storage_key=storage_key + ) + + # 2. Trigger worker + payload = { + "photo_id": str(photo_id), + "image_ref": storage_key, + "event_id": str(event_id), + } + await NatsClient.publish( + NatsSubjects.PHOTO_PROCESS, json.dumps(payload).encode("utf-8") + ) + + # 3. Assertions + Cleanup β€” always run cleanup via try/finally + try: + final_status = await _wait_for_job(photo_id, timeout_s=60) + assert final_status == "completed", f"Processing job ended with: {final_status}" + + async with engine.connect() as conn: + photo_status = ( + await conn.execute( + text("SELECT status FROM photos WHERE id = :pid"), + {"pid": photo_id}, + ) + ).scalar() + assert photo_status == "approved", ( + f"Expected photo status 'approved', got {photo_status}" + ) + finally: + async with engine.begin() as conn: + await _cleanup(conn, photo_id=photo_id, event_id=event_id) + try: + await bucket.delete(storage_key) + except Exception: + pass + + +async def test_photo_ai_pipeline_corrupt_image() -> None: + """Corrupt image β†’ processing job fails, photo status remains pending or marked as error.""" + photo_id = uuid.uuid4() + storage_key = f"e2e_corrupt_{photo_id}.jpg" + + bucket = Bucket(IMAGES_BUCKET_NAME, "") + + # 1. Seed MinIO with corrupt data + DB + await bucket.put_bytes( + object_name=storage_key, + data=b"this is not a valid image file", + content_type="image/jpeg", + ) + async with engine.begin() as conn: + event_id = await _seed_event_and_photo( + conn, photo_id=photo_id, storage_key=storage_key + ) + + # 2. Trigger worker + payload = { + "photo_id": str(photo_id), + "image_ref": storage_key, + "event_id": str(event_id), + } + await NatsClient.publish( + NatsSubjects.PHOTO_PROCESS, json.dumps(payload).encode("utf-8") + ) + + # 3. Assertions + Cleanup + try: + final_status = await _wait_for_job(photo_id, timeout_s=30) + assert final_status == "failed", f"Expected job to fail, but ended with: {final_status}" + finally: + async with engine.begin() as conn: + await _cleanup(conn, photo_id=photo_id, event_id=event_id) + try: + await bucket.delete(storage_key) + except Exception: + pass diff --git a/tests/fixtures/images/face.jpg b/tests/fixtures/images/face.jpg new file mode 100644 index 0000000..f746a3c Binary files /dev/null and b/tests/fixtures/images/face.jpg differ diff --git a/tests/fixtures/images/group.jpg b/tests/fixtures/images/group.jpg new file mode 100644 index 0000000..7991f1a Binary files /dev/null and b/tests/fixtures/images/group.jpg differ diff --git a/tests/fixtures/images/noface.jpg b/tests/fixtures/images/noface.jpg new file mode 100644 index 0000000..4a58e54 Binary files /dev/null and b/tests/fixtures/images/noface.jpg differ diff --git a/tests/integration/test_enrollment_flow.py b/tests/integration/test_enrollment_flow.py new file mode 100644 index 0000000..836f1ff --- /dev/null +++ b/tests/integration/test_enrollment_flow.py @@ -0,0 +1,102 @@ +""" +Integration tests for the Enrollment Flow. + +These tests use a real PostgreSQL database to verify that +the user's face embedding is correctly persisted in the database. +""" + +import uuid +from unittest.mock import AsyncMock, MagicMock + +import pytest +from sqlalchemy import text + +from app.service.users import AuthService +from app.service.face_embedding import FaceImagePayload +from db.generated import user as user_queries + + +# =========================================================================== +# Fixtures +# =========================================================================== + + +@pytest.fixture +def mock_face_embedding() -> AsyncMock: + from app.service.face_embedding import FaceEmbeddingService + svc = MagicMock(spec=FaceEmbeddingService) + # Return a dummy embedding of size 512 + svc.compute_average_embedding_stream = AsyncMock(return_value=[0.1] * 512) + return svc + + +@pytest.fixture +async def db_conn(): + from sqlalchemy.ext.asyncio import create_async_engine + from app.core.config import settings + url = f"postgresql+asyncpg://{settings.POSTGRES_USER}:{settings.POSTGRES_PASSWORD}@{settings.POSTGRES_HOST}:{settings.POSTGRES_PORT}/{settings.POSTGRES_DB}" + engine = create_async_engine(url, pool_pre_ping=True) + async with engine.connect() as conn: + yield conn + await engine.dispose() + +@pytest.fixture +def auth_service(mock_face_embedding: AsyncMock, db_conn) -> AuthService: + from db.generated import session as session_queries + from db.generated import devices as device_queries + + return AuthService( + user_querier=user_queries.AsyncQuerier(db_conn), + session_querier=session_queries.AsyncQuerier(db_conn), + device_querier=device_queries.AsyncQuerier(db_conn), + face_embedding_service=mock_face_embedding, + ) + + +# =========================================================================== +# Tests +# =========================================================================== + +# Mark these as integration tests (require DB) +pytestmark = pytest.mark.integration + + +@pytest.mark.asyncio +async def test_enrollment_persists_embedding( + auth_service: AuthService, + mock_face_embedding: AsyncMock, + db_conn, +) -> None: + """Test the happy path: add_embbed_user updates the user's face_embedding.""" + # 1. Setup: Create a user without an embedding + user = await user_queries.AsyncQuerier(db_conn).create_user( + email=f"test-enroll-{uuid.uuid4()}@multai.com", + hashed_password="hash", + ) + assert user is not None + user_id = user.id + + # 2. Execute enrollment + payload = FaceImagePayload(bytes=b"fake-image", filename="face.jpg", content_type="image/jpeg") + + try: + await auth_service.add_embbed_user( + user_id=user_id, + image_payloads=[payload], + ) + + # 3. Verify: The user should now have an embedding + updated_user = await user_queries.AsyncQuerier(db_conn).get_user_by_id(id=user_id) + assert updated_user is not None + assert updated_user.face_embedding is not None + assert "0.1" in str(updated_user.face_embedding) + + # Face embedding service should have been called + mock_face_embedding.compute_average_embedding_stream.assert_called_once() + + finally: + # Cleanup + await db_conn.execute( + text("DELETE FROM users WHERE id = :uid"), {"uid": user_id} + ) + await db_conn.commit() diff --git a/tests/integration/test_photo_approval_flow.py b/tests/integration/test_photo_approval_flow.py new file mode 100644 index 0000000..2606df8 --- /dev/null +++ b/tests/integration/test_photo_approval_flow.py @@ -0,0 +1,227 @@ +""" +Integration tests for the Photo Approval Flow. + +These tests use a real PostgreSQL database but mock MinIO and NATS. +They verify the full lifecycle of a group photo requiring multi-user approval. +""" + +import uuid +import datetime +from unittest.mock import AsyncMock, MagicMock + +import pytest +from sqlalchemy import text + +from app.service.photo_approval import PhotoApprovalService +from db.generated import user as user_queries +from db.generated import stuff_user as staff_queries +from db.generated import events as event_queries +from db.generated import photos as photo_queries +from db.generated import photo_approvals as approval_queries + + +# =========================================================================== +# Fixtures +# =========================================================================== + + +@pytest.fixture +def mock_storage() -> AsyncMock: + from app.service.staged_upload_storage import StagedUploadStorageService + svc = MagicMock(spec=StagedUploadStorageService) + svc.delete_storage_key = AsyncMock() + return svc + + +@pytest.fixture +def mock_audit() -> AsyncMock: + from app.service.audit import AuditService + svc = MagicMock(spec=AuditService) + svc.create_record = AsyncMock() + return svc + + +@pytest.fixture +async def db_conn(): + from sqlalchemy.ext.asyncio import create_async_engine + from app.core.config import settings + url = f"postgresql+asyncpg://{settings.POSTGRES_USER}:{settings.POSTGRES_PASSWORD}@{settings.POSTGRES_HOST}:{settings.POSTGRES_PORT}/{settings.POSTGRES_DB}" + engine = create_async_engine(url, pool_pre_ping=True) + async with engine.connect() as conn: + yield conn + await engine.dispose() + +@pytest.fixture +def approval_service( + mock_storage: AsyncMock, + mock_audit: AsyncMock, + db_conn, +) -> PhotoApprovalService: + return PhotoApprovalService( + photo_approval_querier=approval_queries.AsyncQuerier(db_conn), + photo_querier=photo_queries.AsyncQuerier(db_conn), + storage_service=mock_storage, + audit_service=mock_audit, + ) + + +# =========================================================================== +# Tests +# =========================================================================== + +# Mark these as integration tests (require DB) +pytestmark = pytest.mark.integration + + +@pytest.mark.asyncio +async def test_group_photo_approval_lifecycle( + approval_service: PhotoApprovalService, + mock_storage: AsyncMock, + mock_audit: AsyncMock, + db_conn, +) -> None: + """Test that a photo becomes 'approved' only when all pending approvals are approved.""" + event_id = uuid.uuid4() + photo_id = uuid.uuid4() + + sq = staff_queries.AsyncQuerier(db_conn) + uq = user_queries.AsyncQuerier(db_conn) + eq = event_queries.AsyncQuerier(db_conn) + pq = photo_queries.AsyncQuerier(db_conn) + aq = approval_queries.AsyncQuerier(db_conn) + + staff = await sq.create_admin(email=f"admin-{uuid.uuid4()}@test.com", password="hash") + event_creator_id = staff.id + + user_ids = [] + for i in range(3): + u = await uq.create_user(email=f"approval-{uuid.uuid4()}@test.com", hashed_password="hash") + user_ids.append(u.id) + + uploader_id, user1_id, user2_id = user_ids + + event = await eq.create_event( + event_queries.CreateEventParams( + name="Approval Test Event", + event_code=f"APP{str(event_id)[:4]}", + event_date=datetime.datetime.now(datetime.timezone.utc), + status="scheduled", + created_by=event_creator_id + ) + ) + event_id = event.id + + await pq.create_photo( + photo_queries.CreatePhotoParams( + event_id=event_id, + storage_key="test/group.jpg", + taken_at=None, + day_number=None, + visibility="public" + ) + ) + + # Set status to pending and id + await db_conn.execute( + text(f"UPDATE photos SET id = '{photo_id}', status = 'pending', uploaded_by = '{uploader_id}' WHERE storage_key = 'test/group.jpg'") + ) + + await aq.create_photo_approval(photo_id=photo_id, user_id=user1_id, decision="pending") + await aq.create_photo_approval(photo_id=photo_id, user_id=user2_id, decision="pending") + + try: + # 2. User 1 approves + result1 = await approval_service.decide(photo_id=photo_id, user_id=user1_id, decision="approved") + assert result1 == "pending", "Photo should remain pending because User 2 hasn't approved yet" + + photo = await photo_queries.AsyncQuerier(db_conn).get_photo_by_id(id=photo_id) + assert photo.status == "pending" + + # 3. User 2 approves + result2 = await approval_service.decide(photo_id=photo_id, user_id=user2_id, decision="approved") + assert result2 == "approved", "Photo should be approved since all users approved" + + photo = await photo_queries.AsyncQuerier(db_conn).get_photo_by_id(id=photo_id) + assert photo.status == "approved" + + finally: + # 4. Cleanup + await db_conn.execute(text(f"DELETE FROM photo_approvals WHERE photo_id = '{photo_id}'")) + await db_conn.execute(text(f"DELETE FROM photos WHERE id = '{photo_id}'")) + await db_conn.execute(text(f"DELETE FROM events WHERE id = '{event_id}'")) + await db_conn.execute(text(f"DELETE FROM users WHERE id IN ('{user1_id}', '{user2_id}')")) + await db_conn.execute(text(f"DELETE FROM staff_users WHERE id = '{event_creator_id}'")) + await db_conn.commit() + + +@pytest.mark.asyncio +async def test_group_photo_rejection_deletes_storage( + approval_service: PhotoApprovalService, + mock_storage: AsyncMock, + db_conn, +) -> None: + """Test that a single rejection sets the photo to 'rejected' and deletes from MinIO.""" + event_id = uuid.uuid4() + photo_id = uuid.uuid4() + + sq = staff_queries.AsyncQuerier(db_conn) + uq = user_queries.AsyncQuerier(db_conn) + eq = event_queries.AsyncQuerier(db_conn) + pq = photo_queries.AsyncQuerier(db_conn) + aq = approval_queries.AsyncQuerier(db_conn) + + staff = await sq.create_admin(email=f"admin-{uuid.uuid4()}@test.com", password="hash") + event_creator_id = staff.id + + user_ids = [] + for i in range(2): + u = await uq.create_user(email=f"reject-{uuid.uuid4()}@test.com", hashed_password="hash") + user_ids.append(u.id) + + uploader_id, user1_id = user_ids + + event = await eq.create_event( + event_queries.CreateEventParams( + name="Reject Test Event", + event_code=f"REJ{str(event_id)[:4]}", + event_date=datetime.datetime.now(datetime.timezone.utc), + status="scheduled", + created_by=event_creator_id + ) + ) + event_id = event.id + + await pq.create_photo( + photo_queries.CreatePhotoParams( + event_id=event_id, + storage_key="test/reject.jpg", + taken_at=None, + day_number=None, + visibility="public" + ) + ) + + await db_conn.execute( + text(f"UPDATE photos SET id = '{photo_id}', status = 'pending', uploaded_by = '{uploader_id}' WHERE storage_key = 'test/reject.jpg'") + ) + + await aq.create_photo_approval(photo_id=photo_id, user_id=user1_id, decision="pending") + + try: + # 2. User 1 rejects + result = await approval_service.decide(photo_id=photo_id, user_id=user1_id, decision="rejected") + + # 3. Verify + assert result == "rejected" + mock_storage.delete_storage_key.assert_called_once_with("test/reject.jpg") + + photo = await photo_queries.AsyncQuerier(db_conn).get_photo_by_id(id=photo_id) + assert photo.status == "rejected" + + finally: + await db_conn.execute(text(f"DELETE FROM photo_approvals WHERE photo_id = '{photo_id}'")) + await db_conn.execute(text(f"DELETE FROM photos WHERE id = '{photo_id}'")) + await db_conn.execute(text(f"DELETE FROM events WHERE id = '{event_id}'")) + await db_conn.execute(text(f"DELETE FROM users WHERE id IN ('{uploader_id}', '{user1_id}')")) + await db_conn.execute(text(f"DELETE FROM staff_users WHERE id = '{event_creator_id}'")) + await db_conn.commit() diff --git a/tests/security/test_auth_security.py b/tests/security/test_auth_security.py new file mode 100644 index 0000000..d010e70 --- /dev/null +++ b/tests/security/test_auth_security.py @@ -0,0 +1,160 @@ +import uuid +import jwt +from datetime import datetime, timedelta, timezone + +import pytest +from httpx import AsyncClient, ASGITransport +from sqlalchemy import text + +from app.main import app +from app.core.config import settings +from db.generated import user as user_queries +from app.infra.database import engine +from app.infra.redis import RedisClient +from app.service.session import SessionService + +pytestmark = pytest.mark.asyncio(loop_scope="session") + +@pytest.fixture(scope="session", autouse=True) +async def setup_infra(): + # We must init Redis since ASGITransport doesn't trigger the lifespan + try: + RedisClient.init( + host=settings.REDIS_HOST, + port=settings.REDIS_PORT, + password=settings.REDIS_PASSWORD, + ) + except RuntimeError: + pass # Already initialized + yield + await RedisClient.get_instance().close() + +@pytest.fixture(scope="session") +async def client(): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://testserver") as c: + yield c + +def create_mock_jwt(user_id: str, exp_delta_hours: int = 24) -> str: + payload = { + "sub": user_id, + "exp": datetime.now(timezone.utc) + timedelta(hours=exp_delta_hours), + } + return jwt.encode(payload, settings.jwt_secret, algorithm="HS256") + +async def test_jwt_validation_invalid_signature(client): + """Test that a JWT with an invalid signature is rejected.""" + payload = { + "sub": str(uuid.uuid4()), + "exp": datetime.now(timezone.utc) + timedelta(hours=1), + } + invalid_token = jwt.encode(payload, "wrong_secret_key", algorithm="HS256") + + # We must use "Bearer " + response = await client.get( + "/user/photos", + headers={"Authorization": f"Bearer {invalid_token}"} + ) + assert response.status_code == 401 + assert "Invalid token" in response.text + +async def test_jwt_validation_expired_token(client): + """Test that an expired JWT is rejected.""" + expired_token = create_mock_jwt(str(uuid.uuid4()), exp_delta_hours=-1) + + response = await client.get( + "/user/photos", + headers={"Authorization": f"Bearer {expired_token}"} + ) + assert response.status_code == 401 + assert "Token has expired" in response.text + +async def test_blocked_user_access(client): + """Test that a blocked user cannot access protected endpoints.""" + async with engine.begin() as conn: + uq = user_queries.AsyncQuerier(conn) + user = await uq.create_user( + email=f"blocked-{uuid.uuid4()}@test.com", + hashed_password="hash" + ) + user_id = user.id + + # We need a session ID to put in the JWT, otherwise the auth dependency fails with "Invalid token" + session_id = uuid.uuid4() + + await uq.set_user_blocked(blocked=True, id=user_id) + + redis = RedisClient.get_instance() + await SessionService.cache_session_for_auth( + redis=redis, + session_id=session_id, + user_id=user_id, + email="blocked@test.com", + expires_at=datetime.now(timezone.utc) + timedelta(hours=1), + blocked=True, + ttl=3600, + ) + + payload = { + "session_id": str(session_id), + "exp": datetime.now(timezone.utc) + timedelta(hours=1), + } + token = jwt.encode(payload, settings.jwt_secret, algorithm="HS256") + + try: + response = await client.get( + "/user/photos", + headers={"Authorization": f"Bearer {token}"} + ) + assert response.status_code in (401, 403), f"Expected 401 or 403, got {response.status_code}" + finally: + async with engine.begin() as conn: + await conn.execute(text("DELETE FROM users WHERE id = :uid"), {"uid": user_id}) + +async def test_rate_limiting(client): + """Test that multiple requests within a short timeframe hit rate limits.""" + async with engine.begin() as conn: + uq = user_queries.AsyncQuerier(conn) + user = await uq.create_user( + email=f"rate-{uuid.uuid4()}@test.com", + hashed_password="hash" + ) + user_id = user.id + session_id = uuid.uuid4() + + redis = RedisClient.get_instance() + await SessionService.cache_session_for_auth( + redis=redis, + session_id=session_id, + user_id=user_id, + email="rate@test.com", + expires_at=datetime.now(timezone.utc) + timedelta(hours=1), + blocked=False, + ttl=3600, + ) + + payload = { + "session_id": str(session_id), + "exp": datetime.now(timezone.utc) + timedelta(hours=1), + } + token = jwt.encode(payload, settings.jwt_secret, algorithm="HS256") + + try: + responses = [] + # We test with enough requests to hit the 20/min limit + for _ in range(25): + res = await client.get( + "/user/photos", + headers={"Authorization": f"Bearer {token}"} + ) + responses.append(res.status_code) + + assert 429 in responses, "Expected to hit rate limit (429) after multiple rapid requests" + finally: + async with engine.begin() as conn: + await conn.execute(text("DELETE FROM users WHERE id = :uid"), {"uid": user_id}) + try: + redis = RedisClient.get_instance() + await redis._client.delete("rate_limit:/user/photos:127.0.0.1") + await redis._client.delete("rate_limit:/user/photos:testclient") + except Exception: + pass diff --git a/tests/unit/test_auth_email_otp.py b/tests/unit/test_auth_email_otp.py index 9a6f825..008a13a 100644 --- a/tests/unit/test_auth_email_otp.py +++ b/tests/unit/test_auth_email_otp.py @@ -139,7 +139,7 @@ async def test_mobile_register_resend_otp_success( mock_redis.incr.return_value = 1 # Act - with patch("app.infra.nats.NatsClient.publish", new_callable=AsyncMock) as mock_publish: + with patch("app.service.users.NatsClient.publish", new_callable=AsyncMock) as mock_publish: res = await auth_service.mobile_register_resend_otp(redis=mock_redis, email=email) # Assert diff --git a/tests/unit/test_auth_service.py b/tests/unit/test_auth_service.py new file mode 100644 index 0000000..2024ba6 --- /dev/null +++ b/tests/unit/test_auth_service.py @@ -0,0 +1,460 @@ +""" +Unit tests for AuthService. + +Tests cover the core mobile auth flow: login, registration, password validation, +blocked user enforcement, session limits, logout, refresh token, and face embedding. +All dependencies (DB queriers, Redis, FaceEmbeddingService) are mocked. +""" + +import uuid +from datetime import datetime, timedelta, timezone +from unittest.mock import AsyncMock, MagicMock + +import pytest +from fastapi.exceptions import HTTPException + +from app.service.users import AuthService +from app.core.securite import hash_password +from app.schema.request.mobile.auth import MobileLoginRequest, MobileRegisterRequest + + +# --------------------------------------------------------------------------- +# Factories +# --------------------------------------------------------------------------- + + +def _make_user( + *, + user_id: uuid.UUID | None = None, + email: str = "user@test.com", + password: str = "Secret123!", + blocked: bool = False, + face_embedding: str | None = None, +) -> MagicMock: + u = MagicMock() + u.id = user_id or uuid.uuid4() + u.email = email + u.hashed_password = hash_password(password) + u.blocked = blocked + u.face_embedding = face_embedding + u.display_name = None + return u + + +def _make_session( + *, + session_id: uuid.UUID | None = None, + user_id: uuid.UUID | None = None, + expires_at: datetime | None = None, +) -> MagicMock: + s = MagicMock() + s.id = session_id or uuid.uuid4() + s.user_id = user_id or uuid.uuid4() + s.device_id = uuid.uuid4() + s.expires_at = expires_at or datetime.now(timezone.utc) + timedelta(days=30) + s.last_active = datetime.now(timezone.utc) + return s + + +def _make_device() -> MagicMock: + d = MagicMock() + d.id = uuid.uuid4() + d.user_id = uuid.uuid4() + d.is_invalid_token = False + d.is_active = True + return d + + +def _make_login_request( + *, + email: str = "user@test.com", + password: str = "Secret123!", +) -> MobileLoginRequest: + return MobileLoginRequest( + email=email, + password=password, + device_id=uuid.uuid4(), + device_name="iPhone 15", + device_type="ios", + ) + + +def _make_register_request( + *, + email: str = "user@test.com", + password: str = "Secret123!", +) -> MobileRegisterRequest: + return MobileRegisterRequest( + email=email, + password=password, + device_id=uuid.uuid4(), + device_name="iPhone 15", + device_type="ios", + ) + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def user_querier() -> AsyncMock: + from db.generated import user as user_queries + q = MagicMock(spec=user_queries.AsyncQuerier) + q.get_user_by_email = AsyncMock(return_value=None) + q.create_user = AsyncMock() + q.get_user_by_id = AsyncMock() + q.find_closest_user_by_embedding = AsyncMock(return_value=None) + q.set_user_embedding = AsyncMock() + return q + + +@pytest.fixture +def device_querier() -> AsyncMock: + from db.generated import devices as device_queries + q = MagicMock(spec=device_queries.AsyncQuerier) + q.get_device_by_id = AsyncMock(return_value=None) + q.create_device = AsyncMock(return_value=_make_device()) + q.activate_device = AsyncMock() + return q + + +@pytest.fixture +def session_querier() -> AsyncMock: + from db.generated import session as session_queries + q = MagicMock(spec=session_queries.AsyncQuerier) + q.count_user_sessions = AsyncMock(return_value=0) + q.upsert_session = AsyncMock(return_value=_make_session()) + q.get_session_by_id = AsyncMock() + return q + + +@pytest.fixture +def face_service() -> AsyncMock: + from app.service.face_embedding import FaceEmbeddingService + svc = MagicMock(spec=FaceEmbeddingService) + svc.compute_average_embedding = AsyncMock(return_value=[0.1] * 512) + return svc + + +@pytest.fixture +def redis() -> AsyncMock: + r = MagicMock() + r.set = AsyncMock() + r.get = AsyncMock(return_value=None) + r.delete = AsyncMock() + r.incr = AsyncMock(return_value=1) + r.expire = AsyncMock() + return r + + +@pytest.fixture +def auth_service( + user_querier: AsyncMock, + device_querier: AsyncMock, + session_querier: AsyncMock, + face_service: AsyncMock, +) -> AuthService: + return AuthService( + user_querier=user_querier, + device_querier=device_querier, + session_querier=session_querier, + face_embedding_service=face_service, + ) + + +# =========================================================================== +# 1. Registration β€” new user +# =========================================================================== + + +class TestRegisterNewUser: + @pytest.mark.asyncio + async def test_new_user_is_created( + self, + auth_service: AuthService, + user_querier: AsyncMock, + redis: AsyncMock, + ) -> None: + new_user = _make_user() + user_querier.get_user_by_email.return_value = None + user_querier.create_user.return_value = new_user + + req = _make_register_request() + result = await auth_service.mobile_register(redis, req) + + user_querier.create_user.assert_not_called() + assert result.status == "pending_verification" + + @pytest.mark.asyncio + async def test_pending_status_returned_on_register( + self, + auth_service: AuthService, + user_querier: AsyncMock, + redis: AsyncMock, + ) -> None: + user_querier.get_user_by_email.return_value = None + + result = await auth_service.mobile_register(redis, _make_register_request()) + + assert result.status == "pending_verification" + assert result.message == "OTP sent to email" + + @pytest.mark.asyncio + async def test_session_cached_in_redis_on_register( + self, + auth_service: AuthService, + user_querier: AsyncMock, + redis: AsyncMock, + ) -> None: + new_user = _make_user() + user_querier.get_user_by_email.return_value = None + user_querier.create_user.return_value = new_user + + await auth_service.mobile_register(redis, _make_register_request()) + + # Redis.set must be called at least once (session key) + redis.set.assert_called() + + +# =========================================================================== +# 2. Login β€” existing user +# =========================================================================== + + +class TestLoginExistingUser: + @pytest.mark.asyncio + async def test_valid_credentials_return_tokens( + self, + auth_service: AuthService, + user_querier: AsyncMock, + redis: AsyncMock, + ) -> None: + existing = _make_user(password="Correctpass1!") + user_querier.get_user_by_email.return_value = existing + + result = await auth_service.mobile_login( + redis, _make_login_request(password="Correctpass1!") + ) + + assert result.is_new_user is False + assert result.access_token + + @pytest.mark.asyncio + async def test_wrong_password_raises_401( + self, + auth_service: AuthService, + user_querier: AsyncMock, + redis: AsyncMock, + ) -> None: + existing = _make_user(password="Rightpassword1!") + user_querier.get_user_by_email.return_value = existing + + with pytest.raises(HTTPException) as exc_info: + await auth_service.mobile_login( + redis, _make_login_request(password="Wrongpassword1!") + ) + assert exc_info.value.status_code == 401 + + @pytest.mark.asyncio + async def test_blocked_user_raises_403( + self, + auth_service: AuthService, + user_querier: AsyncMock, + redis: AsyncMock, + ) -> None: + blocked = _make_user(password="Secret123!", blocked=True) + user_querier.get_user_by_email.return_value = blocked + + with pytest.raises(HTTPException) as exc_info: + await auth_service.mobile_login(redis, _make_login_request()) + assert exc_info.value.status_code == 403 + + +# =========================================================================== +# 3. Session limit enforcement +# =========================================================================== + + +class TestSessionLimit: + @pytest.mark.asyncio + async def test_exceeding_session_limit_raises_403( + self, + auth_service: AuthService, + user_querier: AsyncMock, + session_querier: AsyncMock, + redis: AsyncMock, + ) -> None: + user = _make_user() + user_querier.get_user_by_email.return_value = user + # Return a count >= SESSION_LIMIT + session_querier.count_user_sessions.return_value = AuthService.SESSION_LIMIT + + with pytest.raises(HTTPException) as exc_info: + await auth_service.mobile_login(redis, _make_login_request()) + assert exc_info.value.status_code == 403 + assert "session limit" in exc_info.value.detail.lower() + + @pytest.mark.asyncio + async def test_within_session_limit_succeeds( + self, + auth_service: AuthService, + user_querier: AsyncMock, + session_querier: AsyncMock, + redis: AsyncMock, + ) -> None: + user = _make_user() + user_querier.get_user_by_email.return_value = user + session_querier.count_user_sessions.return_value = AuthService.SESSION_LIMIT - 1 + + result = await auth_service.mobile_login(redis, _make_login_request()) + assert result.access_token + + +# =========================================================================== +# 4. Logout +# =========================================================================== + + +class TestLogout: + @pytest.mark.asyncio + async def test_logout_deletes_session_key_from_redis( + self, + auth_service: AuthService, + redis: AsyncMock, + ) -> None: + user_id = str(uuid.uuid4()) + session_id = str(uuid.uuid4()) + + await auth_service.logout(redis, user_id, session_id) + + redis.delete.assert_called_once() + key_used = redis.delete.call_args.args[0] + assert user_id in key_used + + @pytest.mark.asyncio + async def test_logout_returns_success_message( + self, + auth_service: AuthService, + redis: AsyncMock, + ) -> None: + result = await auth_service.logout(redis, str(uuid.uuid4()), str(uuid.uuid4())) + assert "message" in result + assert "logged out" in result["message"].lower() + + +# =========================================================================== +# 5. Refresh token +# =========================================================================== + + +class TestRefreshToken: + @pytest.mark.asyncio + async def test_valid_refresh_returns_new_tokens( + self, + auth_service: AuthService, + user_querier: AsyncMock, + session_querier: AsyncMock, + redis: AsyncMock, + ) -> None: + from app.core.securite import create_refresh_mobile_token + + session = _make_session() + session_querier.get_session_by_id.return_value = session + user_querier.get_user_by_id.return_value = _make_user(user_id=session.user_id) + + refresh_token = create_refresh_mobile_token(str(session.id)) + result = await auth_service.refresh_token(redis, refresh_token) + + assert result.access_token + assert result.refresh_token + + @pytest.mark.asyncio + async def test_expired_session_raises_401( + self, + auth_service: AuthService, + user_querier: AsyncMock, + session_querier: AsyncMock, + redis: AsyncMock, + ) -> None: + from app.core.securite import create_refresh_mobile_token + + past_session = _make_session( + expires_at=datetime.now(timezone.utc) - timedelta(days=1) + ) + session_querier.get_session_by_id.return_value = past_session + + refresh_token = create_refresh_mobile_token(str(past_session.id)) + + with pytest.raises(HTTPException) as exc_info: + await auth_service.refresh_token(redis, refresh_token) + assert exc_info.value.status_code == 401 + + @pytest.mark.asyncio + async def test_blocked_user_on_refresh_raises_403( + self, + auth_service: AuthService, + user_querier: AsyncMock, + session_querier: AsyncMock, + redis: AsyncMock, + ) -> None: + from app.core.securite import create_refresh_mobile_token + + session = _make_session() + session_querier.get_session_by_id.return_value = session + user_querier.get_user_by_id.return_value = _make_user( + user_id=session.user_id, blocked=True + ) + + refresh_token = create_refresh_mobile_token(str(session.id)) + + with pytest.raises(HTTPException) as exc_info: + await auth_service.refresh_token(redis, refresh_token) + assert exc_info.value.status_code == 403 + + @pytest.mark.asyncio + async def test_invalid_refresh_token_raises_401( + self, + auth_service: AuthService, + redis: AsyncMock, + ) -> None: + with pytest.raises(HTTPException) as exc_info: + await auth_service.refresh_token(redis, "completely.invalid.token") + assert exc_info.value.status_code == 401 + + +# =========================================================================== +# 6. find_closest_user +# =========================================================================== + + +class TestFindClosestUser: + @pytest.mark.asyncio + async def test_returns_none_when_no_row( + self, + auth_service: AuthService, + user_querier: AsyncMock, + ) -> None: + user_querier.find_closest_user_by_embedding.return_value = None + + result = await auth_service.find_closest_user(embedding_literal="[0.1, 0.2]") + + assert result is None + + @pytest.mark.asyncio + async def test_returns_closest_user_match( + self, + auth_service: AuthService, + user_querier: AsyncMock, + ) -> None: + row = MagicMock() + row.id = uuid.uuid4() + row.distance = 0.25 + user_querier.find_closest_user_by_embedding.return_value = row + + result = await auth_service.find_closest_user(embedding_literal="[0.1, 0.2]") + + assert result is not None + assert result.user_id == row.id + assert result.distance == 0.25 diff --git a/tests/unit/test_enroll_security.py b/tests/unit/test_enroll_security.py new file mode 100644 index 0000000..4f31584 --- /dev/null +++ b/tests/unit/test_enroll_security.py @@ -0,0 +1,306 @@ +""" +Unit tests for the enrollment security helpers introduced in fix/enroll. + +These helpers only exist on the fix/enroll branch. On other branches, +all tests are automatically skipped via pytest.importorskip. + +Run with: uv run pytest tests/unit/test_enroll_security.py -v +""" + +import io +import uuid +from unittest.mock import AsyncMock, MagicMock + +import pytest +from fastapi import UploadFile +from fastapi.exceptions import HTTPException + +# Guard: skip gracefully if the helpers don't exist on this branch +_router_mod = pytest.importorskip( + "app.router.mobile.enrollement", + reason="fix/enroll branch required for enrollment security helpers", +) + +_sanitise_filename = getattr(_router_mod, "_sanitise_filename", None) +_validate_dimensions = getattr(_router_mod, "_validate_dimensions", None) +_precheck_upload_headers = getattr(_router_mod, "_precheck_upload_headers", None) +_enrollment_lock_key = getattr(_router_mod, "_enrollment_lock_key", None) +read_limited = getattr(_router_mod, "read_limited", None) + +# If any helper is missing, skip the whole module +if any(fn is None for fn in [_sanitise_filename, _validate_dimensions, + _precheck_upload_headers, _enrollment_lock_key, read_limited]): + pytest.skip( + "Enrollment security helpers not available on this branch", + allow_module_level=True, + ) + +from app.core.constant import MAX_IMAGE_SIZE, MIN_IMAGE_DIM, MAX_IMAGE_DIM # noqa: E402 + + +# --------------------------------------------------------------------------- +# Factories +# --------------------------------------------------------------------------- + + +def _make_jpeg_bytes(width: int = 200, height: int = 200) -> bytes: + """Generate a minimal valid JPEG in memory using Pillow.""" + from PIL import Image + + img = Image.new("RGB", (width, height), color=(100, 149, 237)) + buf = io.BytesIO() + img.save(buf, format="JPEG") + return buf.getvalue() + + +def _make_upload_file( + content: bytes, + filename: str = "face.jpg", + content_type: str | None = "image/jpeg", + content_length: int | None = None, +) -> UploadFile: + headers: dict[str, str] = {} + if content_length is not None: + headers["content-length"] = str(content_length) + + buf = io.BytesIO(content) + mock_file = MagicMock(spec=UploadFile) + mock_file.filename = filename + mock_file.content_type = content_type + mock_file.headers = headers + mock_file.read = AsyncMock(side_effect=lambda n=-1: buf.read(n) if n == -1 else buf.read(n)) + mock_file.seek = AsyncMock(side_effect=lambda pos: buf.seek(pos)) + return mock_file # type: ignore[return-value] + + +# =========================================================================== +# 1. _sanitise_filename +# =========================================================================== + + +class TestSanitiseFilename: + def test_normal_name_gets_uuid_prefix(self) -> None: + result = _sanitise_filename("portrait.jpg", "jpg") + parts = result.split("_", 1) + assert len(parts) == 2 + uuid.UUID(parts[0]) # raises if not a valid UUID + assert parts[1] == "portrait.jpg" + + def test_path_traversal_is_neutralised(self) -> None: + result = _sanitise_filename("../../../etc/passwd.jpg", "jpg") + assert ".." not in result + assert "/" not in result + + def test_null_bytes_are_replaced(self) -> None: + assert "\x00" not in _sanitise_filename("face\x00evil.jpg", "jpg") + + def test_control_characters_are_replaced(self) -> None: + assert "\x1f" not in _sanitise_filename("face\x1fmalicious.jpg", "jpg") + + def test_windows_reserved_chars_are_replaced(self) -> None: + for char in r'\/:*?"<>|': + assert char not in _sanitise_filename(f"face{char}name.jpg", "jpg"), \ + f"char {char!r} must be replaced" + + def test_none_filename_returns_uuid_only(self) -> None: + result = _sanitise_filename(None, "png") + base, ext = result.rsplit(".", 1) + uuid.UUID(base) + assert ext == "png" + + def test_empty_filename_returns_uuid_only(self) -> None: + result = _sanitise_filename("", "jpg") + base, ext = result.rsplit(".", 1) + uuid.UUID(base) + assert ext == "jpg" + + def test_long_filename_is_truncated(self) -> None: + result = _sanitise_filename("a" * 200, "jpg") + name_part = result.split("_", 1)[1] + assert len(name_part) <= 128 + + def test_leading_dots_stripped(self) -> None: + result = _sanitise_filename("...hidden.jpg", "jpg") + name_part = result.split("_", 1)[1] + assert not name_part.startswith(".") + + +# =========================================================================== +# 2. _validate_dimensions +# =========================================================================== + + +class TestValidateDimensions: + def test_valid_image_passes(self) -> None: + _validate_dimensions(_make_jpeg_bytes(200, 200)) # must not raise + + def test_too_small_width_raises_400(self) -> None: + with pytest.raises(HTTPException) as exc_info: + _validate_dimensions(_make_jpeg_bytes(MIN_IMAGE_DIM - 1, 200)) + assert exc_info.value.status_code == 400 + assert "too small" in exc_info.value.detail.lower() + + def test_too_small_height_raises_400(self) -> None: + with pytest.raises(HTTPException) as exc_info: + _validate_dimensions(_make_jpeg_bytes(200, MIN_IMAGE_DIM - 1)) + assert exc_info.value.status_code == 400 + + def test_too_large_width_raises_400(self) -> None: + with pytest.raises(HTTPException) as exc_info: + _validate_dimensions(_make_jpeg_bytes(MAX_IMAGE_DIM + 1, 200)) + assert exc_info.value.status_code == 400 + assert "too large" in exc_info.value.detail.lower() + + def test_corrupt_bytes_raises_400(self) -> None: + with pytest.raises(HTTPException) as exc_info: + _validate_dimensions(b"this-is-not-an-image") + assert exc_info.value.status_code == 400 + + def test_boundary_min_dimension_passes(self) -> None: + _validate_dimensions(_make_jpeg_bytes(MIN_IMAGE_DIM, MIN_IMAGE_DIM)) + + def test_boundary_max_dimension_passes(self) -> None: + _validate_dimensions(_make_jpeg_bytes(MAX_IMAGE_DIM, MAX_IMAGE_DIM)) + + +# =========================================================================== +# 3. _precheck_upload_headers +# =========================================================================== + + +class TestPrecheckUploadHeaders: + def test_valid_jpeg_header_passes(self) -> None: + _precheck_upload_headers(_make_upload_file(b"", content_type="image/jpeg")) + + def test_valid_png_header_passes(self) -> None: + _precheck_upload_headers(_make_upload_file(b"", content_type="image/png")) + + def test_missing_content_type_raises_400(self) -> None: + f = _make_upload_file(b"", content_type=None) + with pytest.raises(HTTPException) as exc_info: + _precheck_upload_headers(f) + assert exc_info.value.status_code == 400 + + def test_unsupported_content_type_raises_400(self) -> None: + with pytest.raises(HTTPException) as exc_info: + _precheck_upload_headers(_make_upload_file(b"", content_type="application/pdf")) + assert exc_info.value.status_code == 400 + + def test_content_type_with_charset_param_accepted(self) -> None: + # "image/jpeg; charset=utf-8" should normalise to "image/jpeg" + _precheck_upload_headers( + _make_upload_file(b"", content_type="image/jpeg; charset=utf-8") + ) + + def test_oversized_content_length_raises_400(self) -> None: + with pytest.raises(HTTPException) as exc_info: + _precheck_upload_headers( + _make_upload_file(b"", content_type="image/jpeg", content_length=MAX_IMAGE_SIZE + 1) + ) + assert exc_info.value.status_code == 400 + + def test_valid_content_length_passes(self) -> None: + _precheck_upload_headers( + _make_upload_file(b"", content_type="image/jpeg", content_length=1024) + ) + + def test_invalid_content_length_string_raises_400(self) -> None: + f = _make_upload_file(b"", content_type="image/jpeg") + f.headers = {"content-length": "not_a_number"} + with pytest.raises(HTTPException) as exc_info: + _precheck_upload_headers(f) + assert exc_info.value.status_code == 400 + + +# =========================================================================== +# 4. read_limited +# =========================================================================== + + +class TestReadLimited: + @pytest.mark.asyncio + async def test_small_file_returns_full_bytes(self) -> None: + data = b"hello world" + result = await read_limited(_make_upload_file(data), MAX_IMAGE_SIZE) + assert result == data + + @pytest.mark.asyncio + async def test_exceeds_limit_raises_400(self) -> None: + data = b"x" * (MAX_IMAGE_SIZE + 1) + with pytest.raises(HTTPException) as exc_info: + await read_limited(_make_upload_file(data), MAX_IMAGE_SIZE) + assert exc_info.value.status_code == 400 + + @pytest.mark.asyncio + async def test_exactly_at_limit_passes(self) -> None: + data = b"x" * MAX_IMAGE_SIZE + result = await read_limited(_make_upload_file(data), MAX_IMAGE_SIZE) + assert len(result) == MAX_IMAGE_SIZE + + @pytest.mark.asyncio + async def test_empty_file_returns_empty_bytes(self) -> None: + result = await read_limited(_make_upload_file(b""), MAX_IMAGE_SIZE) + assert result == b"" + + @pytest.mark.asyncio + async def test_aborts_early_without_reading_entire_stream(self) -> None: + """Must abort after exceeding the limit β€” not exhaust the stream.""" + call_count = 0 + chunk = b"x" * 65536 + + async def mock_read(n: int = -1) -> bytes: + nonlocal call_count + call_count += 1 + if call_count > 10: + return b"" + return chunk + + f = MagicMock(spec=UploadFile) + f.read = mock_read + f.seek = AsyncMock() + + limit = 65536 * 5 # 5 chunks + with pytest.raises(HTTPException): + await read_limited(f, limit) + + assert call_count <= 7, "read_limited must abort early" + + +# =========================================================================== +# 5. _enrollment_lock_key +# =========================================================================== + + +class TestEnrollmentLockKey: + def test_key_contains_user_id(self) -> None: + user_id = uuid.uuid4() + assert str(user_id) in _enrollment_lock_key(user_id) + + def test_different_users_have_different_keys(self) -> None: + assert _enrollment_lock_key(uuid.uuid4()) != _enrollment_lock_key(uuid.uuid4()) + + def test_key_format(self) -> None: + user_id = uuid.uuid4() + assert _enrollment_lock_key(user_id) == f"enroll:in_progress:{user_id}" + + +# =========================================================================== +# 6. Magic byte sniffing (unit-level verification) +# =========================================================================== + + +class TestMagicByteSniffing: + def test_pdf_bytes_not_classified_as_image(self) -> None: + import filetype # type: ignore[import-untyped] + + pdf_bytes = b"%PDF-1.4 fake pdf content" + kind = filetype.guess(pdf_bytes) + allowed = {"image/jpeg", "image/png", "image/heic", "image/heif"} + assert kind is None or kind.mime not in allowed + + def test_real_jpeg_classified_as_jpeg(self) -> None: + import filetype # type: ignore[import-untyped] + + kind = filetype.guess(_make_jpeg_bytes(100, 100)) + assert kind is not None + assert kind.mime == "image/jpeg" diff --git a/tests/unit/test_face_match_service.py b/tests/unit/test_face_match_service.py new file mode 100644 index 0000000..8de215b --- /dev/null +++ b/tests/unit/test_face_match_service.py @@ -0,0 +1,404 @@ +""" +Unit tests for SingleFaceMatchService. + +All database and service collaborators are mocked β€” no live infrastructure required. +Tests cover all decision branches: auto-approve, threshold rejection, happy-path match, +idempotency guards, and notification side-effects. +""" + +import json +import uuid +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from app.service.face_match import SingleFaceMatchService +from app.schema.internal.single_face_match import BBoxPayload, SingleFaceMatchJob + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def photo_face_querier() -> AsyncMock: + from db.generated import photo_faces as pf_queries + q = MagicMock(spec=pf_queries.AsyncQuerier) + q.photo_faces_photo_exists = AsyncMock(return_value=object()) # truthy β†’ photo exists + q.photo_faces_match_exists_for_photo = AsyncMock(return_value=None) # no duplicate + q.photo_faces_ensure_face_match = AsyncMock() + return q + + +@pytest.fixture +def photo_querier() -> AsyncMock: + from db.generated import photos as photo_queries + q = MagicMock(spec=photo_queries.AsyncQuerier) + q.update_photo_status = AsyncMock(return_value=None) + return q + + +@pytest.fixture +def user_match_service() -> AsyncMock: + from app.service.users import AuthService + svc = MagicMock(spec=AuthService) + svc.find_closest_user = AsyncMock() + return svc + + +@pytest.fixture +def notification_service() -> AsyncMock: + from app.service.user_notification import UserNotificationService + svc = MagicMock(spec=UserNotificationService) + svc.create_notification = AsyncMock() + return svc + + +@pytest.fixture +def service( + photo_face_querier: AsyncMock, + photo_querier: AsyncMock, + user_match_service: AsyncMock, + notification_service: AsyncMock, +) -> SingleFaceMatchService: + return SingleFaceMatchService( + conn=MagicMock(), + photo_face_querier=photo_face_querier, + photo_querier=photo_querier, + user_match_service=user_match_service, + user_notification_service=notification_service, + ) + + +@pytest.fixture +def job() -> SingleFaceMatchJob: + return SingleFaceMatchJob( + photo_id=uuid.uuid4(), + image_ref="photos/test.jpg", + face_index=0, + ) + + +@pytest.fixture +def bbox() -> BBoxPayload: + return BBoxPayload(x1=10.0, y1=20.0, x2=100.0, y2=200.0) + + +def _make_embedding(value: float = 0.5) -> list[float]: + return [value] * 512 + + +def _closest_match(distance: float = 0.3) -> object: + from app.schema.internal.single_face_match import ClosestUserMatch + return ClosestUserMatch(user_id=uuid.uuid4(), distance=distance) + + +# =========================================================================== +# 1. Auto-approve β€” no enrolled users in DB +# =========================================================================== + + +class TestAutoApproveNoUsers: + @pytest.mark.asyncio + async def test_photo_approved_when_no_users( + self, + service: SingleFaceMatchService, + job: SingleFaceMatchJob, + photo_querier: AsyncMock, + user_match_service: AsyncMock, + ) -> None: + user_match_service.find_closest_user.return_value = None # empty DB + + await service.process_detected_face(job, _make_embedding(), bbox=None) + + photo_querier.update_photo_status.assert_called_once_with( + id=job.photo_id, status="approved" + ) + + @pytest.mark.asyncio + async def test_no_notification_when_auto_approved( + self, + service: SingleFaceMatchService, + job: SingleFaceMatchJob, + user_match_service: AsyncMock, + notification_service: AsyncMock, + ) -> None: + user_match_service.find_closest_user.return_value = None + + await service.process_detected_face(job, _make_embedding(), bbox=None) + + notification_service.create_notification.assert_not_called() + + +# =========================================================================== +# 2. Auto-approve β€” distance above threshold +# =========================================================================== + + +class TestAutoApproveDistanceThreshold: + @pytest.mark.asyncio + async def test_photo_approved_when_distance_exceeds_threshold( + self, + service: SingleFaceMatchService, + job: SingleFaceMatchJob, + photo_querier: AsyncMock, + user_match_service: AsyncMock, + ) -> None: + from app.worker.photo_worker.settings import settings as worker_settings + + # Distance just above the threshold β†’ no match β†’ auto-approve + bad_match = _closest_match(distance=worker_settings.similarity_threshold + 0.01) + user_match_service.find_closest_user.return_value = bad_match + + await service.process_detected_face(job, _make_embedding(), bbox=None) + + photo_querier.update_photo_status.assert_called_once_with( + id=job.photo_id, status="approved" + ) + + @pytest.mark.asyncio + async def test_no_db_write_when_distance_exceeds_threshold( + self, + service: SingleFaceMatchService, + job: SingleFaceMatchJob, + photo_face_querier: AsyncMock, + user_match_service: AsyncMock, + ) -> None: + from app.worker.photo_worker.settings import settings as worker_settings + + bad_match = _closest_match(distance=worker_settings.similarity_threshold + 0.01) + user_match_service.find_closest_user.return_value = bad_match + + await service.process_detected_face(job, _make_embedding(), bbox=None) + + photo_face_querier.photo_faces_ensure_face_match.assert_not_called() + + +# =========================================================================== +# 3. Happy-path match β†’ notification sent +# =========================================================================== + + +class TestSuccessfulMatch: + @pytest.mark.asyncio + async def test_face_match_stored_in_db( + self, + service: SingleFaceMatchService, + job: SingleFaceMatchJob, + photo_face_querier: AsyncMock, + user_match_service: AsyncMock, + ) -> None: + good_match = _closest_match(distance=0.1) + user_match_service.find_closest_user.return_value = good_match + + face_match_result = MagicMock() + face_match_result.face_match_id = uuid.uuid4() + photo_face_querier.photo_faces_ensure_face_match.return_value = face_match_result + + await service.process_detected_face(job, _make_embedding(), bbox=None) + + photo_face_querier.photo_faces_ensure_face_match.assert_called_once() + + @pytest.mark.asyncio + async def test_photo_status_set_to_approved_on_match( + self, + service: SingleFaceMatchService, + job: SingleFaceMatchJob, + photo_face_querier: AsyncMock, + photo_querier: AsyncMock, + user_match_service: AsyncMock, + ) -> None: + good_match = _closest_match(distance=0.1) + user_match_service.find_closest_user.return_value = good_match + + face_match_result = MagicMock() + face_match_result.face_match_id = uuid.uuid4() + photo_face_querier.photo_faces_ensure_face_match.return_value = face_match_result + + await service.process_detected_face(job, _make_embedding(), bbox=None) + + photo_querier.update_photo_status.assert_called_once_with( + id=job.photo_id, status="approved" + ) + + @pytest.mark.asyncio + async def test_notification_sent_on_successful_match( + self, + service: SingleFaceMatchService, + job: SingleFaceMatchJob, + photo_face_querier: AsyncMock, + user_match_service: AsyncMock, + notification_service: AsyncMock, + ) -> None: + good_match = _closest_match(distance=0.1) + user_match_service.find_closest_user.return_value = good_match + + face_match_result = MagicMock() + face_match_result.face_match_id = uuid.uuid4() + photo_face_querier.photo_faces_ensure_face_match.return_value = face_match_result + + await service.process_detected_face(job, _make_embedding(), bbox=None) + + notification_service.create_notification.assert_called_once() + call_kwargs = notification_service.create_notification.call_args.kwargs + assert call_kwargs["type"] == "face_match" + assert call_kwargs["user_id"] == good_match.user_id # type: ignore[union-attr] + + @pytest.mark.asyncio + async def test_bbox_serialised_as_json( + self, + service: SingleFaceMatchService, + job: SingleFaceMatchJob, + photo_face_querier: AsyncMock, + user_match_service: AsyncMock, + bbox: BBoxPayload, + ) -> None: + good_match = _closest_match(distance=0.1) + user_match_service.find_closest_user.return_value = good_match + + face_match_result = MagicMock() + face_match_result.face_match_id = uuid.uuid4() + photo_face_querier.photo_faces_ensure_face_match.return_value = face_match_result + + await service.process_detected_face(job, _make_embedding(), bbox=bbox) + + call_args = photo_face_querier.photo_faces_ensure_face_match.call_args + params = call_args.args[0] + parsed_bbox = json.loads(params.bbox) + assert parsed_bbox == {"x1": 10.0, "y1": 20.0, "x2": 100.0, "y2": 200.0} + + +# =========================================================================== +# 4. Idempotency guards +# =========================================================================== + + +class TestIdempotencyGuards: + @pytest.mark.asyncio + async def test_skips_if_photo_not_found( + self, + service: SingleFaceMatchService, + job: SingleFaceMatchJob, + photo_face_querier: AsyncMock, + photo_querier: AsyncMock, + ) -> None: + photo_face_querier.photo_faces_photo_exists.return_value = None # not found + + await service.process_detected_face(job, _make_embedding(), bbox=None) + + photo_querier.update_photo_status.assert_not_called() + photo_face_querier.photo_faces_ensure_face_match.assert_not_called() + + @pytest.mark.asyncio + async def test_skips_if_match_already_exists( + self, + service: SingleFaceMatchService, + job: SingleFaceMatchJob, + photo_face_querier: AsyncMock, + photo_querier: AsyncMock, + ) -> None: + photo_face_querier.photo_faces_match_exists_for_photo.return_value = object() # exists + + await service.process_detected_face(job, _make_embedding(), bbox=None) + + photo_querier.update_photo_status.assert_not_called() + photo_face_querier.photo_faces_ensure_face_match.assert_not_called() + + @pytest.mark.asyncio + async def test_skips_if_missing_image_ref( + self, + service: SingleFaceMatchService, + photo_querier: AsyncMock, + ) -> None: + job_no_ref = SingleFaceMatchJob( + photo_id=uuid.uuid4(), + image_ref="", # empty + face_index=0, + ) + + await service.process_detected_face(job_no_ref, _make_embedding(), bbox=None) + + photo_querier.update_photo_status.assert_not_called() + + @pytest.mark.asyncio + async def test_no_notification_if_face_match_already_existed( + self, + service: SingleFaceMatchService, + job: SingleFaceMatchJob, + photo_face_querier: AsyncMock, + user_match_service: AsyncMock, + notification_service: AsyncMock, + ) -> None: + """If ensure_face_match returns a result with face_match_id=None, it means + the match was already there β€” no duplicate notification should be sent.""" + good_match = _closest_match(distance=0.1) + user_match_service.find_closest_user.return_value = good_match + + result_already_existed = MagicMock() + result_already_existed.face_match_id = None # already existed + photo_face_querier.photo_faces_ensure_face_match.return_value = result_already_existed + + await service.process_detected_face(job, _make_embedding(), bbox=None) + + notification_service.create_notification.assert_not_called() + + +# =========================================================================== +# 5. Resilience β€” DB errors don't crash the worker +# =========================================================================== + + +class TestResilience: + @pytest.mark.asyncio + async def test_db_error_is_handled_gracefully( + self, + service: SingleFaceMatchService, + job: SingleFaceMatchJob, + photo_face_querier: AsyncMock, + user_match_service: AsyncMock, + notification_service: AsyncMock, + ) -> None: + from sqlalchemy.exc import SQLAlchemyError + + good_match = _closest_match(distance=0.1) + user_match_service.find_closest_user.return_value = good_match + photo_face_querier.photo_faces_ensure_face_match.side_effect = SQLAlchemyError("DB down") + + # Should not raise β€” worker must stay alive + await service.process_detected_face(job, _make_embedding(), bbox=None) + notification_service.create_notification.assert_not_called() + + @pytest.mark.asyncio + async def test_memory_error_is_handled_gracefully( + self, + service: SingleFaceMatchService, + job: SingleFaceMatchJob, + user_match_service: AsyncMock, + ) -> None: + user_match_service.find_closest_user.side_effect = MemoryError("OOM") + + # Should not raise + await service.process_detected_face(job, _make_embedding(), bbox=None) + + +# =========================================================================== +# 6. Static helpers +# =========================================================================== + + +class TestStaticHelpers: + def test_vector_literal_format(self) -> None: + embedding = [0.1, 0.2, 0.3] + result = SingleFaceMatchService._vector_literal(embedding) + assert result == "[0.1, 0.2, 0.3]" + + def test_serialize_bbox_none_returns_none(self) -> None: + assert SingleFaceMatchService._serialize_bbox(None) is None + + def test_serialize_bbox_valid(self) -> None: + bbox = BBoxPayload(x1=1.0, y1=2.0, x2=3.0, y2=4.0) + result = SingleFaceMatchService._serialize_bbox(bbox) + assert result is not None + parsed = json.loads(result) + assert parsed == {"x1": 1.0, "y1": 2.0, "x2": 3.0, "y2": 4.0} diff --git a/tests/unit/test_photo_approval_lifecycle.py b/tests/unit/test_photo_approval_lifecycle.py index 52bcd68..00577e6 100644 --- a/tests/unit/test_photo_approval_lifecycle.py +++ b/tests/unit/test_photo_approval_lifecycle.py @@ -45,7 +45,7 @@ def mock_staged_upload_storage_service() -> AsyncMock: @pytest.mark.asyncio -async def test_group_photo_auto_approve_no_users( +async def test_group_photo_pending_no_users( mock_conn: AsyncMock, mock_face_embedding_service: AsyncMock, mock_single_face_service: AsyncMock, @@ -84,9 +84,9 @@ async def test_group_photo_auto_approve_no_users( # Verify notifications were NOT sent mock_notification_service.create_notification.assert_not_called() - # Verify photo is marked public and approved - mock_photo_querier.update_photo_status.assert_called_once_with(id=photo_id, status="approved") - mock_photo_querier.update_photo_visibility.assert_called_once_with(id=photo_id, visibility="public") + # Verify photo stays pending (not marked public or approved) + mock_photo_querier.update_photo_status.assert_not_called() + mock_photo_querier.update_photo_visibility.assert_not_called() @pytest.mark.asyncio diff --git a/tests/unit/test_photo_approval_service.py b/tests/unit/test_photo_approval_service.py new file mode 100644 index 0000000..d10c62b --- /dev/null +++ b/tests/unit/test_photo_approval_service.py @@ -0,0 +1,333 @@ +""" +Unit tests for PhotoApprovalService. + +All queriers and the storage service are mocked β€” no live infrastructure. +Tests cover: approve/reject/pending decision logic, storage cleanup on rejection, +audit logging, and error resilience. +""" + +import uuid +from unittest.mock import AsyncMock, MagicMock + +import pytest +from fastapi.exceptions import HTTPException + +from app.service.photo_approval import PhotoApprovalService +from app.core.constant import AuditEventType + + +# --------------------------------------------------------------------------- +# Minimal stubs for DB models +# --------------------------------------------------------------------------- + + +def _make_approval(decision: str, user_id: uuid.UUID | None = None) -> MagicMock: + a = MagicMock() + a.decision = decision + a.user_id = user_id or uuid.uuid4() + return a + + +def _make_photo(storage_key: str = "photos/test.jpg") -> MagicMock: + p = MagicMock() + p.storage_key = storage_key + return p + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture +def approval_querier() -> AsyncMock: + from db.generated import photo_approvals as pa_queries + q = MagicMock(spec=pa_queries.AsyncQuerier) + q.update_photo_approval_decision = AsyncMock() + q.get_photo_approvals_by_photo_id = MagicMock() # async generator + return q + + +@pytest.fixture +def photo_querier() -> AsyncMock: + from db.generated import photos as photo_queries + q = MagicMock(spec=photo_queries.AsyncQuerier) + q.update_photo_status = AsyncMock(return_value=None) + q.get_photo_by_id = AsyncMock(return_value=_make_photo()) + return q + + +@pytest.fixture +def storage_service() -> AsyncMock: + from app.service.staged_upload_storage import StagedUploadStorageService + svc = MagicMock(spec=StagedUploadStorageService) + svc.delete_storage_key = AsyncMock() + return svc + + +@pytest.fixture +def audit_service() -> AsyncMock: + from app.service.audit import AuditService + svc = MagicMock(spec=AuditService) + svc.create_record = AsyncMock() + return svc + + +def _make_service( + approval_querier: AsyncMock, + photo_querier: AsyncMock, + storage_service: AsyncMock, + audit_service: AsyncMock | None = None, +) -> PhotoApprovalService: + return PhotoApprovalService( + photo_approval_querier=approval_querier, + photo_querier=photo_querier, + storage_service=storage_service, + audit_service=audit_service, + ) + + +def _mock_async_iter(items: list[object]): # type: ignore[type-arg] + """Return a MagicMock that behaves like an async for loop.""" + async def _gen(): # type: ignore[return] + for item in items: + yield item + return _gen() + + +# =========================================================================== +# 1. All decisions == "approved" β†’ photo status becomes "approved" +# =========================================================================== + + +class TestApproveDecision: + @pytest.mark.asyncio + async def test_all_approved_sets_photo_status( + self, + approval_querier: AsyncMock, + photo_querier: AsyncMock, + storage_service: AsyncMock, + ) -> None: + photo_id = uuid.uuid4() + user_id = uuid.uuid4() + approvals = [_make_approval("approved"), _make_approval("approved")] + + approval_querier.update_photo_approval_decision.return_value = MagicMock() + approval_querier.get_photo_approvals_by_photo_id.return_value = _mock_async_iter(approvals) + + service = _make_service(approval_querier, photo_querier, storage_service) + result = await service.decide(photo_id=photo_id, user_id=user_id, decision="approved") + + assert result == "approved" + photo_querier.update_photo_status.assert_called_once_with( + id=photo_id, status="approved" + ) + + @pytest.mark.asyncio + async def test_all_approved_does_not_delete_storage( + self, + approval_querier: AsyncMock, + photo_querier: AsyncMock, + storage_service: AsyncMock, + ) -> None: + photo_id = uuid.uuid4() + user_id = uuid.uuid4() + approvals = [_make_approval("approved")] + + approval_querier.update_photo_approval_decision.return_value = MagicMock() + approval_querier.get_photo_approvals_by_photo_id.return_value = _mock_async_iter(approvals) + + service = _make_service(approval_querier, photo_querier, storage_service) + await service.decide(photo_id=photo_id, user_id=user_id, decision="approved") + + storage_service.delete_storage_key.assert_not_called() + + +# =========================================================================== +# 2. One rejection β†’ photo status becomes "rejected" + storage deleted +# =========================================================================== + + +class TestRejectDecision: + @pytest.mark.asyncio + async def test_one_rejection_sets_photo_status_rejected( + self, + approval_querier: AsyncMock, + photo_querier: AsyncMock, + storage_service: AsyncMock, + ) -> None: + photo_id = uuid.uuid4() + user_id = uuid.uuid4() + approvals = [_make_approval("approved"), _make_approval("rejected")] + + approval_querier.update_photo_approval_decision.return_value = MagicMock() + approval_querier.get_photo_approvals_by_photo_id.return_value = _mock_async_iter(approvals) + + service = _make_service(approval_querier, photo_querier, storage_service) + result = await service.decide(photo_id=photo_id, user_id=user_id, decision="rejected") + + assert result == "rejected" + photo_querier.update_photo_status.assert_called_once_with( + id=photo_id, status="rejected" + ) + + @pytest.mark.asyncio + async def test_rejection_triggers_storage_deletion( + self, + approval_querier: AsyncMock, + photo_querier: AsyncMock, + storage_service: AsyncMock, + ) -> None: + photo_id = uuid.uuid4() + user_id = uuid.uuid4() + approvals = [_make_approval("rejected")] + storage_key = "photos/reject-me.jpg" + photo_querier.get_photo_by_id.return_value = _make_photo(storage_key=storage_key) + + approval_querier.update_photo_approval_decision.return_value = MagicMock() + approval_querier.get_photo_approvals_by_photo_id.return_value = _mock_async_iter(approvals) + + service = _make_service(approval_querier, photo_querier, storage_service) + await service.decide(photo_id=photo_id, user_id=user_id, decision="rejected") + + storage_service.delete_storage_key.assert_called_once_with(storage_key) + + @pytest.mark.asyncio + async def test_storage_deletion_failure_does_not_raise( + self, + approval_querier: AsyncMock, + photo_querier: AsyncMock, + storage_service: AsyncMock, + ) -> None: + """Storage errors must be swallowed β€” they should not surface to the client.""" + photo_id = uuid.uuid4() + user_id = uuid.uuid4() + approvals = [_make_approval("rejected")] + + approval_querier.update_photo_approval_decision.return_value = MagicMock() + approval_querier.get_photo_approvals_by_photo_id.return_value = _mock_async_iter(approvals) + storage_service.delete_storage_key.side_effect = Exception("MinIO unavailable") + + service = _make_service(approval_querier, photo_querier, storage_service) + # Must not raise despite MinIO being unavailable + result = await service.decide(photo_id=photo_id, user_id=user_id, decision="rejected") + assert result == "rejected" + + +# =========================================================================== +# 3. Pending approvals β†’ returns "pending", no status update +# =========================================================================== + + +class TestPendingDecision: + @pytest.mark.asyncio + async def test_pending_approval_returns_pending( + self, + approval_querier: AsyncMock, + photo_querier: AsyncMock, + storage_service: AsyncMock, + ) -> None: + photo_id = uuid.uuid4() + user_id = uuid.uuid4() + approvals = [_make_approval("approved"), _make_approval("pending")] + + approval_querier.update_photo_approval_decision.return_value = MagicMock() + approval_querier.get_photo_approvals_by_photo_id.return_value = _mock_async_iter(approvals) + + service = _make_service(approval_querier, photo_querier, storage_service) + result = await service.decide(photo_id=photo_id, user_id=user_id, decision="approved") + + assert result == "pending" + + @pytest.mark.asyncio + async def test_pending_does_not_update_photo_status( + self, + approval_querier: AsyncMock, + photo_querier: AsyncMock, + storage_service: AsyncMock, + ) -> None: + photo_id = uuid.uuid4() + user_id = uuid.uuid4() + approvals = [_make_approval("pending"), _make_approval("pending")] + + approval_querier.update_photo_approval_decision.return_value = MagicMock() + approval_querier.get_photo_approvals_by_photo_id.return_value = _mock_async_iter(approvals) + + service = _make_service(approval_querier, photo_querier, storage_service) + await service.decide(photo_id=photo_id, user_id=user_id, decision="approved") + + photo_querier.update_photo_status.assert_not_called() + + +# =========================================================================== +# 4. Not found β†’ raises 404 +# =========================================================================== + + +class TestNotFound: + @pytest.mark.asyncio + async def test_not_found_raises_404( + self, + approval_querier: AsyncMock, + photo_querier: AsyncMock, + storage_service: AsyncMock, + ) -> None: + photo_id = uuid.uuid4() + approval_querier.update_photo_approval_decision.return_value = None # not found + + service = _make_service(approval_querier, photo_querier, storage_service) + with pytest.raises(HTTPException) as exc_info: + await service.decide( + photo_id=photo_id, user_id=uuid.uuid4(), decision="approved" + ) + assert exc_info.value.status_code == 404 + + +# =========================================================================== +# 5. Audit logging +# =========================================================================== + + +class TestAuditLogging: + @pytest.mark.asyncio + async def test_audit_called_with_correct_event_type( + self, + approval_querier: AsyncMock, + photo_querier: AsyncMock, + storage_service: AsyncMock, + audit_service: AsyncMock, + ) -> None: + photo_id = uuid.uuid4() + user_id = uuid.uuid4() + approvals = [_make_approval("approved")] + + approval_querier.update_photo_approval_decision.return_value = MagicMock() + approval_querier.get_photo_approvals_by_photo_id.return_value = _mock_async_iter(approvals) + + service = _make_service(approval_querier, photo_querier, storage_service, audit_service) + await service.decide(photo_id=photo_id, user_id=user_id, decision="approved") + + audit_service.create_record.assert_called_once() + call_kwargs = audit_service.create_record.call_args.kwargs + assert call_kwargs["event_type"] == AuditEventType.PHOTO_APPROVAL_DECIDED + assert call_kwargs["user_id"] == user_id + assert str(photo_id) in str(call_kwargs["metadata"]) + + @pytest.mark.asyncio + async def test_no_audit_without_audit_service( + self, + approval_querier: AsyncMock, + photo_querier: AsyncMock, + storage_service: AsyncMock, + ) -> None: + """When audit_service=None, no error must be raised.""" + photo_id = uuid.uuid4() + approvals = [_make_approval("approved")] + + approval_querier.update_photo_approval_decision.return_value = MagicMock() + approval_querier.get_photo_approvals_by_photo_id.return_value = _mock_async_iter(approvals) + + # No audit_service passed β†’ audit_service=None + service = _make_service(approval_querier, photo_querier, storage_service, audit_service=None) + await service.decide(photo_id=photo_id, user_id=uuid.uuid4(), decision="approved") + # no assertion needed β€” just must not raise