ilia/punimtag
1
1
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request 'test: cover 7 previously-untested backend API route modules' (#61) from test/backend-pytest-coverage-gaps into dev

Browse Source
This commit is contained in:
ilia 2026-07-14 17:17:20 -05:00
commit fc8b47863d
8 changed files with 975 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

157
tests/conftest.py
View File

@ -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."""

49
tests/test_api_click_log.py Normal file
View File

@ -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

18
tests/test_api_metrics.py Normal file
View File

@ -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

191
tests/test_api_pending_identifications.py Normal file
View File

@ -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

156
tests/test_api_pending_linkages.py Normal file
View File

@ -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"]

157
tests/test_api_reported_photos.py Normal file
View File

@ -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

79
tests/test_api_role_permissions.py Normal file
View File

@ -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

171
tests/test_api_videos.py Normal file
View File

@ -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