curo/kite
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test(06.2-04): add failing tests for handle enrichment, user_handle filter, daily exports

Browse Source 
- test_audit_log_includes_user_handle: asserts user_handle/actor_handle in items
- test_audit_log_filter_by_handle: asserts filtering by handle works correctly
- test_audit_log_filter_unknown_handle: asserts 200+empty for unknown handle
- test_daily_exports_list: mocks MinIO list_objects, asserts sorted items
- test_daily_export_download: mocks MinIO get_object, asserts CSV response + 404 on bad date
This commit is contained in:
curo1305
2026-05-31 15:15:46 +02:00
parent eab5f124f6
commit d7cfc5ccee
1 changed files with 135 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/test_audit.py
+135 -7
View File
@@ -195,30 +195,158 @@ async def test_audit_log_export_csv(async_client, admin_user, db_session):
# ---------------------------------------------------------------------------
# Phase 6.2 Wave 0 xfail stubs — ADMIN-06 audit enrichment + daily exports
# Phase 6.2 — ADMIN-06 audit enrichment + daily exports (promoted stubs)
# ---------------------------------------------------------------------------
async def test_audit_log_includes_user_handle(async_client, admin_user, db_session):
"""Audit log items include user_handle and actor_handle strings (D-11)"""
pytest.xfail("Phase 6.2 — 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
body = response.json()
items = body["items"]
assert len(items) >= 1, "expected at least one seeded audit entry"
first = items[0]
assert "user_handle" in first, "missing key 'user_handle' in audit item"
assert "actor_handle" in first, "missing key 'actor_handle' in audit item"
# The seeded entry was created for admin_user — handle must match
assert first["user_handle"] == admin_user["user"].handle, (
f"expected user_handle={admin_user['user'].handle!r}, got {first['user_handle']!r}"
)
async def test_audit_log_filter_by_handle(async_client, admin_user, db_session):
async def test_audit_log_filter_by_handle(async_client, admin_user, db_session, second_auth_user):
"""GET /api/admin/audit-log?user_handle=X filters to matching entries (D-12)"""
pytest.xfail("Phase 6.2 — not implemented yet")
# Seed one entry for admin_user and one for second_auth_user
await _seed_audit(db_session, admin_user["user"].id)
await _seed_audit(db_session, second_auth_user["user"].id)
response = await async_client.get(
"/api/admin/audit-log",
params={"user_handle": admin_user["user"].handle},
headers=admin_user["headers"],
)
assert response.status_code == 200
body = response.json()
items = body["items"]
assert len(items) >= 1, "expected at least one filtered entry for admin_user"
for item in items:
assert item["user_handle"] == admin_user["user"].handle, (
f"filter returned entry for wrong user: {item['user_handle']!r}"
)
# Second user's entry must not appear
second_handle = second_auth_user["user"].handle
assert not any(item["user_handle"] == second_handle for item in items), (
f"second user's entry appeared in filtered results"
)
async def test_audit_log_filter_unknown_handle(async_client, admin_user, db_session):
"""GET /api/admin/audit-log?user_handle=unknown returns empty items list, not 422 (D-12)"""
pytest.xfail("Phase 6.2 — not implemented yet")
response = await async_client.get(
"/api/admin/audit-log",
params={"user_handle": "definitely_does_not_exist"},
headers=admin_user["headers"],
)
assert response.status_code == 200, (
f"expected 200 for unknown handle, got {response.status_code}: {response.text[:200]}"
)
body = response.json()
assert body["items"] == [], f"expected empty items list, got {body['items']}"
assert body["total"] == 0, f"expected total=0, got {body['total']}"
async def test_daily_exports_list(async_client, admin_user):
"""GET /api/admin/audit-log/daily-exports returns {items: [...]} (D-15)"""
pytest.xfail("Phase 6.2 — not implemented yet")
from unittest.mock import MagicMock, patch
# Create fake MinIO objects
fake_obj1 = MagicMock()
fake_obj1.object_name = "audit-logs/2026-05-30.csv"
fake_obj1.is_dir = False
fake_obj2 = MagicMock()
fake_obj2.object_name = "audit-logs/2026-05-29.csv"
fake_obj2.is_dir = False
mock_client = MagicMock()
mock_client.list_objects.return_value = iter([fake_obj1, fake_obj2])
mock_backend = MagicMock()
mock_backend._client = mock_client
from storage.minio_backend import MinIOBackend
with patch("api.audit.get_storage_backend", return_value=mock_backend), \
patch("api.audit.MinIOBackend", MinIOBackend):
response = await async_client.get(
"/api/admin/audit-log/daily-exports",
headers=admin_user["headers"],
)
assert response.status_code == 200
body = response.json()
assert "items" in body, f"expected 'items' key in response, got: {body}"
items = body["items"]
assert isinstance(items, list)
# Items must be sorted descending by date
if len(items) >= 2:
dates = [item["date"] for item in items]
assert dates == sorted(dates, reverse=True), (
f"expected dates sorted descending, got {dates}"
)
async def test_daily_export_download(async_client, admin_user):
"""GET /api/admin/audit-log/daily-exports/{date} returns CSV bytes with Content-Disposition (D-16)"""
pytest.xfail("Phase 6.2 — not implemented yet")
from unittest.mock import MagicMock, patch
fake_csv = b"id,event_type,user_id\n1,document.uploaded,abc\n"
mock_response = MagicMock()
mock_response.read.return_value = fake_csv
mock_response.close.return_value = None
mock_response.release_conn.return_value = None
mock_client = MagicMock()
mock_client.get_object.return_value = mock_response
mock_backend = MagicMock()
mock_backend._client = mock_client
with patch("api.audit.get_storage_backend", return_value=mock_backend):
response = await async_client.get(
"/api/admin/audit-log/daily-exports/2026-05-30",
headers=admin_user["headers"],
)
assert response.status_code == 200
content_type = response.headers.get("content-type", "")
assert "text/csv" in content_type, (
f"expected content-type text/csv, got {content_type!r}"
)
content_disposition = response.headers.get("content-disposition", "")
assert "2026-05-30" in content_disposition, (
f"expected '2026-05-30' in Content-Disposition, got {content_disposition!r}"
)
# Invalid date must return 404
with patch("api.audit.get_storage_backend", return_value=mock_backend):
bad_response = await async_client.get(
"/api/admin/audit-log/daily-exports/invalid-date",
headers=admin_user["headers"],
)
assert bad_response.status_code == 404, (
f"expected 404 for invalid date, got {bad_response.status_code}"
)