"""日志查询接口集成测试（需要 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