c83287819a
- 操作日志审计机制：audit 上下文（成功与失败都记、独立提交、操作人标量 固化、commit 后查询前置约定），用户/角色全部写操作接入；request_data 按键名的结构化脱敏 + UTF-8 字节边界截断 - 角色管理：CRUD 与启停；admin 禁删禁停、有绑定禁删（FOR UPDATE 锁行）、 编码创建后不可改、停用角色不可再分配 - 日志查询：操作/登录两端点，列表与详情两段式（详情含已脱敏请求参数）； 时间参数统一转本地 naive（修混用时区 500 与偏移丢弃） - 用户管理：补齐 reset_password 操作人、编辑空串归一 NULL、并发唯一键兜底 400 - 测试 301 例全绿（单元 223 / 集成 78） Co-Authored-By: Claude Code <noreply@anthropic.com>
199 lines
7.7 KiB
Python
199 lines
7.7 KiB
Python
"""日志查询接口集成测试（需要 MySQL 与 Redis）。
|
||
|
||
守的是：两类日志的筛选、分页、权限，以及**写入与查询的口径一致** ——
|
||
写操作产生的审计记录必须能被查询接口按同样口径检索到（否则"有写入
|
||
没查询"或"查询条件对不上"，日志功能形同虚设）。
|
||
|
||
登录日志造数直接 ORM 插入，不经过登录接口：绕开验证码依赖，且能精确
|
||
控制 success/fail_reason/ip 组合。
|
||
|
||
运行前置：``docker compose -f docker/docker-compose.dev.yml up -d``
|
||
"""
|
||
|
||
import uuid
|
||
from datetime import datetime, timedelta
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy import delete
|
||
|
||
from app.core.db import get_session_factory
|
||
from app.models import LoginLog
|
||
|
||
pytestmark = pytest.mark.integration
|
||
|
||
_OPERATION_LOGS = "/api/v1/system/operation-logs"
|
||
_LOGIN_LOGS = "/api/v1/system/login-logs"
|
||
_USERS = "/api/v1/system/users"
|
||
|
||
|
||
def _make_audit(client: TestClient, admin: dict, action: str = "删除") -> None:
|
||
"""触发一条审计记录（失败操作最省事、不留业务数据）。"""
|
||
resp = client.delete(f"{_USERS}/999999999", headers=admin["headers"])
|
||
assert resp.status_code == 404
|
||
assert action == "删除"
|
||
|
||
|
||
class TestOperationLogQuery:
|
||
"""操作日志查询。"""
|
||
|
||
def test_写操作后可按模块与操作筛选(self, client: TestClient, admin: dict, make_user) -> None:
|
||
target = make_user()
|
||
resp = client.delete(f"{_USERS}/{target['id']}", headers=admin["headers"])
|
||
assert resp.status_code == 200
|
||
|
||
resp = client.get(
|
||
_OPERATION_LOGS,
|
||
params={"username": admin["username"], "module": "用户管理", "action": "删除"},
|
||
headers=admin["headers"],
|
||
)
|
||
data = resp.json()["data"]
|
||
assert data["total"] >= 1
|
||
row = data["list"][0]
|
||
assert row["success"] == 1
|
||
assert target["username"] in row["summary"]
|
||
assert row["module"] == "用户管理"
|
||
|
||
def test_按结果筛选出失败记录(self, client: TestClient, admin: dict) -> None:
|
||
_make_audit(client, admin)
|
||
resp = client.get(
|
||
_OPERATION_LOGS,
|
||
params={"username": admin["username"], "success": 0},
|
||
headers=admin["headers"],
|
||
)
|
||
data = resp.json()["data"]
|
||
assert data["total"] >= 1
|
||
assert data["list"][0]["error_message"]
|
||
|
||
def test_时间范围筛选(self, client: TestClient, admin: dict) -> None:
|
||
_make_audit(client, admin)
|
||
now = datetime.now()
|
||
fmt = "%Y-%m-%d %H:%M:%S"
|
||
|
||
resp = client.get(
|
||
_OPERATION_LOGS,
|
||
params={
|
||
"username": admin["username"],
|
||
"start_time": (now - timedelta(minutes=5)).strftime(fmt),
|
||
"end_time": (now + timedelta(minutes=5)).strftime(fmt),
|
||
},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.json()["data"]["total"] >= 1
|
||
|
||
# 已过去的时间窗（排除测试刚产生的记录）
|
||
resp = client.get(
|
||
_OPERATION_LOGS,
|
||
params={
|
||
"username": admin["username"],
|
||
"start_time": "2020-01-01 00:00:00",
|
||
"end_time": "2020-01-02 00:00:00",
|
||
},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.json()["data"]["total"] == 0
|
||
|
||
def test_起始晚于结束返回400(self, client: TestClient, admin: dict) -> None:
|
||
resp = client.get(
|
||
_OPERATION_LOGS,
|
||
params={"start_time": "2026-10-02 00:00:00", "end_time": "2026-10-01 00:00:00"},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 400
|
||
|
||
def test_列表响应不含请求参数字段(self, client: TestClient, admin: dict) -> None:
|
||
"""列表为性能排除 request_data（TEXT 大字段），防止翻页负担随日志表增长。"""
|
||
resp = client.get(_OPERATION_LOGS, headers=admin["headers"])
|
||
rows = resp.json()["data"]["list"]
|
||
if rows:
|
||
assert "request_data" not in rows[0]
|
||
|
||
def test_详情端点返回已脱敏的请求参数(self, client: TestClient, admin: dict, make_user) -> None:
|
||
"""列表 → 详情 的两段式：详情含 request_data 且已脱敏。"""
|
||
secret = "S3cret@2026!"
|
||
target = make_user()
|
||
resp = client.put(
|
||
f"{_USERS}/{target['id']}/password",
|
||
json={"new_password": secret},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 200
|
||
|
||
resp = client.get(
|
||
_OPERATION_LOGS,
|
||
params={"username": admin["username"], "action": "重置密码"},
|
||
headers=admin["headers"],
|
||
)
|
||
row = resp.json()["data"]["list"][0]
|
||
assert "request_data" not in row
|
||
|
||
detail = client.get(f"{_OPERATION_LOGS}/{row['id']}", headers=admin["headers"])
|
||
assert detail.status_code == 200
|
||
request_data = detail.json()["data"]["request_data"] or ""
|
||
assert secret not in request_data
|
||
assert "***" in request_data
|
||
|
||
def test_详情端点不存在返回404(self, client: TestClient, admin: dict) -> None:
|
||
resp = client.get(f"{_OPERATION_LOGS}/999999999", headers=admin["headers"])
|
||
assert resp.status_code == 404
|
||
|
||
def test_详情端点普通用户被拒(self, client: TestClient, normal: dict) -> None:
|
||
resp = client.get(f"{_OPERATION_LOGS}/1", headers=normal["headers"])
|
||
assert resp.status_code == 403
|
||
|
||
def test_普通用户被拒(self, client: TestClient, normal: dict) -> None:
|
||
assert client.get(_OPERATION_LOGS, headers=normal["headers"]).status_code == 403
|
||
|
||
|
||
class TestLoginLogQuery:
|
||
"""登录日志查询。"""
|
||
|
||
@pytest.fixture
|
||
def seeded_logs(self):
|
||
"""直插两条登录日志（一成功一失败），用后清理。"""
|
||
username = f"t_login_{uuid.uuid4().hex[:6]}"
|
||
factory = get_session_factory()
|
||
with factory() as session:
|
||
session.add(
|
||
LoginLog(
|
||
username=username,
|
||
ip="10.1.2.3",
|
||
user_agent="pytest",
|
||
success=0,
|
||
fail_reason="用户名或密码错误",
|
||
)
|
||
)
|
||
session.add(LoginLog(username=username, ip="10.1.2.4", user_agent="pytest", success=1))
|
||
session.commit()
|
||
yield username
|
||
with factory() as session:
|
||
session.execute(delete(LoginLog).where(LoginLog.username == username))
|
||
session.commit()
|
||
|
||
def test_按用户名查询并返回两端结果(
|
||
self, client: TestClient, admin: dict, seeded_logs: str
|
||
) -> None:
|
||
resp = client.get(_LOGIN_LOGS, params={"username": seeded_logs}, headers=admin["headers"])
|
||
data = resp.json()["data"]
|
||
assert data["total"] == 2
|
||
assert {row["success"] for row in data["list"]} == {0, 1}
|
||
|
||
def test_按IP精确筛选(self, client: TestClient, admin: dict, seeded_logs: str) -> None:
|
||
resp = client.get(_LOGIN_LOGS, params={"ip": "10.1.2.3"}, headers=admin["headers"])
|
||
data = resp.json()["data"]
|
||
assert data["total"] == 1
|
||
assert data["list"][0]["fail_reason"] == "用户名或密码错误"
|
||
|
||
def test_按结果筛选(self, client: TestClient, admin: dict, seeded_logs: str) -> None:
|
||
resp = client.get(
|
||
_LOGIN_LOGS,
|
||
params={"username": seeded_logs, "success": 0},
|
||
headers=admin["headers"],
|
||
)
|
||
data = resp.json()["data"]
|
||
assert data["total"] == 1
|
||
assert data["list"][0]["success"] == 0
|
||
|
||
def test_普通用户被拒(self, client: TestClient, normal: dict) -> None:
|
||
assert client.get(_LOGIN_LOGS, headers=normal["headers"]).status_code == 403
|