diff --git a/backend/tests/test_admin_overview.py b/backend/tests/test_admin_overview.py new file mode 100644 index 0000000..51eb39d --- /dev/null +++ b/backend/tests/test_admin_overview.py @@ -0,0 +1,260 @@ +""" +Admin overview endpoint tests — ADMIN-11. + +Tests cover: + - GET /api/admin/overview (requires admin, aggregate stats + last 10 audit entries) + - Security invariants: no password_hash, credentials_enc, extracted_text, etc. + +Fixture pattern copied verbatim from test_admin_api.py (make_admin_user, admin_client). +""" +from __future__ import annotations + +import uuid +from datetime import datetime, timezone + +import pytest +import pytest_asyncio +from httpx import ASGITransport, AsyncClient +from sqlalchemy.ext.asyncio import AsyncSession + +from db.models import AuditLog, Document, Quota, User +from deps.auth import get_current_admin +from deps.db import get_db +from services.auth import hash_password +from tests.test_auth_api import FakeRedis + + +# ── Helpers ─────────────────────────────────────────────────────────────────── + +async def make_admin_user(session: AsyncSession) -> User: + """Insert an admin User + Quota row and return the ORM object.""" + user = User( + id=uuid.uuid4(), + handle=f"admin_{uuid.uuid4().hex[:6]}", + email=f"admin_{uuid.uuid4().hex[:6]}@example.com", + password_hash=hash_password("AdminPass1!Secret"), + role="admin", + is_active=True, + totp_enabled=False, + password_must_change=False, + ) + session.add(user) + quota = Quota(user_id=user.id, limit_bytes=104857600, used_bytes=0) + session.add(quota) + await session.flush() + return user + + +async def make_regular_user(session: AsyncSession) -> User: + """Insert a regular User + Quota row and return the ORM object.""" + user = User( + id=uuid.uuid4(), + handle=f"user_{uuid.uuid4().hex[:6]}", + email=f"user_{uuid.uuid4().hex[:6]}@example.com", + password_hash=hash_password("UserPass1!Secret"), + role="user", + is_active=True, + totp_enabled=False, + password_must_change=False, + ) + session.add(user) + quota = Quota(user_id=user.id, limit_bytes=104857600, used_bytes=0) + session.add(quota) + await session.flush() + return user + + +@pytest_asyncio.fixture +async def admin_client(db_session: AsyncSession): + """Async client with get_current_admin overridden to an admin user. + + Phase 7.2: app.state.redis is set to a FakeRedis() instance so that + the admin deactivation handler can call request.app.state.redis.set(...) + without AttributeError. + """ + from main import app + + admin = await make_admin_user(db_session) + app.dependency_overrides[get_db] = lambda: db_session + app.dependency_overrides[get_current_admin] = lambda: admin + app.state.redis = FakeRedis() + + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c, admin, db_session + + app.dependency_overrides.clear() + app.state.redis = None + + +# ── Tests ───────────────────────────────────────────────────────────────────── + + +@pytest.mark.asyncio +@pytest.mark.xfail(strict=True, reason="ADMIN-11: overview endpoint not yet implemented") +async def test_overview_requires_admin(async_client: AsyncClient): + """Unauthenticated GET /api/admin/overview returns 401 or 403.""" + resp = await async_client.get("/api/admin/overview") + assert resp.status_code in (401, 403) + + +@pytest.mark.asyncio +@pytest.mark.xfail(strict=True, reason="ADMIN-11: overview endpoint not yet implemented") +async def test_overview_non_admin_forbidden(db_session: AsyncSession): + """Regular-user JWT returns 403 from GET /api/admin/overview.""" + from main import app + from deps.auth import get_current_user + + regular = await make_regular_user(db_session) + app.dependency_overrides[get_db] = lambda: db_session + app.dependency_overrides[get_current_user] = lambda: regular + + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + resp = await c.get("/api/admin/overview") + + app.dependency_overrides.clear() + assert resp.status_code == 403 + + +@pytest.mark.asyncio +@pytest.mark.xfail(strict=True, reason="ADMIN-11: overview endpoint not yet implemented") +async def test_overview_returns_expected_keys(admin_client): + """Admin GET returns 200 with all four top-level keys present.""" + c, admin, session = admin_client + resp = await c.get("/api/admin/overview") + assert resp.status_code == 200 + body = resp.json() + for key in ("user_count", "total_storage_bytes", "doc_status", "recent_audit"): + assert key in body, f"missing key '{key}' in overview response" + + +@pytest.mark.asyncio +@pytest.mark.xfail(strict=True, reason="ADMIN-11: overview endpoint not yet implemented") +async def test_overview_user_count_excludes_admins(admin_client): + """user_count reflects only role='user' accounts, not admins.""" + c, admin, session = admin_client + # Seed two regular users (admin already exists from fixture) + await make_regular_user(session) + await make_regular_user(session) + await session.commit() + + resp = await c.get("/api/admin/overview") + assert resp.status_code == 200 + body = resp.json() + assert body["user_count"] == 2, ( + f"expected user_count=2, got {body['user_count']}" + ) + + +@pytest.mark.asyncio +@pytest.mark.xfail(strict=True, reason="ADMIN-11: overview endpoint not yet implemented") +async def test_overview_total_storage_sums_quotas(admin_client): + """total_storage_bytes sums used_bytes across all Quota rows.""" + c, admin, session = admin_client + # Seed two regular users with known quota usage + user_a = await make_regular_user(session) + user_b = await make_regular_user(session) + + # Update quota used_bytes directly + quota_a = await session.get(Quota, user_a.id) + quota_b = await session.get(Quota, user_b.id) + quota_a.used_bytes = 1024 + quota_b.used_bytes = 2048 + session.add(quota_a) + session.add(quota_b) + await session.commit() + + resp = await c.get("/api/admin/overview") + assert resp.status_code == 200 + body = resp.json() + # Admin quota (0) + user_a (1024) + user_b (2048) = 3072 + assert body["total_storage_bytes"] == 3072, ( + f"expected total_storage_bytes=3072, got {body['total_storage_bytes']}" + ) + + +@pytest.mark.asyncio +@pytest.mark.xfail(strict=True, reason="ADMIN-11: overview endpoint not yet implemented") +async def test_overview_doc_status_groups_by_status(admin_client): + """doc_status maps each status to its count.""" + c, admin, session = admin_client + user = await make_regular_user(session) + await session.commit() + + # Seed documents with statuses: ready x2, processing x1, failed x1 + for status_val in ("ready", "ready", "processing", "failed"): + doc = Document( + id=uuid.uuid4(), + user_id=user.id, + filename=f"test_{uuid.uuid4().hex[:4]}.pdf", + object_key=f"{user.id}/{uuid.uuid4()}.pdf", + content_type="application/pdf", + size_bytes=100, + status=status_val, + ) + session.add(doc) + await session.commit() + + resp = await c.get("/api/admin/overview") + assert resp.status_code == 200 + body = resp.json() + doc_status = body["doc_status"] + assert doc_status.get("ready") == 2, f"expected ready=2, got {doc_status}" + assert doc_status.get("processing") == 1, f"expected processing=1, got {doc_status}" + assert doc_status.get("failed") == 1, f"expected failed=1, got {doc_status}" + + +@pytest.mark.asyncio +@pytest.mark.xfail(strict=True, reason="ADMIN-11: overview endpoint not yet implemented") +async def test_overview_recent_audit_limit_10(admin_client): + """recent_audit returns at most 10 entries ordered newest first.""" + c, admin, session = admin_client + user = await make_regular_user(session) + await session.commit() + + # Seed 15 audit log entries directly via ORM (no helper coupling) + for i in range(15): + entry = AuditLog( + event_type="document.uploaded", + user_id=user.id, + actor_id=user.id, + resource_id=None, + ip_address=None, + metadata_={"seq": i}, + ) + session.add(entry) + await session.commit() + + resp = await c.get("/api/admin/overview") + assert resp.status_code == 200 + body = resp.json() + recent = body["recent_audit"] + assert isinstance(recent, list), "recent_audit must be a list" + assert len(recent) == 10, ( + f"expected len(recent_audit)=10, got {len(recent)}" + ) + # Entries should be ordered newest first (descending created_at) + timestamps = [entry["created_at"] for entry in recent] + assert timestamps == sorted(timestamps, reverse=True), ( + "recent_audit entries must be ordered newest first" + ) + + +@pytest.mark.asyncio +@pytest.mark.xfail(strict=True, reason="ADMIN-11: overview endpoint not yet implemented") +async def test_overview_no_sensitive_fields(admin_client): + """Response body must not contain any sensitive field name.""" + c, admin, session = admin_client + resp = await c.get("/api/admin/overview") + assert resp.status_code == 200 + body_text = resp.text + forbidden = ( + "password_hash", + "credentials_enc", + "extracted_text", + "totp_secret", + "api_key_enc", + ) + for field in forbidden: + assert field not in body_text, ( + f"forbidden field '{field}' found in overview response body" + )