From 2d6bbf1d13bba5d09f8737c6cc658abd8fe07dfc Mon Sep 17 00:00:00 2001 From: ilia Date: Tue, 14 Jul 2026 18:15:15 -0400 Subject: [PATCH] test: cover 7 previously-untested backend API route modules MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds pytest coverage for metrics, click_log, role_permissions, videos, pending-identifications, pending-linkages, and reported-photos — the last three required extending conftest with a second (auth) test database + DDL mirroring the Prisma schema-auth.prisma tables, since those endpoints read/write DATABASE_URL_AUTH via raw SQL. pending-photos.py is intentionally left uncovered for now (heavy filesystem/hash-based duplicate-detection logic needs its own mocking strategy) and tracked as a follow-up. --- tests/conftest.py | 157 +++++++++++++++++- tests/test_api_click_log.py | 49 ++++++ tests/test_api_metrics.py | 18 ++ tests/test_api_pending_identifications.py | 191 ++++++++++++++++++++++ tests/test_api_pending_linkages.py | 156 ++++++++++++++++++ tests/test_api_reported_photos.py | 157 ++++++++++++++++++ tests/test_api_role_permissions.py | 79 +++++++++ tests/test_api_videos.py | 171 +++++++++++++++++++ 8 files changed, 975 insertions(+), 3 deletions(-) create mode 100644 tests/test_api_click_log.py create mode 100644 tests/test_api_metrics.py create mode 100644 tests/test_api_pending_identifications.py create mode 100644 tests/test_api_pending_linkages.py create mode 100644 tests/test_api_reported_photos.py create mode 100644 tests/test_api_role_permissions.py create mode 100644 tests/test_api_videos.py diff --git a/tests/conftest.py b/tests/conftest.py index f28a63a..8b7cb7f 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -16,7 +16,7 @@ os.environ["SKIP_DEEPFACE_IN_TESTS"] = "1" from backend.app import create_app from backend.db.base import Base -from backend.db.session import get_db +from backend.db.session import get_db, get_auth_db # Test database URL - use environment variable or default TEST_DATABASE_URL = os.getenv( @@ -24,6 +24,82 @@ TEST_DATABASE_URL = os.getenv( "postgresql+psycopg2://postgres:postgres@localhost:5432/punimtag_test" ) +# Test auth database URL - separate DB, mirrors the Prisma `schema-auth.prisma` +# tables (viewer-frontend owns that schema; we recreate it here via raw DDL +# since the backend only ever talks to it through raw SQL, never SQLAlchemy). +TEST_AUTH_DATABASE_URL = os.getenv( + "DATABASE_URL_AUTH", + "postgresql+psycopg2://postgres:postgres@localhost:5432/punimtag_auth_test", +) + +AUTH_DB_DDL = """ +CREATE TABLE IF NOT EXISTS users ( + id SERIAL PRIMARY KEY, + email TEXT UNIQUE NOT NULL, + name TEXT NOT NULL, + password_hash TEXT NOT NULL, + is_admin BOOLEAN NOT NULL DEFAULT FALSE, + has_write_access BOOLEAN NOT NULL DEFAULT FALSE, + email_verified BOOLEAN NOT NULL DEFAULT FALSE, + is_active BOOLEAN NOT NULL DEFAULT TRUE, + created_at TIMESTAMP NOT NULL DEFAULT NOW(), + updated_at TIMESTAMP NOT NULL DEFAULT NOW() +); + +CREATE TABLE IF NOT EXISTS pending_identifications ( + id SERIAL PRIMARY KEY, + face_id INTEGER NOT NULL, + user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, + first_name TEXT NOT NULL, + last_name TEXT NOT NULL, + middle_name TEXT, + maiden_name TEXT, + date_of_birth DATE, + status TEXT NOT NULL DEFAULT 'pending', + created_at TIMESTAMP NOT NULL DEFAULT NOW(), + updated_at TIMESTAMP NOT NULL DEFAULT NOW() +); + +CREATE TABLE IF NOT EXISTS pending_photos ( + id SERIAL PRIMARY KEY, + user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, + filename TEXT NOT NULL, + original_filename TEXT NOT NULL, + file_path TEXT NOT NULL, + file_size INTEGER NOT NULL, + mime_type TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending', + submitted_at TIMESTAMP NOT NULL DEFAULT NOW(), + reviewed_at TIMESTAMP, + reviewed_by INTEGER, + rejection_reason TEXT +); + +CREATE TABLE IF NOT EXISTS pending_linkages ( + id SERIAL PRIMARY KEY, + photo_id INTEGER NOT NULL, + tag_id INTEGER, + tag_name TEXT, + user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, + status TEXT NOT NULL DEFAULT 'pending', + notes TEXT, + created_at TIMESTAMP NOT NULL DEFAULT NOW(), + updated_at TIMESTAMP NOT NULL DEFAULT NOW() +); + +CREATE TABLE IF NOT EXISTS inappropriate_photo_reports ( + id SERIAL PRIMARY KEY, + photo_id INTEGER NOT NULL, + user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, + status TEXT NOT NULL DEFAULT 'pending', + reported_at TIMESTAMP NOT NULL DEFAULT NOW(), + reviewed_at TIMESTAMP, + reviewed_by INTEGER, + review_notes TEXT, + report_comment TEXT +); +""" + @pytest.fixture(scope="session") def test_db_engine(): @@ -58,15 +134,60 @@ def test_db_session(test_db_engine) -> Generator[Session, None, None]: connection.close() +@pytest.fixture(scope="session") +def test_auth_db_engine(): + """Create the auth-database engine and (re)create its tables via raw DDL. + + The auth DB schema is owned by Prisma (`viewer-frontend/prisma/schema-auth.prisma`); + the backend only ever touches it with raw SQL, so we mirror the same tables + here instead of pulling in Prisma for tests. + """ + engine = create_engine(TEST_AUTH_DATABASE_URL, future=True) + + with engine.begin() as conn: + for statement in AUTH_DB_DDL.strip().split(";\n\n"): + if statement.strip(): + conn.exec_driver_sql(statement) + + yield engine + + with engine.begin() as conn: + conn.exec_driver_sql( + "DROP TABLE IF EXISTS inappropriate_photo_reports, pending_linkages, " + "pending_photos, pending_identifications, users CASCADE" + ) + engine.dispose() + + @pytest.fixture(scope="function") -def test_client(test_db_session: Session) -> Generator[TestClient, None, None]: - """Create a test client with test database dependency override.""" +def test_auth_db_session(test_auth_db_engine) -> Generator[Session, None, None]: + """Auth-database session with transaction rollback for test isolation.""" + connection = test_auth_db_engine.connect() + transaction = connection.begin() + session = sessionmaker(bind=connection, autoflush=False, autocommit=False)() + + yield session + + session.close() + transaction.rollback() + connection.close() + + +@pytest.fixture(scope="function") +def test_client( + test_db_session: Session, test_auth_db_session: Session +) -> Generator[TestClient, None, None]: + """Create a test client with test database dependency overrides.""" app = create_app() def override_get_db() -> Generator[Session, None, None]: yield test_db_session + + def override_get_auth_db() -> Generator[Session, None, None]: + yield test_auth_db_session app.dependency_overrides[get_db] = override_get_db + app.dependency_overrides[get_auth_db] = override_get_auth_db with TestClient(app) as client: yield client @@ -75,6 +196,36 @@ def test_client(test_db_session: Session) -> Generator[TestClient, None, None]: app.dependency_overrides.clear() +@pytest.fixture +def auth_db_user(test_auth_db_session: Session): + """Create a row in the auth `users` table (NextAuth/viewer-frontend account). + + Returns the new row's id. Callers that need multiple users should call + this fixture function pattern manually via `_create_auth_db_user`. + """ + return _create_auth_db_user(test_auth_db_session, email="reporter@example.com") + + +def _create_auth_db_user( + session: Session, email: str = "reporter@example.com", name: str = "Reporter" +) -> int: + """Insert a row into the auth `users` table and return its id.""" + from sqlalchemy import text as sa_text + + result = session.execute( + sa_text( + """ + INSERT INTO users (email, name, password_hash) + VALUES (:email, :name, 'not-a-real-hash') + RETURNING id + """ + ), + {"email": email, "name": name}, + ) + session.commit() + return result.scalar_one() + + @pytest.fixture def admin_user(test_db_session: Session): """Create an admin user for testing.""" diff --git a/tests/test_api_click_log.py b/tests/test_api_click_log.py new file mode 100644 index 0000000..ccad443 --- /dev/null +++ b/tests/test_api_click_log.py @@ -0,0 +1,49 @@ +"""Tests for the admin click-logging endpoint (/api/v1/log/click).""" + +from __future__ import annotations + + +class TestLogClickEvent: + def test_log_click_success(self, test_client, admin_user, auth_headers): + """Verify a well-formed click event is accepted and logged.""" + response = test_client.post( + "/api/v1/log/click", + json={ + "page": "/identify", + "element_type": "button", + "element_id": "save-btn", + "element_text": "Save", + "context": {"photo_id": 123}, + }, + headers=auth_headers, + ) + assert response.status_code == 200 + body = response.json() + assert body["status"] == "ok" + + def test_log_click_without_optional_fields(self, test_client, admin_user, auth_headers): + """Verify optional fields (element_id, element_text, context) can be omitted.""" + response = test_client.post( + "/api/v1/log/click", + json={"page": "/search", "element_type": "link"}, + headers=auth_headers, + ) + assert response.status_code == 200 + assert response.json()["status"] == "ok" + + def test_log_click_missing_required_field(self, test_client, admin_user, auth_headers): + """Verify 422 when a required field (element_type) is missing.""" + response = test_client.post( + "/api/v1/log/click", + json={"page": "/identify"}, + headers=auth_headers, + ) + assert response.status_code == 422 + + def test_log_click_unauthenticated(self, test_client): + """Verify 401 without an Authorization header.""" + response = test_client.post( + "/api/v1/log/click", + json={"page": "/identify", "element_type": "button"}, + ) + assert response.status_code == 401 diff --git a/tests/test_api_metrics.py b/tests/test_api_metrics.py new file mode 100644 index 0000000..bd06955 --- /dev/null +++ b/tests/test_api_metrics.py @@ -0,0 +1,18 @@ +"""Tests for the /metrics endpoint.""" + +from __future__ import annotations + + +class TestMetricsEndpoint: + def test_metrics_endpoint_success(self, test_client): + """Verify /metrics returns 200 with the placeholder status payload.""" + response = test_client.get("/metrics") + assert response.status_code == 200 + body = response.json() + assert body["status"] == "ok" + assert "message" in body + + def test_metrics_endpoint_no_auth_required(self, test_client): + """Metrics endpoint should be reachable without an Authorization header.""" + response = test_client.get("/metrics") + assert response.status_code != 401 diff --git a/tests/test_api_pending_identifications.py b/tests/test_api_pending_identifications.py new file mode 100644 index 0000000..df10043 --- /dev/null +++ b/tests/test_api_pending_identifications.py @@ -0,0 +1,191 @@ +"""Tests for the pending-identifications approval workflow (/api/v1/pending-identifications). + +These endpoints read/write the separate auth database (raw SQL against tables +owned by `viewer-frontend/prisma/schema-auth.prisma`), so setup here inserts +rows directly via `test_auth_db_session` rather than going through the API. +""" + +from __future__ import annotations + +from datetime import date + +import pytest +from sqlalchemy import text + +from tests.conftest import _create_auth_db_user + + +def _insert_pending_identification( + session, + face_id: int, + user_id: int, + first_name: str = "Jane", + last_name: str = "Doe", + status: str = "pending", +) -> int: + result = session.execute( + text( + """ + INSERT INTO pending_identifications + (face_id, user_id, first_name, last_name, status) + VALUES (:face_id, :user_id, :first_name, :last_name, :status) + RETURNING id + """ + ), + { + "face_id": face_id, + "user_id": user_id, + "first_name": first_name, + "last_name": last_name, + "status": status, + }, + ) + session.commit() + return result.scalar_one() + + +@pytest.fixture +def auth_reporter(test_auth_db_session): + return _create_auth_db_user(test_auth_db_session, email="viewer@example.com") + + +class TestListPendingIdentifications: + def test_list_default_excludes_denied( + self, test_client, admin_user, auth_headers, test_auth_db_session, test_face, auth_reporter + ): + _insert_pending_identification(test_auth_db_session, test_face.id, auth_reporter, status="pending") + _insert_pending_identification( + test_auth_db_session, test_face.id, auth_reporter, first_name="Denied", status="denied" + ) + + response = test_client.get("/api/v1/pending-identifications", headers=auth_headers) + assert response.status_code == 200 + body = response.json() + names = {item["first_name"] for item in body["items"]} + assert "Jane" in names + assert "Denied" not in names + + def test_list_include_denied( + self, test_client, admin_user, auth_headers, test_auth_db_session, test_face, auth_reporter + ): + _insert_pending_identification( + test_auth_db_session, test_face.id, auth_reporter, first_name="Denied", status="denied" + ) + + response = test_client.get( + "/api/v1/pending-identifications", + params={"include_denied": True}, + headers=auth_headers, + ) + assert response.status_code == 200 + names = {item["first_name"] for item in response.json()["items"]} + assert "Denied" in names + + def test_list_unauthenticated(self, test_client): + response = test_client.get("/api/v1/pending-identifications") + assert response.status_code == 401 + + +class TestApproveDenyPendingIdentifications: + def test_approve_creates_person_and_identifies_face( + self, test_client, admin_user, auth_headers, test_auth_db_session, test_face, auth_reporter + ): + pending_id = _insert_pending_identification( + test_auth_db_session, test_face.id, auth_reporter, first_name="Approved", last_name="Person" + ) + + response = test_client.post( + "/api/v1/pending-identifications/approve-deny", + json={"decisions": [{"id": pending_id, "decision": "approve"}]}, + headers=auth_headers, + ) + assert response.status_code == 200 + body = response.json() + assert body["approved"] == 1 + assert body["denied"] == 0 + assert body["errors"] == [] + + def test_deny_updates_status_only( + self, test_client, admin_user, auth_headers, test_auth_db_session, test_face, auth_reporter + ): + pending_id = _insert_pending_identification(test_auth_db_session, test_face.id, auth_reporter) + + response = test_client.post( + "/api/v1/pending-identifications/approve-deny", + json={"decisions": [{"id": pending_id, "decision": "deny"}]}, + headers=auth_headers, + ) + assert response.status_code == 200 + body = response.json() + assert body["denied"] == 1 + assert body["approved"] == 0 + + def test_approve_deny_not_found(self, test_client, admin_user, auth_headers): + response = test_client.post( + "/api/v1/pending-identifications/approve-deny", + json={"decisions": [{"id": 999999, "decision": "approve"}]}, + headers=auth_headers, + ) + assert response.status_code == 200 + body = response.json() + assert body["errors"] + assert body["approved"] == 0 + + def test_approve_deny_unauthenticated(self, test_client): + response = test_client.post( + "/api/v1/pending-identifications/approve-deny", + json={"decisions": [{"id": 1, "decision": "approve"}]}, + ) + assert response.status_code == 401 + + +class TestIdentificationReport: + def test_report_counts_identified_faces( + self, test_client, admin_user, auth_headers, test_db_session, identified_face + ): + identified_face.identified_by_user_id = admin_user.id + test_db_session.add(identified_face) + test_db_session.commit() + + response = test_client.get("/api/v1/pending-identifications/report", headers=auth_headers) + assert response.status_code == 200 + body = response.json() + assert body["total_faces"] >= 1 + assert any(item["user_id"] == admin_user.id for item in body["items"]) + + def test_report_invalid_date_format(self, test_client, admin_user, auth_headers): + response = test_client.get( + "/api/v1/pending-identifications/report", + params={"date_from": "not-a-date"}, + headers=auth_headers, + ) + assert response.status_code == 400 + + +class TestClearDeniedIdentifications: + def test_clear_denied_deletes_denied_records( + self, test_client, admin_user, auth_headers, test_auth_db_session, test_face, auth_reporter + ): + _insert_pending_identification( + test_auth_db_session, test_face.id, auth_reporter, status="denied" + ) + + response = test_client.post( + "/api/v1/pending-identifications/clear-denied", headers=auth_headers + ) + assert response.status_code == 200 + body = response.json() + assert body["deleted_records"] == 1 + + def test_clear_denied_no_records(self, test_client, admin_user, auth_headers): + response = test_client.post( + "/api/v1/pending-identifications/clear-denied", headers=auth_headers + ) + assert response.status_code == 200 + assert response.json()["deleted_records"] == 0 + + def test_clear_denied_non_admin(self, test_client, regular_auth_headers, admin_user): + response = test_client.post( + "/api/v1/pending-identifications/clear-denied", headers=regular_auth_headers + ) + assert response.status_code == 403 diff --git a/tests/test_api_pending_linkages.py b/tests/test_api_pending_linkages.py new file mode 100644 index 0000000..7ae3b7a --- /dev/null +++ b/tests/test_api_pending_linkages.py @@ -0,0 +1,156 @@ +"""Tests for the pending tag-linkage review workflow (/api/v1/pending-linkages).""" + +from __future__ import annotations + +import pytest +from sqlalchemy import text + +from tests.conftest import _create_auth_db_user + + +def _insert_pending_linkage( + session, + photo_id: int, + user_id: int, + tag_name: str = "vacation", + tag_id: int | None = None, + status: str = "pending", +) -> int: + result = session.execute( + text( + """ + INSERT INTO pending_linkages (photo_id, tag_id, tag_name, user_id, status) + VALUES (:photo_id, :tag_id, :tag_name, :user_id, :status) + RETURNING id + """ + ), + { + "photo_id": photo_id, + "tag_id": tag_id, + "tag_name": tag_name, + "user_id": user_id, + "status": status, + }, + ) + session.commit() + return result.scalar_one() + + +@pytest.fixture +def auth_reporter(test_auth_db_session): + return _create_auth_db_user(test_auth_db_session, email="tagger@example.com") + + +class TestListPendingLinkages: + def test_list_success( + self, test_client, admin_user, auth_headers, test_auth_db_session, test_photo, auth_reporter + ): + _insert_pending_linkage(test_auth_db_session, test_photo.id, auth_reporter) + + response = test_client.get("/api/v1/pending-linkages", headers=auth_headers) + assert response.status_code == 200 + body = response.json() + assert body["total"] >= 1 + assert body["items"][0]["proposed_tag_name"] == "vacation" + assert body["items"][0]["photo_filename"] == test_photo.filename + + def test_list_filter_by_status( + self, test_client, admin_user, auth_headers, test_auth_db_session, test_photo, auth_reporter + ): + _insert_pending_linkage(test_auth_db_session, test_photo.id, auth_reporter, status="denied") + + response = test_client.get( + "/api/v1/pending-linkages", + params={"status_filter": "pending"}, + headers=auth_headers, + ) + assert response.status_code == 200 + assert response.json()["total"] == 0 + + def test_list_invalid_status_filter(self, test_client, admin_user, auth_headers): + response = test_client.get( + "/api/v1/pending-linkages", + params={"status_filter": "not_a_status"}, + headers=auth_headers, + ) + assert response.status_code == 400 + + def test_list_unauthenticated(self, test_client): + response = test_client.get("/api/v1/pending-linkages") + assert response.status_code == 401 + + +class TestReviewPendingLinkages: + def test_approve_creates_tag_and_linkage( + self, test_client, admin_user, auth_headers, test_auth_db_session, test_photo, auth_reporter + ): + linkage_id = _insert_pending_linkage( + test_auth_db_session, test_photo.id, auth_reporter, tag_name="new-tag" + ) + + response = test_client.post( + "/api/v1/pending-linkages/review", + json={"decisions": [{"id": linkage_id, "decision": "approve"}]}, + headers=auth_headers, + ) + assert response.status_code == 200 + body = response.json() + assert body["approved"] == 1 + assert body["tags_created"] == 1 + assert body["linkages_created"] == 1 + + tags = test_client.get(f"/api/v1/tags/photos/{test_photo.id}") + assert any(t["tag_name"] == "new-tag" for t in tags.json()["tags"]) + + def test_deny_only_updates_status( + self, test_client, admin_user, auth_headers, test_auth_db_session, test_photo, auth_reporter + ): + linkage_id = _insert_pending_linkage(test_auth_db_session, test_photo.id, auth_reporter) + + response = test_client.post( + "/api/v1/pending-linkages/review", + json={"decisions": [{"id": linkage_id, "decision": "deny"}]}, + headers=auth_headers, + ) + assert response.status_code == 200 + body = response.json() + assert body["denied"] == 1 + assert body["tags_created"] == 0 + + def test_review_not_found(self, test_client, admin_user, auth_headers): + response = test_client.post( + "/api/v1/pending-linkages/review", + json={"decisions": [{"id": 999999, "decision": "approve"}]}, + headers=auth_headers, + ) + assert response.status_code == 200 + assert response.json()["errors"] + + def test_review_unauthenticated(self, test_client): + response = test_client.post( + "/api/v1/pending-linkages/review", + json={"decisions": [{"id": 1, "decision": "approve"}]}, + ) + assert response.status_code == 401 + + +class TestCleanupPendingLinkages: + def test_cleanup_deletes_resolved_records( + self, test_client, admin_user, auth_headers, test_auth_db_session, test_photo, auth_reporter + ): + _insert_pending_linkage(test_auth_db_session, test_photo.id, auth_reporter, status="approved") + _insert_pending_linkage(test_auth_db_session, test_photo.id, auth_reporter, status="pending") + + response = test_client.post("/api/v1/pending-linkages/cleanup", headers=auth_headers) + assert response.status_code == 200 + assert response.json()["deleted_records"] == 1 + + remaining = test_client.get("/api/v1/pending-linkages", headers=auth_headers) + assert remaining.json()["total"] == 1 + + def test_cleanup_no_records_returns_warning(self, test_client, admin_user, auth_headers): + response = test_client.post("/api/v1/pending-linkages/cleanup", headers=auth_headers) + assert response.status_code == 200 + body = response.json() + assert body["deleted_records"] == 0 + assert body["warnings"] diff --git a/tests/test_api_reported_photos.py b/tests/test_api_reported_photos.py new file mode 100644 index 0000000..4267baa --- /dev/null +++ b/tests/test_api_reported_photos.py @@ -0,0 +1,157 @@ +"""Tests for the inappropriate-photo-report review workflow (/api/v1/reported-photos).""" + +from __future__ import annotations + +import pytest +from sqlalchemy import text + +from tests.conftest import _create_auth_db_user + + +def _insert_report( + session, + photo_id: int, + user_id: int, + report_comment: str = "inappropriate", + status: str = "pending", +) -> int: + result = session.execute( + text( + """ + INSERT INTO inappropriate_photo_reports (photo_id, user_id, report_comment, status) + VALUES (:photo_id, :user_id, :report_comment, :status) + RETURNING id + """ + ), + { + "photo_id": photo_id, + "user_id": user_id, + "report_comment": report_comment, + "status": status, + }, + ) + session.commit() + return result.scalar_one() + + +@pytest.fixture +def auth_reporter(test_auth_db_session): + return _create_auth_db_user(test_auth_db_session, email="reporter2@example.com") + + +class TestListReportedPhotos: + def test_list_success( + self, test_client, admin_user, auth_headers, test_auth_db_session, test_photo, auth_reporter + ): + _insert_report(test_auth_db_session, test_photo.id, auth_reporter) + + response = test_client.get("/api/v1/reported-photos", headers=auth_headers) + assert response.status_code == 200 + body = response.json() + assert body["total"] >= 1 + assert body["items"][0]["photo_filename"] == test_photo.filename + + def test_list_filter_by_status( + self, test_client, admin_user, auth_headers, test_auth_db_session, test_photo, auth_reporter + ): + _insert_report(test_auth_db_session, test_photo.id, auth_reporter, status="dismissed") + + response = test_client.get( + "/api/v1/reported-photos", params={"status_filter": "pending"}, headers=auth_headers + ) + assert response.status_code == 200 + assert response.json()["total"] == 0 + + def test_list_unauthenticated(self, test_client): + response = test_client.get("/api/v1/reported-photos") + assert response.status_code == 401 + + +class TestReviewReportedPhotos: + def test_keep_marks_reviewed_and_preserves_photo( + self, + test_client, + admin_user, + auth_headers, + test_auth_db_session, + test_db_session, + test_photo, + auth_reporter, + ): + report_id = _insert_report(test_auth_db_session, test_photo.id, auth_reporter) + + response = test_client.post( + "/api/v1/reported-photos/review", + json={"decisions": [{"id": report_id, "decision": "keep"}]}, + headers=auth_headers, + ) + assert response.status_code == 200 + body = response.json() + assert body["kept"] == 1 + assert body["removed"] == 0 + + # Photo must still exist in the main database. + still_there = test_client.get(f"/api/v1/photos/{test_photo.id}", headers=auth_headers) + assert still_there.status_code == 200 + + def test_remove_deletes_photo_from_main_db( + self, test_client, admin_user, auth_headers, test_auth_db_session, test_photo, auth_reporter + ): + report_id = _insert_report(test_auth_db_session, test_photo.id, auth_reporter) + + response = test_client.post( + "/api/v1/reported-photos/review", + json={"decisions": [{"id": report_id, "decision": "remove"}]}, + headers=auth_headers, + ) + assert response.status_code == 200 + body = response.json() + assert body["removed"] == 1 + + gone = test_client.get(f"/api/v1/photos/{test_photo.id}", headers=auth_headers) + assert gone.status_code == 404 + + def test_review_not_found(self, test_client, admin_user, auth_headers): + response = test_client.post( + "/api/v1/reported-photos/review", + json={"decisions": [{"id": 999999, "decision": "keep"}]}, + headers=auth_headers, + ) + assert response.status_code == 200 + assert response.json()["errors"] + + def test_review_unauthenticated(self, test_client): + response = test_client.post( + "/api/v1/reported-photos/review", + json={"decisions": [{"id": 1, "decision": "keep"}]}, + ) + assert response.status_code == 401 + + +class TestCleanupReportedPhotos: + def test_cleanup_removes_dismissed_when_filter_is_remove( + self, test_client, admin_user, auth_headers, test_auth_db_session, test_photo, auth_reporter + ): + _insert_report(test_auth_db_session, test_photo.id, auth_reporter, status="dismissed") + + response = test_client.post( + "/api/v1/reported-photos/cleanup", + params={"status_filter": "remove"}, + headers=auth_headers, + ) + assert response.status_code == 200 + assert response.json()["deleted_records"] == 1 + + def test_cleanup_invalid_status_filter(self, test_client, admin_user, auth_headers): + response = test_client.post( + "/api/v1/reported-photos/cleanup", + params={"status_filter": "not_a_valid_filter"}, + headers=auth_headers, + ) + assert response.status_code == 400 + + def test_cleanup_non_admin(self, test_client, regular_auth_headers, admin_user): + response = test_client.post( + "/api/v1/reported-photos/cleanup", headers=regular_auth_headers + ) + assert response.status_code == 403 diff --git a/tests/test_api_role_permissions.py b/tests/test_api_role_permissions.py new file mode 100644 index 0000000..6758d3d --- /dev/null +++ b/tests/test_api_role_permissions.py @@ -0,0 +1,79 @@ +"""Tests for role/feature permission management (/api/v1/role-permissions).""" + +from __future__ import annotations + + +class TestListRolePermissions: + def test_list_role_permissions_success(self, test_client, admin_user, auth_headers): + """Verify the full feature/permission matrix is returned for admins.""" + response = test_client.get("/api/v1/role-permissions", headers=auth_headers) + assert response.status_code == 200 + body = response.json() + assert "features" in body + assert "permissions" in body + assert len(body["features"]) > 0 + assert all("key" in f and "label" in f for f in body["features"]) + + def test_list_role_permissions_non_admin( + self, test_client, regular_auth_headers, admin_user + ): + """Verify 403 for non-admin users.""" + response = test_client.get("/api/v1/role-permissions", headers=regular_auth_headers) + assert response.status_code == 403 + + def test_list_role_permissions_unauthenticated(self, test_client): + """Verify 401 without a token.""" + response = test_client.get("/api/v1/role-permissions") + assert response.status_code == 401 + + +class TestUpdateRolePermissions: + def test_update_role_permissions_success(self, test_client, admin_user, auth_headers): + """Verify a valid permission update round-trips through the API.""" + response = test_client.put( + "/api/v1/role-permissions", + json={"permissions": {"viewer": {"user_identified": False}}}, + headers=auth_headers, + ) + assert response.status_code == 200 + body = response.json() + assert body["permissions"]["viewer"]["user_identified"] is False + + # Persisted, not just echoed back + follow_up = test_client.get("/api/v1/role-permissions", headers=auth_headers) + assert follow_up.json()["permissions"]["viewer"]["user_identified"] is False + + def test_update_role_permissions_invalid_role(self, test_client, admin_user, auth_headers): + """Verify an unknown role key is rejected. + + The request schema types `permissions` keys as the `UserRole` enum, so + Pydantic itself rejects unknown roles with 422 before the endpoint's + own `Invalid role(s)` 400 check ever runs. + """ + response = test_client.put( + "/api/v1/role-permissions", + json={"permissions": {"not_a_real_role": {"user_identified": True}}}, + headers=auth_headers, + ) + assert response.status_code == 422 + + def test_update_role_permissions_invalid_feature(self, test_client, admin_user, auth_headers): + """Verify 400 when an unknown feature key is submitted.""" + response = test_client.put( + "/api/v1/role-permissions", + json={"permissions": {"viewer": {"not_a_real_feature": True}}}, + headers=auth_headers, + ) + assert response.status_code == 400 + assert "Invalid feature" in response.json()["detail"] + + def test_update_role_permissions_non_admin( + self, test_client, regular_auth_headers, admin_user + ): + """Verify 403 for non-admin users.""" + response = test_client.put( + "/api/v1/role-permissions", + json={"permissions": {"viewer": {"user_identified": True}}}, + headers=regular_auth_headers, + ) + assert response.status_code == 403 diff --git a/tests/test_api_videos.py b/tests/test_api_videos.py new file mode 100644 index 0000000..757057a --- /dev/null +++ b/tests/test_api_videos.py @@ -0,0 +1,171 @@ +"""Tests for video person-identification endpoints (/api/v1/videos).""" + +from __future__ import annotations + +from datetime import date + +import pytest + + +@pytest.fixture +def test_video(test_db_session): + """Create a test video (a Photo row with media_type='video').""" + from backend.db.models import Photo + + video = Photo( + path="/test/path/video1.mp4", + filename="video1.mp4", + date_taken=date(2024, 1, 15), + processed=True, + media_type="video", + ) + test_db_session.add(video) + test_db_session.commit() + test_db_session.refresh(video) + return video + + +class TestListVideos: + def test_list_videos_success(self, test_client, admin_user, auth_headers, test_video): + response = test_client.get("/api/v1/videos", headers=auth_headers) + assert response.status_code == 200 + body = response.json() + assert body["total"] >= 1 + assert any(item["id"] == test_video.id for item in body["items"]) + + def test_list_videos_unauthenticated(self, test_client): + response = test_client.get("/api/v1/videos") + assert response.status_code == 401 + + def test_list_videos_invalid_sort_by(self, test_client, admin_user, auth_headers): + response = test_client.get( + "/api/v1/videos", params={"sort_by": "not_a_field"}, headers=auth_headers + ) + assert response.status_code == 400 + + def test_list_videos_invalid_date_format(self, test_client, admin_user, auth_headers): + response = test_client.get( + "/api/v1/videos", params={"date_from": "not-a-date"}, headers=auth_headers + ) + assert response.status_code == 400 + + +class TestGetVideoPeople: + def test_get_video_people_no_people(self, test_client, test_video): + """This endpoint has no auth dependency.""" + response = test_client.get(f"/api/v1/videos/{test_video.id}/people") + assert response.status_code == 200 + body = response.json() + assert body["video_id"] == test_video.id + assert body["people"] == [] + + def test_get_video_people_not_found(self, test_client): + response = test_client.get("/api/v1/videos/999999/people") + assert response.status_code == 404 + + def test_get_video_people_rejects_non_video_photo(self, test_client, test_photo): + """A regular photo (media_type='image') should not be treated as a video.""" + response = test_client.get(f"/api/v1/videos/{test_photo.id}/people") + assert response.status_code == 404 + + +class TestIdentifyPersonInVideo: + def test_identify_creates_new_person( + self, test_client, admin_user, auth_headers, test_video + ): + response = test_client.post( + f"/api/v1/videos/{test_video.id}/identify", + json={"first_name": "Jane", "last_name": "Smith"}, + headers=auth_headers, + ) + assert response.status_code == 200 + body = response.json() + assert body["video_id"] == test_video.id + assert body["created_person"] is True + + people = test_client.get(f"/api/v1/videos/{test_video.id}/people") + assert people.json()["people"][0]["first_name"] == "Jane" + + def test_identify_with_existing_person( + self, test_client, admin_user, auth_headers, test_video, test_person + ): + response = test_client.post( + f"/api/v1/videos/{test_video.id}/identify", + json={"person_id": test_person.id}, + headers=auth_headers, + ) + assert response.status_code == 200 + body = response.json() + assert body["created_person"] is False + assert body["person_id"] == test_person.id + + def test_identify_unauthenticated(self, test_client, test_video): + response = test_client.post( + f"/api/v1/videos/{test_video.id}/identify", + json={"first_name": "Jane", "last_name": "Smith"}, + ) + assert response.status_code == 401 + + +class TestRemovePersonFromVideo: + def test_remove_person_success( + self, test_client, admin_user, auth_headers, test_video, test_person + ): + test_client.post( + f"/api/v1/videos/{test_video.id}/identify", + json={"person_id": test_person.id}, + headers=auth_headers, + ) + + response = test_client.delete( + f"/api/v1/videos/{test_video.id}/people/{test_person.id}", + headers=auth_headers, + ) + assert response.status_code == 200 + assert response.json()["removed"] is True + + def test_remove_person_not_linked( + self, test_client, admin_user, auth_headers, test_video, test_person + ): + response = test_client.delete( + f"/api/v1/videos/{test_video.id}/people/{test_person.id}", + headers=auth_headers, + ) + assert response.status_code == 200 + assert response.json()["removed"] is False + + def test_remove_person_unauthenticated(self, test_client, test_video, test_person): + response = test_client.delete( + f"/api/v1/videos/{test_video.id}/people/{test_person.id}", + ) + assert response.status_code == 401 + + +class TestVideoFileEndpoints: + def test_get_video_thumbnail_not_found(self, test_client): + response = test_client.get("/api/v1/videos/999999/thumbnail") + assert response.status_code == 404 + + def test_get_video_file_not_found(self, test_client): + response = test_client.get("/api/v1/videos/999999/video") + assert response.status_code == 404 + + def test_get_video_file_missing_on_disk(self, test_client, test_video): + """Video row exists but the file on disk doesn't (path is fake).""" + response = test_client.get(f"/api/v1/videos/{test_video.id}/video") + assert response.status_code == 404 + + +class TestWebPlayback: + def test_prepare_web_playback_redis_unavailable( + self, test_client, test_video + ): + """No Redis running in the test environment -> 503, not a crash.""" + response = test_client.post( + f"/api/v1/videos/{test_video.id}/web-playback/prepare" + ) + assert response.status_code == 503 + + def test_web_playback_status_not_found(self, test_client): + response = test_client.get("/api/v1/videos/999999/web-playback/status") + assert response.status_code == 404 -- 2.49.1