| """稽核日誌後台 endpoint 測試。""" |
| from __future__ import annotations |
|
|
| import pytest |
| from fastapi.testclient import TestClient |
|
|
| from app.auth.passwords import hash_password |
| from app.db import _utcnow, get_conn |
| from app.main import app |
|
|
| PWD = "TestPw123!" |
|
|
|
|
| @pytest.fixture() |
| def client(): |
| return TestClient(app) |
|
|
|
|
| def _mkuser(role, name, dept="業務部"): |
| c = get_conn() |
| c.execute( |
| "INSERT INTO users (username, email, password_hash, role, department, " |
| "created_at, is_active) VALUES (?,?,?,?,?,?,1)", |
| (name, f"{name}@x.tw", hash_password(PWD), role, dept, _utcnow()), |
| ) |
| return c.execute("SELECT id FROM users WHERE username=?", (name,)).fetchone()["id"] |
|
|
|
|
| def _mkaudit(user_id, action, ts=None, *, result="success", detail=None): |
| get_conn().execute( |
| "INSERT INTO audit_log (user_id, action, resource_type, resource_id, " |
| "detail, result, timestamp) VALUES (?,?,?,?,?,?,?)", |
| (user_id, action, "x", "0", detail, result, ts or _utcnow()), |
| ) |
|
|
|
|
| def _login(c, u): |
| return c.post("/api/auth/login", json={"username": u, "password": PWD}).json()["access_token"] |
|
|
|
|
| def _auth(t): |
| return {"Authorization": f"Bearer {t}"} |
|
|
|
|
| |
|
|
| def test_admin_can_list_audit(client): |
| uid = _mkuser("admin", "adm") |
| _mkaudit(uid, "login_success") |
| _mkaudit(uid, "view_document") |
| tok = _login(client, "adm") |
| r = client.get("/api/admin/audit-log", headers=_auth(tok)) |
| assert r.status_code == 200 |
| j = r.json() |
| assert "items" in j and "total" in j |
| assert j["total"] >= 2 |
| |
| assert all("category" in it and "risk" in it for it in j["items"]) |
|
|
|
|
| def test_user_blocked(client): |
| _mkuser("admin", "adm") |
| _mkuser("user", "u1") |
| tok = _login(client, "u1") |
| r = client.get("/api/admin/audit-log", headers=_auth(tok)) |
| assert r.status_code in (401, 403) |
|
|
|
|
| def test_filter_by_user(client): |
| a = _mkuser("admin", "adm") |
| b = _mkuser("user", "u1") |
| _mkaudit(a, "view_document") |
| _mkaudit(b, "view_document") |
| tok = _login(client, "adm") |
| r = client.get(f"/api/admin/audit-log?user_id={b}", headers=_auth(tok)) |
| items = r.json()["items"] |
| assert all(it["user_id"] == b for it in items) |
|
|
|
|
| def test_filter_by_action(client): |
| uid = _mkuser("admin", "adm") |
| _mkaudit(uid, "login_success") |
| _mkaudit(uid, "delete_document") |
| tok = _login(client, "adm") |
| r = client.get("/api/admin/audit-log?action=delete_document", headers=_auth(tok)) |
| items = r.json()["items"] |
| assert all(it["action"] == "delete_document" for it in items) |
|
|
|
|
| def test_filter_by_category_security(client): |
| uid = _mkuser("admin", "adm") |
| _mkaudit(uid, "login_failed", result="fail") |
| _mkaudit(uid, "view_document") |
| tok = _login(client, "adm") |
| r = client.get("/api/admin/audit-log?category=security", headers=_auth(tok)) |
| items = r.json()["items"] |
| assert all(it["category"] == "security" for it in items) |
| assert any(it["action"] == "login_failed" for it in items) |
|
|
|
|
| def test_filter_by_risk_high(client): |
| uid = _mkuser("admin", "adm") |
| _mkaudit(uid, "delete_user") |
| _mkaudit(uid, "view_document") |
| tok = _login(client, "adm") |
| r = client.get("/api/admin/audit-log?risk=high", headers=_auth(tok)) |
| items = r.json()["items"] |
| assert all(it["risk"] == "high" for it in items) |
|
|
|
|
| def test_pagination(client): |
| uid = _mkuser("admin", "adm") |
| for i in range(10): |
| _mkaudit(uid, f"action_{i}") |
| tok = _login(client, "adm") |
| r1 = client.get("/api/admin/audit-log?limit=5&offset=0", headers=_auth(tok)) |
| r2 = client.get("/api/admin/audit-log?limit=5&offset=5", headers=_auth(tok)) |
| assert len(r1.json()["items"]) == 5 |
| assert len(r2.json()["items"]) >= 1 |
| ids1 = {it["id"] for it in r1.json()["items"]} |
| ids2 = {it["id"] for it in r2.json()["items"]} |
| assert not (ids1 & ids2) |
|
|
|
|
| |
|
|
| def test_stats_keys(client): |
| _mkuser("admin", "adm") |
| tok = _login(client, "adm") |
| r = client.get("/api/admin/audit-log/stats", headers=_auth(tok)) |
| assert r.status_code == 200 |
| j = r.json() |
| assert set(j) == {"today_events", "high_risk_7d", |
| "failed_login_7d", "confidential_access_7d"} |
|
|
|
|
| def test_stats_today_count(client): |
| uid = _mkuser("admin", "adm") |
| _mkaudit(uid, "view_document") |
| _mkaudit(uid, "view_document") |
| tok = _login(client, "adm") |
| j = client.get("/api/admin/audit-log/stats", headers=_auth(tok)).json() |
| assert j["today_events"] >= 2 |
|
|
|
|
| def test_stats_high_risk_count(client): |
| uid = _mkuser("admin", "adm") |
| _mkaudit(uid, "delete_user") |
| _mkaudit(uid, "delete_document") |
| _mkaudit(uid, "view_document") |
| tok = _login(client, "adm") |
| j = client.get("/api/admin/audit-log/stats", headers=_auth(tok)).json() |
| assert j["high_risk_7d"] >= 2 |
|
|
