feat(6.1-02): promote test_audit.py stubs to real tests (ADMIN-06)
- Replace all 4 @pytest.mark.xfail stubs with real assertions - Add _seed_audit() helper calling write_audit_log() + commit - test_audit_log_viewer: verifies paginated JSON shape and total >= 1 - test_audit_log_no_doc_content: asserts no filename/extracted_text in items - test_audit_log_regular_user_403: asserts 403 for regular users - test_audit_log_export_csv: asserts text/csv content-type and CSV header line - Remove unused 'import os' - Add pytestmark = pytest.mark.asyncio at module level
This commit is contained in:
+124
-22
@@ -1,43 +1,145 @@
|
|||||||
"""
|
"""
|
||||||
Audit log API tests — Wave 0 xfail stubs for Phase 4.
|
Audit log API tests — ADMIN-06.
|
||||||
|
|
||||||
All tests in this file are xfail stubs. They will be implemented in Plan 04-07.
|
|
||||||
The stubs ensure pytest collects them and keeps CI green before implementation
|
|
||||||
code exists.
|
|
||||||
|
|
||||||
Requirement: ADMIN-06 — admin audit log viewer, no doc content, export CSV.
|
Requirement: ADMIN-06 — admin audit log viewer, no doc content, export CSV.
|
||||||
|
|
||||||
|
Tests:
|
||||||
|
- test_audit_log_viewer: paginated JSON viewer returns seeded entry
|
||||||
|
- test_audit_log_no_doc_content: response items never expose filename / extracted_text
|
||||||
|
- test_audit_log_regular_user_403: regular users are blocked with 403
|
||||||
|
- test_audit_log_export_csv: CSV export returns correct headers and CSV structure
|
||||||
"""
|
"""
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
import os
|
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
|
pytestmark = pytest.mark.asyncio
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# Seed helper
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
async def _seed_audit(db_session, user_id) -> None:
|
||||||
|
"""Insert one AuditLog row and commit.
|
||||||
|
|
||||||
|
Imports write_audit_log inside the function body to avoid top-level import
|
||||||
|
ordering issues when conftest patches db models before this module loads.
|
||||||
|
"""
|
||||||
|
from services.audit import write_audit_log
|
||||||
|
|
||||||
|
await write_audit_log(
|
||||||
|
session=db_session,
|
||||||
|
event_type="document.uploaded",
|
||||||
|
user_id=user_id,
|
||||||
|
actor_id=user_id,
|
||||||
|
resource_id=None,
|
||||||
|
ip_address=None,
|
||||||
|
metadata_={"size_bytes": 100},
|
||||||
|
)
|
||||||
|
await db_session.commit()
|
||||||
|
|
||||||
|
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
# ADMIN-06: Audit log viewer
|
# ADMIN-06: Audit log viewer
|
||||||
# ---------------------------------------------------------------------------
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.xfail(strict=False)
|
async def test_audit_log_viewer(async_client, admin_user, db_session):
|
||||||
async def test_audit_log_viewer(async_client, admin_user):
|
"""GET /api/admin/audit-log returns paginated entries with correct shape."""
|
||||||
"""GET /api/admin/audit-log returns paginated entries."""
|
await _seed_audit(db_session, admin_user["user"].id)
|
||||||
pytest.xfail("not implemented yet")
|
|
||||||
|
response = await async_client.get(
|
||||||
|
"/api/admin/audit-log",
|
||||||
|
headers=admin_user["headers"],
|
||||||
|
)
|
||||||
|
|
||||||
|
assert response.status_code == 200
|
||||||
|
body = response.json()
|
||||||
|
|
||||||
|
# Pagination envelope keys
|
||||||
|
for key in ("items", "total", "page", "per_page"):
|
||||||
|
assert key in body, f"missing key '{key}' in response body"
|
||||||
|
|
||||||
|
assert body["total"] >= 1, "expected at least 1 audit entry after seeding"
|
||||||
|
|
||||||
|
items = body["items"]
|
||||||
|
assert isinstance(items, list), "items must be a list"
|
||||||
|
assert len(items) >= 1, "items list must be non-empty"
|
||||||
|
|
||||||
|
first = items[0]
|
||||||
|
for key in ("id", "event_type", "user_id", "created_at"):
|
||||||
|
assert key in first, f"missing key '{key}' in audit item"
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.xfail(strict=False)
|
async def test_audit_log_no_doc_content(async_client, admin_user, db_session):
|
||||||
async def test_audit_log_no_doc_content(async_client, admin_user):
|
"""Audit log items must never contain filename, extracted_text, password_hash,
|
||||||
"""Audit log entries contain no 'filename' or 'extracted_text' keys in metadata."""
|
or credentials_enc in any field — including nested inside metadata_ (ADMIN-06, D-15)."""
|
||||||
pytest.xfail("not implemented yet")
|
await _seed_audit(db_session, admin_user["user"].id)
|
||||||
|
|
||||||
|
response = await async_client.get(
|
||||||
|
"/api/admin/audit-log",
|
||||||
|
headers=admin_user["headers"],
|
||||||
|
)
|
||||||
|
|
||||||
|
assert response.status_code == 200
|
||||||
|
items = response.json()["items"]
|
||||||
|
|
||||||
|
forbidden_keys = {"filename", "extracted_text", "password_hash", "credentials_enc"}
|
||||||
|
|
||||||
|
for item in items:
|
||||||
|
# Top-level key check
|
||||||
|
for key in forbidden_keys:
|
||||||
|
assert key not in item, (
|
||||||
|
f"forbidden key '{key}' found at top level of audit item"
|
||||||
|
)
|
||||||
|
|
||||||
|
# Nested metadata_ check
|
||||||
|
meta = item.get("metadata_")
|
||||||
|
if isinstance(meta, dict):
|
||||||
|
for key in ("filename", "extracted_text"):
|
||||||
|
assert key not in meta, (
|
||||||
|
f"forbidden key '{key}' found inside metadata_ of audit item"
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.xfail(strict=False)
|
|
||||||
async def test_audit_log_regular_user_403(async_client, auth_user):
|
async def test_audit_log_regular_user_403(async_client, auth_user):
|
||||||
"""GET /api/admin/audit-log with regular user token returns 403."""
|
"""GET /api/admin/audit-log with a regular user token must return 403."""
|
||||||
pytest.xfail("not implemented yet")
|
response = await async_client.get(
|
||||||
|
"/api/admin/audit-log",
|
||||||
|
headers=auth_user["headers"],
|
||||||
|
)
|
||||||
|
assert response.status_code == 403
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.xfail(strict=False)
|
async def test_audit_log_export_csv(async_client, admin_user, db_session):
|
||||||
async def test_audit_log_export_csv(async_client, admin_user):
|
"""GET /api/admin/audit-log/export?format=csv returns CSV with correct headers."""
|
||||||
"""GET /api/admin/audit-log/export?format=csv returns CSV content-type."""
|
await _seed_audit(db_session, admin_user["user"].id)
|
||||||
pytest.xfail("not implemented yet")
|
|
||||||
|
response = await async_client.get(
|
||||||
|
"/api/admin/audit-log/export",
|
||||||
|
params={"format": "csv"},
|
||||||
|
headers=admin_user["headers"],
|
||||||
|
)
|
||||||
|
|
||||||
|
assert response.status_code == 200
|
||||||
|
|
||||||
|
content_type = response.headers.get("content-type", "")
|
||||||
|
assert content_type.startswith("text/csv"), (
|
||||||
|
f"expected content-type to start with 'text/csv', got '{content_type}'"
|
||||||
|
)
|
||||||
|
|
||||||
|
content_disposition = response.headers.get("content-disposition", "")
|
||||||
|
assert "audit-export.csv" in content_disposition, (
|
||||||
|
f"expected content-disposition to contain 'audit-export.csv', "
|
||||||
|
f"got '{content_disposition}'"
|
||||||
|
)
|
||||||
|
|
||||||
|
expected_header = (
|
||||||
|
"id,event_type,user_id,actor_id,resource_id,ip_address,metadata_,created_at"
|
||||||
|
)
|
||||||
|
assert expected_header in response.text, (
|
||||||
|
f"CSV header line not found in response. "
|
||||||
|
f"First 200 chars: {response.text[:200]!r}"
|
||||||
|
)
|
||||||
|
|||||||
Reference in New Issue
Block a user