curo1305
41d81c089f
- New backend/api/admin/overview.py: aggregate stats (user_count, total_storage_bytes, doc_status) + last 10 audit entries in one payload - Sub-router carries NO prefix (parent __init__.py carries /api/admin) - Reuses _build_filtered_query_with_handles + _audit_to_dict_with_handles from api.audit — no new serializer or query logic - Registered overview_router on admin parent router in __init__.py - All 8 ADMIN-11 tests pass; existing admin and audit suites stay green
253 lines
8.7 KiB
Python
253 lines
8.7 KiB
Python
"""
|
|
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
|
|
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
|
|
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
|
|
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
|
|
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
|
|
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
|
|
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
|
|
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
|
|
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"
|
|
)
|