test(09-01): add failing xfail tests for ADMIN-11 overview endpoint

- 8 xfail tests covering auth guards, aggregate stats, and sensitive field scan - Tests will become real assertions once overview.py is implemented in Task 2
2026-06-12 15:47:18 +02:00
parent 2fd7591eac
commit 4caeed2d22
1 changed files with 260 additions and 0 deletions
@@ -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"
+        )