313 lines
12 KiB
Python
313 lines
12 KiB
Python
|
"""审批流集成测试（二次分配方案为首个接入对象）。
|
|||
|
|
|
|||
|
|
守的是（计划 T4 验收矩阵）：
|
|||
|
|
- 开关关闭（种子默认 0）：提单即「自动通过」——status=3 且两级节点
|
|||
|
|
action=3 留痕（留痕不断链，关闭=直接生效的需求口径）
|
|||
|
|
- 开关开启：status=1 → 财务通过 → status=2 → 领导通过 → status=3
|
|||
|
|
- 驳回：整体驳回 status=4，意见留痕；驳回必填意见（空驳 400）
|
|||
|
|
- 节点准入：status=1 仅财务、status=2 仅领导；**admin 不代办**（403）；
|
|||
|
|
普通用户/营销后勤进不了模块（403）
|
|||
|
|
- 再编辑撤旧建新：在途单 status=5 + node=0/action=4 撤回留痕，新单新建
|
|||
|
|
- 方案为空不提单
|
|||
|
|
- 开关读一次：同一提交路径内参数变化不影响判定（种子级默认行为锁定）
|
|||
|
|
|
|||
|
|
角色用户经 make_user 造（finance/leader/marketing_ops 三新角色种子在库）。
|
|||
|
|
开关联动走参数变更批次（PUT /system/params 端到端，不直改库）。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
import uuid
|
|||
|
|
|
|||
|
|
import pytest
|
|||
|
|
from fastapi.testclient import TestClient
|
|||
|
|
|
|||
|
|
from tests.integration.factories import business_payload, customer
|
|||
|
|
|
|||
|
|
pytestmark = pytest.mark.integration
|
|||
|
|
|
|||
|
|
_BUSINESSES = "/api/v1/business/businesses"
|
|||
|
|
_APPROVALS = "/api/v1/system/approvals"
|
|||
|
|
_PARAMS = "/api/v1/system/params"
|
|||
|
|
_PERSONNEL = "/api/v1/base/personnel"
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _person(client: TestClient, headers: dict) -> dict:
|
|||
|
|
resp = client.post(
|
|||
|
|
_PERSONNEL,
|
|||
|
|
json={
|
|||
|
|
"personnel_name": f"t_审批_{uuid.uuid4().hex[:8]}",
|
|||
|
|
"business_role_id": _any_role_id(),
|
|||
|
|
"salary_standard": "3000.00",
|
|||
|
|
},
|
|||
|
|
headers=headers,
|
|||
|
|
)
|
|||
|
|
assert resp.status_code == 200, resp.text
|
|||
|
|
return resp.json()["data"]
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _any_role_id() -> int:
|
|||
|
|
from sqlalchemy import select
|
|||
|
|
|
|||
|
|
from app.core.constants import BUSINESS_ROLE_SALESPERSON
|
|||
|
|
from app.core.db import get_session_factory
|
|||
|
|
from app.models import BusinessRole
|
|||
|
|
|
|||
|
|
factory = get_session_factory()
|
|||
|
|
with factory() as session:
|
|||
|
|
return int(
|
|||
|
|
session.execute(
|
|||
|
|
select(BusinessRole.id).where(
|
|||
|
|
BusinessRole.role_code == BUSINESS_ROLE_SALESPERSON,
|
|||
|
|
BusinessRole.status == 1,
|
|||
|
|
)
|
|||
|
|
).scalar_one()
|
|||
|
|
)
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _create_with_plan(client: TestClient, headers: dict, personnel_id: int) -> dict:
|
|||
|
|
cust = customer(client, headers)
|
|||
|
|
resp = client.post(
|
|||
|
|
_BUSINESSES,
|
|||
|
|
json=business_payload(
|
|||
|
|
cust["id"],
|
|||
|
|
secondary_allocations=[{"personnel_id": personnel_id, "ratio": "50.00"}],
|
|||
|
|
),
|
|||
|
|
headers=headers,
|
|||
|
|
)
|
|||
|
|
assert resp.status_code == 200, resp.text
|
|||
|
|
return resp.json()["data"]
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _latest_order(client: TestClient, headers: dict, refer_no: str) -> dict:
|
|||
|
|
resp = client.get(_APPROVALS, params={"keyword": refer_no, "pageSize": 1}, headers=headers)
|
|||
|
|
assert resp.status_code == 200, resp.text
|
|||
|
|
rows = resp.json()["data"]["list"]
|
|||
|
|
assert rows, f"未找到审批单：{refer_no}"
|
|||
|
|
detail = client.get(f"{_APPROVALS}/{rows[0]['id']}", headers=headers)
|
|||
|
|
return detail.json()["data"]
|
|||
|
|
|
|||
|
|
|
|||
|
|
def _set_switch(client: TestClient, admin_headers: dict, value: str) -> None:
|
|||
|
|
"""经参数变更批次把二次分配审批开关置值（立即生效）。"""
|
|||
|
|
resp = client.post(
|
|||
|
|
f"{_PARAMS}/changes",
|
|||
|
|
json={
|
|||
|
|
"change_type": 1,
|
|||
|
|
"remark": f"测试切换审批开关={value}",
|
|||
|
|
"items": [
|
|||
|
|
{
|
|||
|
|
"param_code": "approval.secondary_allocation.enabled",
|
|||
|
|
"dimension_value": "",
|
|||
|
|
"value": value,
|
|||
|
|
}
|
|||
|
|
],
|
|||
|
|
},
|
|||
|
|
headers=admin_headers,
|
|||
|
|
)
|
|||
|
|
assert resp.status_code == 200, resp.text
|
|||
|
|
|
|||
|
|
|
|||
|
|
@pytest.fixture
|
|||
|
|
def finance(make_user, auth_headers) -> dict:
|
|||
|
|
"""财务角色账号（审批第二节点）。"""
|
|||
|
|
user = make_user(["finance"])
|
|||
|
|
user["headers"] = auth_headers(user)
|
|||
|
|
return user
|
|||
|
|
|
|||
|
|
|
|||
|
|
@pytest.fixture
|
|||
|
|
def leader(make_user, auth_headers) -> dict:
|
|||
|
|
"""领导角色账号（审批第三节点）。"""
|
|||
|
|
user = make_user(["leader"])
|
|||
|
|
user["headers"] = auth_headers(user)
|
|||
|
|
return user
|
|||
|
|
|
|||
|
|
|
|||
|
|
@pytest.fixture
|
|||
|
|
def switched_on(client: TestClient, admin: dict) -> None:
|
|||
|
|
"""把二次分配审批开关置 1（立即生效；用例结束由 conftest 恢复种子基线）。"""
|
|||
|
|
_set_switch(client, admin["headers"], "1")
|
|||
|
|
|
|||
|
|
|
|||
|
|
class TestSwitchOff:
|
|||
|
|
"""开关关闭（种子默认）：提单即自动通过。"""
|
|||
|
|
|
|||
|
|
def test_创建带方案_自动通过全节点留痕(
|
|||
|
|
self, client: TestClient, admin: dict, finance: dict
|
|||
|
|
) -> None:
|
|||
|
|
p = _person(client, admin["headers"])
|
|||
|
|
data = _create_with_plan(client, admin["headers"], p["id"])
|
|||
|
|
|
|||
|
|
order = _latest_order(client, finance["headers"], data["business_no"])
|
|||
|
|
assert order["status"] == 3
|
|||
|
|
assert order["order_no"].startswith("AP")
|
|||
|
|
auto_steps = [s for s in order["steps"] if s["action"] == 3]
|
|||
|
|
assert len(auto_steps) == 2
|
|||
|
|
assert {s["node"] for s in auto_steps} == {1, 2}
|
|||
|
|
assert all(s["operator_name"] == "系统" for s in auto_steps)
|
|||
|
|
|
|||
|
|
def test_方案为空不提单(self, client: TestClient, admin: dict, finance: dict) -> None:
|
|||
|
|
cust = customer(client, admin["headers"])
|
|||
|
|
resp = client.post(_BUSINESSES, json=business_payload(cust["id"]), headers=admin["headers"])
|
|||
|
|
assert resp.status_code == 200
|
|||
|
|
rows = client.get(
|
|||
|
|
_APPROVALS,
|
|||
|
|
params={"keyword": resp.json()["data"]["business_no"]},
|
|||
|
|
headers=finance["headers"],
|
|||
|
|
).json()["data"]["list"]
|
|||
|
|
assert rows == []
|
|||
|
|
|
|||
|
|
|
|||
|
|
class TestSwitchOn:
|
|||
|
|
"""开关开启：两节点人工流转。"""
|
|||
|
|
|
|||
|
|
@pytest.fixture
|
|||
|
|
def switched_on(self, client: TestClient, admin: dict) -> None:
|
|||
|
|
_set_switch(client, admin["headers"], "1")
|
|||
|
|
|
|||
|
|
def test_两节点流转至通过(
|
|||
|
|
self, client: TestClient, admin: dict, finance, leader, switched_on
|
|||
|
|
) -> None:
|
|||
|
|
|
|||
|
|
p = _person(client, admin["headers"])
|
|||
|
|
data = _create_with_plan(client, admin["headers"], p["id"])
|
|||
|
|
|
|||
|
|
order = _latest_order(client, finance["headers"], data["business_no"])
|
|||
|
|
assert order["status"] == 1
|
|||
|
|
|
|||
|
|
# 财务通过 → 待领导
|
|||
|
|
resp = client.post(
|
|||
|
|
f"{_APPROVALS}/{order['id']}/action",
|
|||
|
|
json={"action": 1, "opinion": "同意"},
|
|||
|
|
headers=finance["headers"],
|
|||
|
|
)
|
|||
|
|
assert resp.status_code == 200, resp.text
|
|||
|
|
assert resp.json()["data"]["status"] == 2
|
|||
|
|
|
|||
|
|
# 领导通过 → 已通过
|
|||
|
|
resp = client.post(
|
|||
|
|
f"{_APPROVALS}/{order['id']}/action",
|
|||
|
|
json={"action": 1},
|
|||
|
|
headers=leader["headers"],
|
|||
|
|
)
|
|||
|
|
assert resp.status_code == 200, resp.text
|
|||
|
|
assert resp.json()["data"]["status"] == 3
|
|||
|
|
manual = [s for s in resp.json()["data"]["steps"] if s["action"] == 1]
|
|||
|
|
assert {s["node"] for s in manual} == {1, 2}
|
|||
|
|
|
|||
|
|
def test_财务驳回_整体驳回且意见留痕(
|
|||
|
|
self, client: TestClient, admin: dict, finance, leader, switched_on
|
|||
|
|
) -> None:
|
|||
|
|
_ = leader
|
|||
|
|
p = _person(client, admin["headers"])
|
|||
|
|
data = _create_with_plan(client, admin["headers"], p["id"])
|
|||
|
|
order = _latest_order(client, finance["headers"], data["business_no"])
|
|||
|
|
|
|||
|
|
resp = client.post(
|
|||
|
|
f"{_APPROVALS}/{order['id']}/action",
|
|||
|
|
json={"action": 2, "opinion": "比例分配不合理"},
|
|||
|
|
headers=finance["headers"],
|
|||
|
|
)
|
|||
|
|
assert resp.status_code == 200, resp.text
|
|||
|
|
out = resp.json()["data"]
|
|||
|
|
assert out["status"] == 4
|
|||
|
|
rejected = [s for s in out["steps"] if s["action"] == 2]
|
|||
|
|
assert rejected and rejected[0]["opinion"] == "比例分配不合理"
|
|||
|
|
|
|||
|
|
def test_驳回缺意见被拒(
|
|||
|
|
self, client: TestClient, admin: dict, finance, leader, switched_on
|
|||
|
|
) -> None:
|
|||
|
|
p = _person(client, admin["headers"])
|
|||
|
|
data = _create_with_plan(client, admin["headers"], p["id"])
|
|||
|
|
order = _latest_order(client, finance["headers"], data["business_no"])
|
|||
|
|
|
|||
|
|
resp = client.post(
|
|||
|
|
f"{_APPROVALS}/{order['id']}/action",
|
|||
|
|
json={"action": 2},
|
|||
|
|
headers=finance["headers"],
|
|||
|
|
)
|
|||
|
|
assert resp.status_code == 400
|
|||
|
|
|
|||
|
|
def test_节点准入_领导不能审财务节点_admin不代办(
|
|||
|
|
self, client: TestClient, admin: dict, finance, leader, switched_on
|
|||
|
|
) -> None:
|
|||
|
|
p = _person(client, admin["headers"])
|
|||
|
|
data = _create_with_plan(client, admin["headers"], p["id"])
|
|||
|
|
order = _latest_order(client, leader["headers"], data["business_no"])
|
|||
|
|
|
|||
|
|
# status=1 是财务节点：领导不行，admin 也不代办
|
|||
|
|
for headers in (leader["headers"], admin["headers"]):
|
|||
|
|
resp = client.post(
|
|||
|
|
f"{_APPROVALS}/{order['id']}/action",
|
|||
|
|
json={"action": 1},
|
|||
|
|
headers=headers,
|
|||
|
|
)
|
|||
|
|
assert resp.status_code == 403
|
|||
|
|
|
|||
|
|
def test_重复操作已通过单被拒(
|
|||
|
|
self, client: TestClient, admin: dict, finance, leader, switched_on
|
|||
|
|
) -> None:
|
|||
|
|
|
|||
|
|
p = _person(client, admin["headers"])
|
|||
|
|
data = _create_with_plan(client, admin["headers"], p["id"])
|
|||
|
|
order = _latest_order(client, finance["headers"], data["business_no"])
|
|||
|
|
client.post(
|
|||
|
|
f"{_APPROVALS}/{order['id']}/action", json={"action": 1}, headers=finance["headers"]
|
|||
|
|
)
|
|||
|
|
client.post(
|
|||
|
|
f"{_APPROVALS}/{order['id']}/action", json={"action": 1}, headers=leader["headers"]
|
|||
|
|
)
|
|||
|
|
resp = client.post(
|
|||
|
|
f"{_APPROVALS}/{order['id']}/action", json={"action": 1}, headers=finance["headers"]
|
|||
|
|
)
|
|||
|
|
assert resp.status_code == 400
|
|||
|
|
|
|||
|
|
|
|||
|
|
class TestResubmit:
|
|||
|
|
"""再编辑撤旧建新。"""
|
|||
|
|
|
|||
|
|
def test_在途单撤回并新建(
|
|||
|
|
self, client: TestClient, admin: dict, finance, leader, switched_on
|
|||
|
|
) -> None:
|
|||
|
|
p = _person(client, admin["headers"])
|
|||
|
|
data = _create_with_plan(client, admin["headers"], p["id"])
|
|||
|
|
first = _latest_order(client, finance["headers"], data["business_no"])
|
|||
|
|
assert first["status"] == 1
|
|||
|
|
|
|||
|
|
# 再编辑（仍带方案）→ 旧在途单撤回、新单待审
|
|||
|
|
detail = client.get(f"{_BUSINESSES}/{data['id']}", headers=admin["headers"]).json()["data"]
|
|||
|
|
payload = business_payload(
|
|||
|
|
detail["customer_id"],
|
|||
|
|
secondary_allocations=[{"personnel_id": p["id"], "ratio": "80.00"}],
|
|||
|
|
)
|
|||
|
|
resp = client.put(f"{_BUSINESSES}/{data['id']}", json=payload, headers=admin["headers"])
|
|||
|
|
assert resp.status_code == 200, resp.text
|
|||
|
|
|
|||
|
|
rows = client.get(
|
|||
|
|
_APPROVALS,
|
|||
|
|
params={"keyword": data["business_no"], "pageSize": 10},
|
|||
|
|
headers=finance["headers"],
|
|||
|
|
).json()["data"]["list"]
|
|||
|
|
statuses = {row["status"] for row in rows}
|
|||
|
|
assert 5 in statuses and 1 in statuses, f"未见撤回+新建：{statuses}"
|
|||
|
|
old_detail = client.get(f"{_APPROVALS}/{first['id']}", headers=finance["headers"]).json()
|
|||
|
|
withdraw_steps = [s for s in old_detail["data"]["steps"] if s["action"] == 4]
|
|||
|
|
assert withdraw_steps and withdraw_steps[0]["node"] == 0
|
|||
|
|
|
|||
|
|
|
|||
|
|
class TestModuleAccess:
|
|||
|
|
"""模块准入：普通用户与营销后勤不可见（权限清单待澄清 #6）。"""
|
|||
|
|
|
|||
|
|
def test_普通用户403(self, client: TestClient, normal: dict) -> None:
|
|||
|
|
assert client.get(_APPROVALS, headers=normal["headers"]).status_code == 403
|
|||
|
|
|
|||
|
|
def test_营销后勤403_暂不可见(self, client: TestClient, make_user, auth_headers) -> None:
|
|||
|
|
ops = make_user(["marketing_ops"])
|
|||
|
|
ops["headers"] = auth_headers(ops)
|
|||
|
|
assert client.get(_APPROVALS, headers=ops["headers"]).status_code == 403
|
|||
|
|
|
|||
|
|
def test_财务与领导可见(self, client: TestClient, make_user, auth_headers) -> None:
|
|||
|
|
for code in ("finance", "leader"):
|
|||
|
|
user = make_user([code])
|
|||
|
|
user["headers"] = auth_headers(user)
|
|||
|
|
assert client.get(_APPROVALS, headers=user["headers"]).status_code == 200
|