156 lines
5.9 KiB
Python
156 lines
5.9 KiB
Python
"""岗位角色管理接口集成测试（需要 MySQL 与 Redis）。
|
||
|
||
守的是岗位管理的保护规则与模板绑定 —— 内置岗位被删停会让参数维度值
|
||
失效、核算模板断锚；模板编码不校验会让核算引擎取不到公式。
|
||
|
||
运行前置：``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/business-roles"
|
||
_TEMPLATES = "/api/v1/system/business-roles/meta/templates"
|
||
|
||
|
||
def _new_code() -> str:
|
||
return f"t_br_{uuid.uuid4().hex[:6]}"
|
||
|
||
|
||
class TestCrudFlow:
|
||
"""正向链路：建 → 查（列表 + 模板下拉）→ 改 → 删。"""
|
||
|
||
def test_新增自定义岗位并联动列表(self, client: TestClient, admin: dict) -> None:
|
||
code = _new_code()
|
||
resp = client.post(
|
||
_ROLES,
|
||
json={
|
||
"role_code": code,
|
||
"role_name": "集成测试岗位",
|
||
"calc_template": "TPL_SALESPERSON_DECAY",
|
||
"description": "集成测试",
|
||
"sort_order": 50,
|
||
},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
body = resp.json()["data"]
|
||
assert body["role_code"] == code
|
||
assert body["is_builtin"] == 0
|
||
assert body["status"] == 1
|
||
|
||
listed = client.get(_ROLES, params={"keyword": code}, headers=admin["headers"])
|
||
assert any(r["role_code"] == code for r in listed.json()["data"]["list"])
|
||
|
||
# 清理
|
||
assert client.delete(f"{_ROLES}/{body['id']}", headers=admin["headers"]).status_code == 200
|
||
|
||
def test_重复编码拒绝(self, client: TestClient, admin: dict) -> None:
|
||
code = _new_code()
|
||
payload = {
|
||
"role_code": code,
|
||
"role_name": "首个",
|
||
"calc_template": "TPL_SALESPERSON_DECAY",
|
||
}
|
||
assert client.post(_ROLES, json=payload, headers=admin["headers"]).status_code == 200
|
||
dup = client.post(_ROLES, json=payload, headers=admin["headers"])
|
||
assert dup.status_code == 400
|
||
assert "已存在" in dup.json()["message"]
|
||
client.delete(
|
||
f"{_ROLES}/{client.get(_ROLES, params={'keyword': code}, headers=admin['headers']).json()['data']['list'][0]['id']}",
|
||
headers=admin["headers"],
|
||
)
|
||
|
||
def test_模板编码非法拒绝(self, client: TestClient, admin: dict) -> None:
|
||
resp = client.post(
|
||
_ROLES,
|
||
json={
|
||
"role_code": _new_code(),
|
||
"role_name": "坏模板岗位",
|
||
"calc_template": "TPL_NOT_EXIST",
|
||
},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 400
|
||
|
||
def test_编辑岗位换模板生效(self, client: TestClient, admin: dict) -> None:
|
||
code = _new_code()
|
||
created = client.post(
|
||
_ROLES,
|
||
json={
|
||
"role_code": code,
|
||
"role_name": "待改岗位",
|
||
"calc_template": "TPL_SALESPERSON_DECAY",
|
||
},
|
||
headers=admin["headers"],
|
||
).json()["data"]
|
||
updated = client.put(
|
||
f"{_ROLES}/{created['id']}",
|
||
json={
|
||
"role_name": "已改岗位",
|
||
"calc_template": "TPL_ENTERPRISE_FIXED",
|
||
"sort_order": 7,
|
||
},
|
||
headers=admin["headers"],
|
||
)
|
||
assert updated.status_code == 200
|
||
body = updated.json()["data"]
|
||
assert body["role_name"] == "已改岗位"
|
||
assert body["calc_template"] == "TPL_ENTERPRISE_FIXED"
|
||
assert body["role_code"] == code # 编码不变
|
||
client.delete(f"{_ROLES}/{created['id']}", headers=admin["headers"])
|
||
|
||
def test_模板下拉含五个模板且声明参数已注册(self, client: TestClient, admin: dict) -> None:
|
||
options = client.get(_TEMPLATES, headers=admin["headers"]).json()["data"]
|
||
assert len(options) == 5
|
||
for tpl in options:
|
||
assert tpl["params"], f"{tpl['code']} 未声明参数"
|
||
|
||
|
||
class TestBuiltinProtection:
|
||
"""内置岗位（种子 3 个）禁删禁停 —— 参数维度与模板按编码锚定。"""
|
||
|
||
def test_停用内置岗位_403(self, client: TestClient, admin: dict) -> None:
|
||
rows = client.get(
|
||
_ROLES, params={"keyword": "salesperson"}, headers=admin["headers"]
|
||
).json()["data"]["list"]
|
||
target = next(r for r in rows if r["role_code"] == "salesperson" and r["is_builtin"] == 1)
|
||
resp = client.put(
|
||
f"{_ROLES}/{target['id']}/status", json={"status": 0}, headers=admin["headers"]
|
||
)
|
||
assert resp.status_code == 403
|
||
|
||
def test_删除内置岗位_403(self, client: TestClient, admin: dict) -> None:
|
||
rows = client.get(
|
||
_ROLES, params={"keyword": "sales_manager"}, headers=admin["headers"]
|
||
).json()["data"]["list"]
|
||
target = next(r for r in rows if r["role_code"] == "sales_manager")
|
||
resp = client.delete(f"{_ROLES}/{target['id']}", headers=admin["headers"])
|
||
assert resp.status_code == 403
|
||
|
||
def test_种子内置岗位三个且不可删标记正确(self, client: TestClient, admin: dict) -> None:
|
||
rows = client.get(_ROLES, params={"pageSize": 50}, headers=admin["headers"]).json()["data"][
|
||
"list"
|
||
]
|
||
builtins = [r for r in rows if r["is_builtin"] == 1]
|
||
assert {r["role_code"] for r in builtins} == {
|
||
"salesperson",
|
||
"sales_manager",
|
||
"info_officer",
|
||
}
|
||
|
||
|
||
class TestPermission:
|
||
"""权限：普通用户全部 403（当前默认管理员专用）。"""
|
||
|
||
def test_普通用户访问岗位列表_403(self, client: TestClient, make_user, auth_headers) -> None:
|
||
user = make_user([ROLE_USER])
|
||
headers = auth_headers(user)
|
||
assert client.get(_ROLES, headers=headers).status_code == 403
|