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>
210 lines
8.4 KiB
Python
210 lines
8.4 KiB
Python
"""角色管理接口集成测试（需要 MySQL 与 Redis）。
|
||
|
||
守的是角色管理的**保护规则** —— 它们不是界面体验，而是系统自举与授权
|
||
安全的最低防线：
|
||
|
||
- `admin` 角色禁删禁停（cli 建管理员与 require_admin 判定都依赖该编码）
|
||
- 有用户绑定的角色禁删（user_roles 外键是 CASCADE，直接删会静默剥离权限）
|
||
- `role_code` 创建后不可改（授权判定的锚点）
|
||
|
||
运行前置：``docker compose -f docker/docker-compose.dev.yml up -d``
|
||
"""
|
||
|
||
import uuid
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
from app.core.constants import ROLE_USER
|
||
|
||
pytestmark = pytest.mark.integration
|
||
|
||
_ROLES = "/api/v1/system/roles"
|
||
|
||
|
||
def _new_code() -> str:
|
||
return f"t_role_{uuid.uuid4().hex[:6]}"
|
||
|
||
|
||
def _admin_role_id(client: TestClient, admin: dict) -> int:
|
||
"""从管理列表取内置 admin 角色的 id。"""
|
||
resp = client.get(_ROLES, params={"keyword": "admin"}, headers=admin["headers"])
|
||
rows = resp.json()["data"]["list"]
|
||
matches = [row for row in rows if row["role_code"] == "admin"]
|
||
assert matches, "内置 admin 角色缺失"
|
||
return matches[0]["id"]
|
||
|
||
|
||
class TestCrudFlow:
|
||
"""正向链路：建 → 查（列表 + 下拉）→ 改 → 停用 → 删。"""
|
||
|
||
def test_新增角色并联动列表与下拉(self, client: TestClient, admin: dict) -> None:
|
||
code = _new_code()
|
||
resp = client.post(
|
||
_ROLES,
|
||
json={"role_code": code, "role_name": "对外协作岗", "sort_order": 50},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
role_id = resp.json()["data"]["id"]
|
||
assert resp.json()["data"]["user_count"] == 0
|
||
|
||
# 管理列表可查（keyword 同时匹配编码与名称）
|
||
resp = client.get(_ROLES, params={"keyword": code}, headers=admin["headers"])
|
||
assert resp.json()["data"]["total"] == 1
|
||
|
||
# 用户表单下拉包含（启用状态）
|
||
resp = client.get("/api/v1/system/users/meta/roles", headers=admin["headers"])
|
||
assert code in [row["role_code"] for row in resp.json()["data"]]
|
||
|
||
assert client.delete(f"{_ROLES}/{role_id}", headers=admin["headers"]).status_code == 200
|
||
|
||
def test_重复编码被拒(self, client: TestClient, admin: dict, make_role) -> None:
|
||
role = make_role()
|
||
resp = client.post(
|
||
_ROLES,
|
||
json={"role_code": role["role_code"], "role_name": "重复"},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 400
|
||
assert "已存在" in resp.json()["message"]
|
||
|
||
def test_编码格式非法被拒(self, client: TestClient, admin: dict) -> None:
|
||
resp = client.post(
|
||
_ROLES, json={"role_code": "BadCode", "role_name": "大写"}, headers=admin["headers"]
|
||
)
|
||
assert resp.status_code == 400
|
||
|
||
def test_编辑角色名称与排序(self, client: TestClient, admin: dict, make_role) -> None:
|
||
role = make_role()
|
||
resp = client.put(
|
||
f"{_ROLES}/{role['id']}",
|
||
json={"role_name": "改名后", "description": "新的说明", "sort_order": 7},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 200
|
||
data = resp.json()["data"]
|
||
assert data["role_name"] == "改名后"
|
||
assert data["description"] == "新的说明"
|
||
assert data["sort_order"] == 7
|
||
|
||
def test_编码不可被修改(self, client: TestClient, admin: dict, make_role) -> None:
|
||
"""请求体携带 role_code 会被忽略（编辑模型不接收该字段），编码保持不变。"""
|
||
role = make_role()
|
||
resp = client.put(
|
||
f"{_ROLES}/{role['id']}",
|
||
json={"role_name": "改名", "role_code": "hacked"},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["data"]["role_code"] == role["role_code"]
|
||
|
||
def test_停用后不在下拉但仍在管理列表(self, client: TestClient, admin: dict, make_role) -> None:
|
||
role = make_role()
|
||
resp = client.put(
|
||
f"{_ROLES}/{role['id']}/status", json={"status": 0}, headers=admin["headers"]
|
||
)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["data"]["status"] == 0
|
||
|
||
resp = client.get("/api/v1/system/users/meta/roles", headers=admin["headers"])
|
||
assert role["role_code"] not in [row["role_code"] for row in resp.json()["data"]]
|
||
|
||
resp = client.get(_ROLES, params={"keyword": role["role_code"]}, headers=admin["headers"])
|
||
assert resp.json()["data"]["total"] == 1
|
||
|
||
|
||
class TestProtection:
|
||
"""保护规则：内置角色与有绑定的角色。"""
|
||
|
||
def test_admin角色不可删除(self, client: TestClient, admin: dict) -> None:
|
||
role_id = _admin_role_id(client, admin)
|
||
resp = client.delete(f"{_ROLES}/{role_id}", headers=admin["headers"])
|
||
assert resp.status_code == 403
|
||
assert "不可删除" in resp.json()["message"]
|
||
|
||
def test_admin角色不可停用(self, client: TestClient, admin: dict) -> None:
|
||
role_id = _admin_role_id(client, admin)
|
||
resp = client.put(
|
||
f"{_ROLES}/{role_id}/status", json={"status": 0}, headers=admin["headers"]
|
||
)
|
||
assert resp.status_code == 403
|
||
assert "不可停用" in resp.json()["message"]
|
||
|
||
def test_有绑定的角色不可删除且解绑后可删除(
|
||
self, client: TestClient, admin: dict, make_role, make_user
|
||
) -> None:
|
||
"""`user_roles` 外键是 CASCADE —— 后端 400 拦截是防止静默剥离权限的
|
||
唯一防线，前端按钮禁用只是体验。"""
|
||
role = make_role()
|
||
bound = make_user([role["role_code"]])
|
||
|
||
resp = client.delete(f"{_ROLES}/{role['id']}", headers=admin["headers"])
|
||
assert resp.status_code == 400
|
||
assert "1 个用户" in resp.json()["message"]
|
||
|
||
# 解绑（换回内置 user 角色）后可删除
|
||
resp = client.put(
|
||
f"/api/v1/system/users/{bound['id']}",
|
||
json={"roles": [ROLE_USER]},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 200
|
||
assert client.delete(f"{_ROLES}/{role['id']}", headers=admin["headers"]).status_code == 200
|
||
|
||
def test_列表返回绑定用户数(
|
||
self, client: TestClient, admin: dict, make_role, make_user
|
||
) -> None:
|
||
role = make_role()
|
||
make_user([role["role_code"]])
|
||
resp = client.get(_ROLES, params={"keyword": role["role_code"]}, headers=admin["headers"])
|
||
assert resp.json()["data"]["list"][0]["user_count"] == 1
|
||
|
||
def test_编辑不存在的角色404(self, client: TestClient, admin: dict) -> None:
|
||
resp = client.put(
|
||
f"{_ROLES}/999999999", json={"role_name": "不存在"}, headers=admin["headers"]
|
||
)
|
||
assert resp.status_code == 404
|
||
|
||
def test_停用角色不可再分配给用户(
|
||
self, client: TestClient, admin: dict, make_role, make_user
|
||
) -> None:
|
||
"""回归：停用只挡住了表单下拉，接口层此前仍可绑定（或使用停用前
|
||
加载的旧选项提交）—— 与「停用后不可再分配」的语义矛盾。"""
|
||
role = make_role()
|
||
resp = client.put(
|
||
f"{_ROLES}/{role['id']}/status", json={"status": 0}, headers=admin["headers"]
|
||
)
|
||
assert resp.status_code == 200
|
||
|
||
username = f"t_{uuid.uuid4().hex[:8]}"
|
||
resp = client.post(
|
||
"/api/v1/system/users",
|
||
json={"username": username, "password": "TestPwd@2026", "roles": [role["role_code"]]},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 400
|
||
assert "停用" in resp.json()["message"]
|
||
|
||
# 已绑定该角色的用户（历史数据）改回启用角色后可正常保存
|
||
bound = make_user([role["role_code"]])
|
||
resp = client.put(
|
||
f"/api/v1/system/users/{bound['id']}",
|
||
json={"roles": [ROLE_USER]},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 200
|
||
|
||
|
||
class TestAuthorization:
|
||
"""权限：角色管理为管理员专属。"""
|
||
|
||
def test_普通用户访问被拒(self, client: TestClient, normal: dict) -> None:
|
||
assert client.get(_ROLES, headers=normal["headers"]).status_code == 403
|
||
resp = client.post(
|
||
_ROLES,
|
||
json={"role_code": _new_code(), "role_name": "越权"},
|
||
headers=normal["headers"],
|
||
)
|
||
assert resp.status_code == 403
|