76ded6d5ec
- 月结锁定拦截补全：原仅新增回款一处，现覆盖回款编辑（原月+新月，防跨月 搬数）/删除、发货新增（按发货日）/编辑（原月+新月）/删除、核算与重算 （已审定月份金额只走调整单）；拦截先于「被引用/已核算」判定 - 台账 settlement_id 回填落地：generate 在 replace_items 后回填纳入行、 重生成前先解绑（clear_ledger_settlement_id）——不回填/不解绑会让 「未入结」判据失效：重生成漏算整月或把已固化行反复计入 - 经理聚合行落库：原只进内存参与 items 汇总，台账表永远无 calc_type=2 行、报表下钻落空；重生成先作废旧聚合行（deactivate_month_aggregates）， _confirm_scope 限定计算类型避免聚合行混入单笔池 - 月结取参锚定所属月月末（_month_end）：原用 date.today()，补生成历史 月份时分成/月工资口径随生成日漂移；聚合行 receipt_date 同步为月末 - generate/submit 出参固化前置到 commit 之前（commit 后 SELECT 失败会把 已生成的月结记成失败审计，与库中事实相反） - V9 迁移：business_secondary_allocations 唯一键条件式补齐（幂等裸 SQL， 临时库验证「无键→补」「有键→跳过」两分支），修复旧 V6 升级库拿不到 V6 就地加入的约束；V8 迁移 revision 声明补类型注解（原格式漂移使 schema 同步测试的 head 推断静默漏检 V8）；init-dev 版本号推进至 V9 - V7 留存 SQL 约束/索引名单数改复数（与 alembic op.f 与 init-dev 对齐， 消除 SHOW CREATE TABLE 双轨漂移） - 测试：显式控制审批开关（不依赖 V8 翻转后的种子默认值）；新增回归 8 例 （回填落库/聚合不重复/月末取参/月末纯函数 5 例/锁定拦截回款与发货与 核算）；全量 773 passed Co-Authored-By: Claude Code <noreply@anthropic.com>
271 lines
11 KiB
Python
271 lines
11 KiB
Python
"""二次分配方案录入（业务创建/编辑同事务）集成测试。
|
||
|
||
守的是（计划 T3 验收）：
|
||
- 创建带方案 → 详情回显含 personnel_name 冗余与比例
|
||
- 编辑全量替换（空数组=清空）
|
||
- 离职/不存在人员 → 400「分摊参与人员不存在或已离职」
|
||
- 终态业务禁改方案（与编辑终态门槛同一口径）
|
||
- 合计不强制 100%：hint 展示合计与余额（余额归属待澄清 #2）
|
||
- 删除保护：被方案引用的人员 RESTRICT 禁删（400 而非静默剥离）
|
||
- 人员删除后方案回显仍有名（personnel_name 固化的意义）
|
||
|
||
人员造数走 /api/v1/base/personnel（种子营销员岗位）；离职用状态端点。
|
||
"""
|
||
|
||
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"
|
||
_PERSONNEL = "/api/v1/base/personnel"
|
||
|
||
|
||
def _person(client: TestClient, headers: dict, **overrides: object) -> dict:
|
||
role_id = overrides.pop("business_role_id", None)
|
||
body = {
|
||
"personnel_name": f"t_分摊_{uuid.uuid4().hex[:8]}",
|
||
"business_role_id": role_id or _any_role_id(),
|
||
"salary_standard": "3000.00",
|
||
**overrides,
|
||
}
|
||
resp = client.post(_PERSONNEL, json=body, headers=headers)
|
||
assert resp.status_code == 200, resp.text
|
||
return resp.json()["data"]
|
||
|
||
|
||
def _any_role_id() -> int:
|
||
"""种子营销员岗位 ID（内置启用）。"""
|
||
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_allocations(client: TestClient, headers: dict, allocations: list[dict]) -> dict:
|
||
cust = customer(client, headers)
|
||
resp = client.post(
|
||
_BUSINESSES,
|
||
json=business_payload(cust["id"], secondary_allocations=allocations),
|
||
headers=headers,
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
return resp.json()["data"]
|
||
|
||
|
||
def _set_switch(client: TestClient, headers: dict, value: str) -> None:
|
||
"""显式设置二次分配审批开关（V8 起默认开启，用例不得依赖种子默认值）。"""
|
||
resp = client.post(
|
||
"/api/v1/system/params/changes",
|
||
json={
|
||
"change_type": 1,
|
||
"remark": f"用例设置二次分配审批开关={value}",
|
||
"items": [
|
||
{
|
||
"param_code": "approval.secondary_allocation.enabled",
|
||
"dimension_value": "",
|
||
"value": value,
|
||
}
|
||
],
|
||
},
|
||
headers=headers,
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
|
||
|
||
def _detail(client: TestClient, headers: dict, business_id: int) -> dict:
|
||
resp = client.get(f"{_BUSINESSES}/{business_id}", headers=headers)
|
||
assert resp.status_code == 200, resp.text
|
||
return resp.json()["data"]
|
||
|
||
|
||
class TestCreateAndEcho:
|
||
"""创建带方案与详情回显。"""
|
||
|
||
def test_带方案创建_详情回显姓名与比例(self, client: TestClient, admin: dict) -> None:
|
||
p1 = _person(client, admin["headers"])
|
||
p2 = _person(client, admin["headers"])
|
||
data = _create_with_allocations(
|
||
client,
|
||
admin["headers"],
|
||
[
|
||
{"personnel_id": p1["id"], "ratio": "30.00"},
|
||
{"personnel_id": p2["id"], "ratio": "20.00"},
|
||
],
|
||
)
|
||
detail = _detail(client, admin["headers"], data["id"])
|
||
assert len(detail["secondary_allocations"]) == 2
|
||
names = {a["personnel_name"] for a in detail["secondary_allocations"]}
|
||
assert p1["personnel_name"] in names and p2["personnel_name"] in names
|
||
ratios = sorted(a["ratio"] for a in detail["secondary_allocations"])
|
||
assert ratios == ["20.00", "30.00"]
|
||
# 合计 50%，余额 50% —— 只提示不落库（待澄清 #2）
|
||
assert "50.00" in (detail["allocation_hint"] or "")
|
||
assert "余额" in (detail["allocation_hint"] or "")
|
||
|
||
def test_不建方案_详情为空且无hint(self, client: TestClient, admin: dict) -> None:
|
||
data = _create_with_allocations(client, admin["headers"], [])
|
||
detail = _detail(client, admin["headers"], data["id"])
|
||
assert detail["secondary_allocations"] == []
|
||
assert detail["allocation_hint"] is None
|
||
|
||
def test_合计满额_hint只报合计(self, client: TestClient, admin: dict) -> None:
|
||
p1 = _person(client, admin["headers"])
|
||
p2 = _person(client, admin["headers"])
|
||
data = _create_with_allocations(
|
||
client,
|
||
admin["headers"],
|
||
[
|
||
{"personnel_id": p1["id"], "ratio": "60"},
|
||
{"personnel_id": p2["id"], "ratio": "40.00"},
|
||
],
|
||
)
|
||
hint = _detail(client, admin["headers"], data["id"])["allocation_hint"]
|
||
assert hint is not None and "余额" not in hint
|
||
|
||
|
||
class TestGuards:
|
||
"""校验链：在职/终态/去重已由 schema 层覆盖，此处打服务层口径。"""
|
||
|
||
def test_离职人员参与被拒(self, client: TestClient, admin: dict) -> None:
|
||
p = _person(client, admin["headers"])
|
||
resp = client.put(
|
||
f"{_PERSONNEL}/{p['id']}/status", json={"status": 0}, headers=admin["headers"]
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
|
||
cust = customer(client, admin["headers"])
|
||
resp = client.post(
|
||
_BUSINESSES,
|
||
json=business_payload(
|
||
cust["id"],
|
||
secondary_allocations=[{"personnel_id": p["id"], "ratio": "50.00"}],
|
||
),
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 400
|
||
assert "分摊参与" in resp.json()["message"]
|
||
|
||
def test_离职历史参与人_编辑其余字段放行(self, client: TestClient, admin: dict) -> None:
|
||
"""历史参与人离职后：编辑业务（保留该行）不应被 400 卡死——
|
||
personnel_name 冗余固化的设计本就允许离职后回显；仅新增参与人
|
||
才做在职校验。"""
|
||
p = _person(client, admin["headers"])
|
||
data = _create_with_allocations(
|
||
client, admin["headers"], [{"personnel_id": p["id"], "ratio": "50.00"}]
|
||
)
|
||
resp = client.put(
|
||
f"{_PERSONNEL}/{p['id']}/status", json={"status": 0}, headers=admin["headers"]
|
||
)
|
||
assert resp.status_code == 200
|
||
|
||
detail = _detail(client, admin["headers"], data["id"])
|
||
payload = business_payload(
|
||
detail["customer_id"],
|
||
secondary_allocations=[{"personnel_id": p["id"], "ratio": "50.00"}],
|
||
)
|
||
resp = client.put(f"{_BUSINESSES}/{data['id']}", json=payload, headers=admin["headers"])
|
||
assert resp.status_code == 200, resp.text
|
||
after = _detail(client, admin["headers"], data["id"])
|
||
assert after["secondary_allocations"][0]["personnel_name"] == p["personnel_name"]
|
||
|
||
# 新增一名离职人员仍要拦（新增才做在职校验）
|
||
p2 = _person(client, admin["headers"])
|
||
client.put(f"{_PERSONNEL}/{p2['id']}/status", json={"status": 0}, headers=admin["headers"])
|
||
payload = business_payload(
|
||
detail["customer_id"],
|
||
secondary_allocations=[
|
||
{"personnel_id": p["id"], "ratio": "30.00"},
|
||
{"personnel_id": p2["id"], "ratio": "20.00"},
|
||
],
|
||
)
|
||
resp = client.put(f"{_BUSINESSES}/{data['id']}", json=payload, headers=admin["headers"])
|
||
assert resp.status_code == 400
|
||
assert "分摊参与" in resp.json()["message"]
|
||
|
||
def test_终态业务禁改方案(self, client: TestClient, admin: dict) -> None:
|
||
p = _person(client, admin["headers"])
|
||
data = _create_with_allocations(
|
||
client, admin["headers"], [{"personnel_id": p["id"], "ratio": "100"}]
|
||
)
|
||
resp = client.put(
|
||
f"{_BUSINESSES}/{data['id']}/status",
|
||
json={"status": 2},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
|
||
detail = _detail(client, admin["headers"], data["id"])
|
||
payload = business_payload(detail["customer_id"], secondary_allocations=[])
|
||
for key in ("business_no", "status", "owner_personnel_id", "total_sale_amount"):
|
||
payload.pop(key, None)
|
||
resp = client.put(f"{_BUSINESSES}/{data['id']}", json=payload, headers=admin["headers"])
|
||
assert resp.status_code == 400
|
||
assert "禁止编辑" in resp.json()["message"]
|
||
|
||
def test_编辑全量替换与清空(self, client: TestClient, admin: dict) -> None:
|
||
p1 = _person(client, admin["headers"])
|
||
p2 = _person(client, admin["headers"])
|
||
data = _create_with_allocations(
|
||
client, admin["headers"], [{"personnel_id": p1["id"], "ratio": "40.00"}]
|
||
)
|
||
|
||
detail = _detail(client, admin["headers"], data["id"])
|
||
payload = business_payload(
|
||
detail["customer_id"],
|
||
secondary_allocations=[{"personnel_id": p2["id"], "ratio": "70.00"}],
|
||
)
|
||
resp = client.put(f"{_BUSINESSES}/{data['id']}", json=payload, headers=admin["headers"])
|
||
assert resp.status_code == 200, resp.text
|
||
|
||
replaced = _detail(client, admin["headers"], data["id"])
|
||
assert len(replaced["secondary_allocations"]) == 1
|
||
assert replaced["secondary_allocations"][0]["personnel_id"] == p2["id"]
|
||
|
||
payload = business_payload(detail["customer_id"], secondary_allocations=[])
|
||
resp = client.put(f"{_BUSINESSES}/{data['id']}", json=payload, headers=admin["headers"])
|
||
assert resp.status_code == 200, resp.text
|
||
assert _detail(client, admin["headers"], data["id"])["secondary_allocations"] == []
|
||
|
||
|
||
class TestDeleteProtection:
|
||
"""RESTRICT 外键：被方案引用的人员禁删，方案随业务删除而清。"""
|
||
|
||
def test_被方案引用的人员禁删(self, client: TestClient, admin: dict) -> None:
|
||
p = _person(client, admin["headers"])
|
||
data = _create_with_allocations(
|
||
client, admin["headers"], [{"personnel_id": p["id"], "ratio": "50.00"}]
|
||
)
|
||
resp = client.delete(f"{_PERSONNEL}/{p['id']}", headers=admin["headers"])
|
||
assert resp.status_code == 400
|
||
# 方案仍在，回显有名（RESTRICT 保证未静默剥离）
|
||
detail = _detail(client, admin["headers"], data["id"])
|
||
assert detail["secondary_allocations"][0]["personnel_name"] == p["personnel_name"]
|
||
|
||
def test_删除业务后可删人员(self, client: TestClient, admin: dict) -> None:
|
||
# 显式关开关：开启时审批单在途、删业务被 400 拦截（产品行为，另有用例覆盖）
|
||
_set_switch(client, admin["headers"], "0")
|
||
p = _person(client, admin["headers"])
|
||
data = _create_with_allocations(
|
||
client, admin["headers"], [{"personnel_id": p["id"], "ratio": "50.00"}]
|
||
)
|
||
resp = client.delete(f"{_BUSINESSES}/{data['id']}", headers=admin["headers"])
|
||
assert resp.status_code == 200, resp.text # 方案 CASCADE 随业务删除
|
||
assert client.delete(f"{_PERSONNEL}/{p['id']}", headers=admin["headers"]).status_code == 200
|