File size: 3,940 Bytes
5b98477 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 | """
Tests — Feature 2.3: Admin Routes (Audit Logs, Cache Flush)
"""
import pytest
class TestAuditLogs:
async def test_admin_can_list_audit_logs(self, client, admin_token):
resp = await client.get(
"/admin/audit-logs",
headers={"Authorization": f"Bearer {admin_token}"},
)
assert resp.status_code == 200
data = resp.json()
assert "items" in data
assert "total" in data
assert "page" in data
async def test_audit_logs_default_pagination(self, client, admin_token):
resp = await client.get(
"/admin/audit-logs",
headers={"Authorization": f"Bearer {admin_token}"},
)
data = resp.json()
assert data["page"] == 1
assert data["limit"] <= 100
async def test_audit_logs_pagination_params(self, client, admin_token):
resp = await client.get(
"/admin/audit-logs?page=2&limit=5",
headers={"Authorization": f"Bearer {admin_token}"},
)
assert resp.status_code == 200
data = resp.json()
assert data["page"] == 2
assert data["limit"] == 5
async def test_audit_logs_filter_by_method(self, client, admin_token):
resp = await client.get(
"/admin/audit-logs?method=POST",
headers={"Authorization": f"Bearer {admin_token}"},
)
assert resp.status_code == 200
data = resp.json()
# All returned items should have method=POST (or list is empty)
for item in data["items"]:
assert item["method"] == "POST"
async def test_audit_logs_filter_by_endpoint(self, client, admin_token):
resp = await client.get(
"/admin/audit-logs?endpoint=/auth",
headers={"Authorization": f"Bearer {admin_token}"},
)
assert resp.status_code == 200
async def test_audit_logs_filter_by_user_id(self, client, admin_token, seeded_db):
user_id = seeded_db["super_admin"].id
resp = await client.get(
f"/admin/audit-logs?user_id={user_id}",
headers={"Authorization": f"Bearer {admin_token}"},
)
assert resp.status_code == 200
async def test_audit_logs_date_filter(self, client, admin_token):
resp = await client.get(
"/admin/audit-logs?start=2020-01-01T00:00:00&end=2099-12-31T23:59:59",
headers={"Authorization": f"Bearer {admin_token}"},
)
assert resp.status_code == 200
async def test_audit_logs_non_admin_forbidden(self, client, doctor_token):
resp = await client.get(
"/admin/audit-logs",
headers={"Authorization": f"Bearer {doctor_token}"},
)
assert resp.status_code == 403
class TestCacheFlush:
async def test_admin_can_flush_cache(self, client, admin_token):
resp = await client.post(
"/admin/cache/flush",
headers={"Authorization": f"Bearer {admin_token}"},
)
assert resp.status_code == 200
data = resp.json()
assert "message" in data
assert "keys_deleted" in data
async def test_cache_flush_returns_int_keys_deleted(self, client, admin_token):
resp = await client.post(
"/admin/cache/flush",
headers={"Authorization": f"Bearer {admin_token}"},
)
assert isinstance(resp.json()["keys_deleted"], int)
async def test_doctor_cannot_flush_cache(self, client, doctor_token):
resp = await client.post(
"/admin/cache/flush",
headers={"Authorization": f"Bearer {doctor_token}"},
)
assert resp.status_code == 403
async def test_viewer_cannot_flush_cache(self, client, viewer_token):
resp = await client.post(
"/admin/cache/flush",
headers={"Authorization": f"Bearer {viewer_token}"},
)
assert resp.status_code == 403
|