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>
365 lines
15 KiB
Python
365 lines
15 KiB
Python
"""用户管理接口集成测试（需要 MySQL 与 Redis）。
|
||
|
||
守的是：用户 CRUD 的**串联正确性**与**审计留痕**。
|
||
|
||
审计是重点验收项：每个写操作（含失败的）都必须在 operation_logs 留下
|
||
可读记录 —— 模块、操作、摘要、操作人、请求元信息完整，且 request_data
|
||
中**绝无明文密码**；账号被删除后审计行通过「user_id 置空 + 冗余 username」
|
||
保留可追溯性。
|
||
|
||
运行前置：``docker compose -f docker/docker-compose.dev.yml up -d``
|
||
"""
|
||
|
||
import uuid
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy import select
|
||
|
||
from app.core.db import get_session_factory
|
||
from app.models import OperationLog
|
||
|
||
pytestmark = pytest.mark.integration
|
||
|
||
_PASSWORD = "TestPwd@2026"
|
||
_USERS = "/api/v1/system/users"
|
||
|
||
|
||
def _new_username() -> str:
|
||
return f"t_{uuid.uuid4().hex[:8]}"
|
||
|
||
|
||
def _create_via_api(client: TestClient, admin: dict, username: str, **overrides: object) -> dict:
|
||
"""经接口创建用户并返回响应 data（账号归 autouse 清理兜底）。"""
|
||
payload: dict = {
|
||
"username": username,
|
||
"password": _PASSWORD,
|
||
"real_name": "接口新用户",
|
||
"roles": ["user"],
|
||
}
|
||
payload.update(overrides)
|
||
resp = client.post(_USERS, json=payload, headers=admin["headers"])
|
||
assert resp.status_code == 200, resp.text
|
||
return resp.json()["data"]
|
||
|
||
|
||
def _logs_of(username: str) -> list[OperationLog]:
|
||
"""取某操作人最近的审计行（按 id 倒序）。"""
|
||
factory = get_session_factory()
|
||
with factory() as session:
|
||
rows = list(
|
||
session.execute(
|
||
select(OperationLog)
|
||
.where(OperationLog.username == username)
|
||
.order_by(OperationLog.id.desc())
|
||
).scalars()
|
||
)
|
||
return rows
|
||
|
||
|
||
class TestCrudFlow:
|
||
"""正向链路：建 → 查 → 改 → 启停 → 重置密码 → 删。"""
|
||
|
||
def test_新增用户并可在列表查到(self, client: TestClient, admin: dict) -> None:
|
||
username = _new_username()
|
||
data = _create_via_api(client, admin, username)
|
||
assert data["username"] == username
|
||
assert data["roles"] == ["user"]
|
||
assert data["status"] == 1
|
||
|
||
resp = client.get(_USERS, params={"username": username}, headers=admin["headers"])
|
||
body = resp.json()["data"]
|
||
assert body["total"] == 1
|
||
assert body["list"][0]["id"] == data["id"]
|
||
|
||
def test_重名用户被拒(self, client: TestClient, admin: dict, make_user) -> None:
|
||
target = make_user()
|
||
resp = client.post(
|
||
_USERS,
|
||
json={"username": target["username"], "password": _PASSWORD, "roles": ["user"]},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 400
|
||
assert "已存在" in resp.json()["message"]
|
||
|
||
def test_分页参数以驼峰命名生效(self, client: TestClient, admin: dict) -> None:
|
||
"""pageNum/pageSize 是前端契约（PageQuery 的 alias）。
|
||
|
||
防回归：若被改成蛇形命名，FastAPI 会忽略这两个参数、恒返回
|
||
第 1 页 10 条 —— 表现为「翻页没反应」，且没有任何报错。
|
||
"""
|
||
resp = client.get(_USERS, params={"pageNum": 2, "pageSize": 5}, headers=admin["headers"])
|
||
data = resp.json()["data"]
|
||
assert {"list", "total", "pageNum", "pageSize"} <= set(data)
|
||
assert data["pageNum"] == 2
|
||
assert data["pageSize"] == 5
|
||
assert len(data["list"]) <= 5
|
||
|
||
def test_编辑用户信息与角色(self, client: TestClient, admin: dict, make_user) -> None:
|
||
target = make_user()
|
||
resp = client.put(
|
||
f"{_USERS}/{target['id']}",
|
||
json={"real_name": "改后姓名", "roles": ["admin"]},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 200
|
||
data = resp.json()["data"]
|
||
assert data["real_name"] == "改后姓名"
|
||
assert data["roles"] == ["admin"]
|
||
|
||
def test_编辑用户保留既有角色不触发唯一键冲突(
|
||
self, client: TestClient, admin: dict, make_user
|
||
) -> None:
|
||
"""回归：新旧角色集合有交集时（编辑时未改角色必然如此），
|
||
`replace_roles` 若在 clear 后不先落库删除，SQLAlchemy 同一 flush
|
||
内先 INSERT 后 DELETE 会撞 uk_user_roles_user_id_role_id（1062）。
|
||
该缺陷最初由前端 e2e「编辑用户」步骤暴露 —— 集成测试当时只覆盖
|
||
了「换掉全部角色」的场景。"""
|
||
target = make_user(["user"])
|
||
resp = client.put(
|
||
f"{_USERS}/{target['id']}",
|
||
json={"real_name": "保留角色", "roles": ["user", "admin"]},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
assert sorted(resp.json()["data"]["roles"]) == ["admin", "user"]
|
||
|
||
def test_禁用与重新启用(self, client: TestClient, admin: dict, make_user) -> None:
|
||
target = make_user()
|
||
resp = client.put(
|
||
f"{_USERS}/{target['id']}/status", json={"status": 0}, headers=admin["headers"]
|
||
)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["data"]["status"] == 0
|
||
|
||
resp = client.put(
|
||
f"{_USERS}/{target['id']}/status", json={"status": 1}, headers=admin["headers"]
|
||
)
|
||
assert resp.json()["data"]["status"] == 1
|
||
|
||
def test_禁用后旧令牌立即失效(
|
||
self, client: TestClient, admin: dict, make_user, auth_headers
|
||
) -> None:
|
||
"""禁用会提 token_version —— 不依赖 Redis 的「离职即失效」。"""
|
||
target = make_user()
|
||
old_headers = auth_headers(target)
|
||
assert client.get("/api/v1/auth/me", headers=old_headers).status_code == 200
|
||
|
||
client.put(f"{_USERS}/{target['id']}/status", json={"status": 0}, headers=admin["headers"])
|
||
assert client.get("/api/v1/auth/me", headers=old_headers).status_code == 401
|
||
|
||
def test_重置密码后旧令牌立即失效(
|
||
self, client: TestClient, admin: dict, make_user, auth_headers
|
||
) -> None:
|
||
target = make_user()
|
||
old_headers = auth_headers(target)
|
||
assert client.get("/api/v1/auth/me", headers=old_headers).status_code == 200
|
||
|
||
resp = client.put(
|
||
f"{_USERS}/{target['id']}/password",
|
||
json={"new_password": "NewPwd@2026"},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 200
|
||
assert client.get("/api/v1/auth/me", headers=old_headers).status_code == 401
|
||
|
||
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(f"{_USERS}/{target['id']}", headers=admin["headers"])
|
||
assert resp.status_code == 404
|
||
|
||
|
||
class TestSelfProtection:
|
||
"""自我保护：不允许对当前登录账号执行危险操作。"""
|
||
|
||
def test_不能删除自己(self, client: TestClient, admin: dict) -> None:
|
||
resp = client.delete(f"{_USERS}/{admin['id']}", headers=admin["headers"])
|
||
assert resp.status_code == 403
|
||
|
||
def test_不能禁用自己(self, client: TestClient, admin: dict) -> None:
|
||
resp = client.put(
|
||
f"{_USERS}/{admin['id']}/status", json={"status": 0}, headers=admin["headers"]
|
||
)
|
||
assert resp.status_code == 403
|
||
|
||
def test_不能移除自己的管理员角色(self, client: TestClient, admin: dict) -> None:
|
||
resp = client.put(
|
||
f"{_USERS}/{admin['id']}", json={"roles": ["user"]}, headers=admin["headers"]
|
||
)
|
||
assert resp.status_code == 403
|
||
|
||
|
||
class TestValidation:
|
||
"""参数与业务校验。"""
|
||
|
||
def test_无效角色编码被拒(self, client: TestClient, admin: dict) -> None:
|
||
resp = client.post(
|
||
_USERS,
|
||
json={"username": _new_username(), "password": _PASSWORD, "roles": ["nonexist"]},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 400
|
||
assert "角色编码无效" in resp.json()["message"]
|
||
|
||
def test_空角色列表被拒(self, client: TestClient, admin: dict) -> None:
|
||
resp = client.post(
|
||
_USERS,
|
||
json={"username": _new_username(), "password": _PASSWORD, "roles": []},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 400
|
||
|
||
def test_密码25个汉字被拒(self, client: TestClient, admin: dict) -> None:
|
||
"""75 字节超 bcrypt 上限；按字符数校验拦不住（25 字符很短）。"""
|
||
resp = client.post(
|
||
_USERS,
|
||
json={"username": _new_username(), "password": "中" * 25, "roles": ["user"]},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 400
|
||
|
||
def test_查询不存在的用户返回404(self, client: TestClient, admin: dict) -> None:
|
||
resp = client.get(f"{_USERS}/999999999", headers=admin["headers"])
|
||
assert resp.status_code == 404
|
||
|
||
def test_编辑未修改字段不污染空值语义(self, client: TestClient, admin: dict, make_user) -> None:
|
||
"""回归：编辑表单把未修改字段以空串提交，此前会把库中 NULL 写成 ''——
|
||
两类空值（未填 vs 空串）语义漂移且不可逆；'' 现归一为 NULL。"""
|
||
target = make_user() # 直建账号 phone/email/remark 均为 NULL
|
||
resp = client.put(
|
||
f"{_USERS}/{target['id']}",
|
||
json={"real_name": "改名", "phone": "", "email": "", "remark": ""},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 200
|
||
data = resp.json()["data"]
|
||
assert data["real_name"] == "改名"
|
||
assert data["phone"] is None
|
||
assert data["email"] is None
|
||
assert data["remark"] is None
|
||
|
||
|
||
class TestFiltering:
|
||
"""筛选条件的口径。"""
|
||
|
||
def test_含下划线的用户名不误配(self, client: TestClient, admin: dict, make_user) -> None:
|
||
"""回归：模糊筛选未转义 LIKE 通配符时 `_` 匹配任意单字符，
|
||
搜 t_xx 会命中 txXx 之类无关账号 —— 用户名普遍含下划线，非理论场景。"""
|
||
target = make_user()
|
||
decoy_name = target["username"].replace("_", "x", 1)
|
||
make_user(username=decoy_name)
|
||
|
||
resp = client.get(_USERS, params={"username": target["username"]}, headers=admin["headers"])
|
||
names = [row["username"] for row in resp.json()["data"]["list"]]
|
||
assert target["username"] in names
|
||
assert decoy_name not in names
|
||
|
||
|
||
class TestAuditTrail:
|
||
"""审计留痕 —— 本模块的核心验收点。"""
|
||
|
||
def test_每个写操作都留下审计记录(self, client: TestClient, admin: dict, make_user) -> None:
|
||
target = make_user()
|
||
client.put(
|
||
f"{_USERS}/{target['id']}",
|
||
json={"real_name": "审计留痕"},
|
||
headers=admin["headers"],
|
||
)
|
||
client.put(f"{_USERS}/{target['id']}/status", json={"status": 0}, headers=admin["headers"])
|
||
client.put(
|
||
f"{_USERS}/{target['id']}/password",
|
||
json={"new_password": "AuditPwd@2026"},
|
||
headers=admin["headers"],
|
||
)
|
||
client.delete(f"{_USERS}/{target['id']}", headers=admin["headers"])
|
||
|
||
logs = _logs_of(admin["username"])
|
||
actions = {(log.module, log.action) for log in logs}
|
||
assert ("用户管理", "修改") in actions
|
||
assert ("用户管理", "禁用") in actions
|
||
assert ("用户管理", "重置密码") in actions
|
||
assert ("用户管理", "删除") in actions
|
||
|
||
# 摘要含目标用户名（供人阅读的追溯锚点）
|
||
assert any(target["username"] in log.summary for log in logs)
|
||
|
||
# 请求元信息与结果完整
|
||
success_logs = [log for log in logs if log.success == 1]
|
||
assert success_logs
|
||
for log in success_logs:
|
||
assert log.method and log.path
|
||
assert log.duration_ms is not None and log.duration_ms >= 0
|
||
|
||
def test_失败操作记为失败且带中文原因(self, client: TestClient, admin: dict) -> None:
|
||
resp = client.delete(f"{_USERS}/999999999", headers=admin["headers"])
|
||
assert resp.status_code == 404
|
||
|
||
latest = _logs_of(admin["username"])[0]
|
||
assert latest.success == 0
|
||
assert latest.action == "删除"
|
||
assert latest.error_message == "用户不存在"
|
||
|
||
def test_审计请求参数绝不包含明文密码(self, client: TestClient, admin: dict, make_user) -> None:
|
||
"""密码是审计脱敏的红线：全表扫描该操作人的 request_data 不得命中。"""
|
||
secret = "S3cret@2026!"
|
||
_create_via_api(client, admin, _new_username(), password=secret)
|
||
target = make_user()
|
||
client.put(
|
||
f"{_USERS}/{target['id']}/password",
|
||
json={"new_password": secret},
|
||
headers=admin["headers"],
|
||
)
|
||
|
||
factory = get_session_factory()
|
||
with factory() as session:
|
||
rows = list(
|
||
session.execute(
|
||
select(OperationLog.request_data).where(
|
||
OperationLog.username == admin["username"]
|
||
)
|
||
).scalars()
|
||
)
|
||
joined = "\n".join(row for row in rows if row)
|
||
assert secret not in joined
|
||
assert "***" in joined # 键名保留、值被抹除
|
||
|
||
def test_账号删除后审计行保留用户名(
|
||
self, client: TestClient, admin: dict, make_user, auth_headers
|
||
) -> None:
|
||
"""user_id 外键 SET NULL + username 冗余 —— 删除账号不得让审计断链。"""
|
||
victim = make_user(["admin"])
|
||
victim_headers = auth_headers(victim)
|
||
# victim 作为操作人产生一条审计（失败操作最省事，不留业务数据）
|
||
client.delete(f"{_USERS}/999999999", headers=victim_headers)
|
||
|
||
assert (
|
||
client.delete(f"{_USERS}/{victim['id']}", headers=admin["headers"]).status_code == 200
|
||
)
|
||
|
||
factory = get_session_factory()
|
||
with factory() as session:
|
||
logs = list(
|
||
session.execute(
|
||
select(OperationLog).where(OperationLog.username == victim["username"])
|
||
).scalars()
|
||
)
|
||
assert logs, "审计行不应随账号删除而消失"
|
||
assert all(log.user_id is None for log in logs)
|
||
assert all(log.username == victim["username"] for log in logs)
|
||
|
||
|
||
class TestAuthorization:
|
||
"""权限：系统管理为管理员专属。"""
|
||
|
||
def test_普通用户访问用户管理被拒(self, client: TestClient, normal: dict) -> None:
|
||
assert client.get(_USERS, headers=normal["headers"]).status_code == 403
|
||
resp = client.post(
|
||
_USERS,
|
||
json={"username": _new_username(), "password": _PASSWORD, "roles": ["user"]},
|
||
headers=normal["headers"],
|
||
)
|
||
assert resp.status_code == 403
|