liuwenxin
1461a8d113
按 2026-11-02 需求补齐计划交付后端能力： - 角色授权：权限目录化，接口统一准入校验，未登记默认拒绝； 管理员判断改为按数据库角色重新判定，防止权限码持有者绕过 - 系统监控：单进程采集资源、请求与异常指标，分钟级聚合， 阈值配置化；敏感配置增加 repr 保护，Redis 密码 URL 转义 - 试用期计提：按回款日判定并写入快照，年度结清与调整单联动； 人员模型补试用期、离职日期与工资版本历史字段 - 计提模板：公式骨架受控，页面维护名称与参数绑定 - T+ 采集：自动化配置表与进度查询 内置角色禁停用保护对齐模块注释；核算规则版本提为常量并 补充批量核算失败堆栈日志。 影响范围：后端全部模块、数据库迁移 V22-V28 验证：uv run ruff check 通过；922 个单元测试全部通过； 集成测试因本机 MySQL 数据卷被其他容器锁定暂未运行
283 lines
12 KiB
Python
283 lines
12 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 TestInitialPermissions:
|
|
def test_角色和权限一起创建且绑定后原令牌即时继承(self, client, admin, normal):
|
|
import json
|
|
|
|
from sqlalchemy import select
|
|
|
|
from app.core.db import get_session_factory
|
|
from app.models import OperationLog
|
|
|
|
codes = ["base:personnel:view", "base:personnel:regions"]
|
|
code = _new_code()
|
|
response = client.post(_ROLES, headers=admin["headers"], json={
|
|
"role_code": code, "role_name": "区域维护", "permission_codes": codes,
|
|
})
|
|
assert response.status_code == 200, response.text
|
|
role = response.json()["data"]
|
|
assert role["permission_count"] == 2
|
|
with get_session_factory()() as session:
|
|
log = session.scalar(select(OperationLog).where(
|
|
OperationLog.user_id == admin["id"], OperationLog.summary.contains(code),
|
|
OperationLog.success == 1,
|
|
))
|
|
assert log is not None
|
|
request_data = json.loads(log.request_data)
|
|
assert request_data["reason"] == "创建角色并配置初始权限"
|
|
assert request_data["permission_codes"] == codes
|
|
stored = client.get(f"{_ROLES}/{role['id']}/permissions", headers=admin["headers"]).json()["data"]
|
|
assert set(stored["permission_codes"]) == set(codes)
|
|
before = client.get(f"/api/v1/system/users/{normal['id']}/authorization", headers=admin["headers"]).json()["data"]
|
|
response = client.put(f"/api/v1/system/users/{normal['id']}/authorization", headers=admin["headers"], json={
|
|
"role_codes": [role["role_code"]], "expected_revision": before["revision"], "reason": "分配新角色",
|
|
})
|
|
assert response.status_code == 200, response.text
|
|
info = client.get("/api/v1/auth/me", headers=normal["headers"]).json()["data"]
|
|
assert set(info["permissions"]) == set(codes)
|
|
|
|
@pytest.mark.parametrize("codes", [
|
|
["unknown:permission"], ["base:personnel:view", "base:personnel:view"],
|
|
["system:users:view"], ["base:personnel:regions"],
|
|
])
|
|
def test_非法初始权限不创建角色(self, client, admin, codes):
|
|
code = _new_code()
|
|
response = client.post(_ROLES, headers=admin["headers"], json={
|
|
"role_code": code, "role_name": "不得部分创建", "permission_codes": codes, "reason": "校验授权边界",
|
|
})
|
|
assert response.status_code == 400
|
|
result = client.get(_ROLES, headers=admin["headers"], params={"keyword": code}).json()["data"]
|
|
assert result["total"] == 0
|
|
|
|
def test_权限落库失败时角色和版本一起回滚(self, client, admin):
|
|
from sqlalchemy import event
|
|
|
|
from app.models import RolePermission
|
|
|
|
def fail_insert(*_):
|
|
raise RuntimeError("隔离测试模拟初始授权落库失败")
|
|
|
|
code = _new_code()
|
|
before = client.get("/api/v1/auth/me", headers=admin["headers"]).json()["data"]["authorization_revision"]
|
|
event.listen(RolePermission, "before_insert", fail_insert)
|
|
try:
|
|
with pytest.raises(RuntimeError, match="隔离测试模拟初始授权落库失败"):
|
|
client.post(_ROLES, headers=admin["headers"], json={
|
|
"role_code": code, "role_name": "回滚验证", "permission_codes": ["base:personnel:view"], "reason": "验证事务",
|
|
})
|
|
finally:
|
|
event.remove(RolePermission, "before_insert", fail_insert)
|
|
result = client.get(_ROLES, headers=admin["headers"], params={"keyword": code}).json()["data"]
|
|
assert result["total"] == 0
|
|
after = client.get("/api/v1/auth/me", headers=admin["headers"]).json()["data"]["authorization_revision"]
|
|
assert after == before
|
|
|
|
|
|
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
|