238a38c356
- 金标准金额断言锚定 DB 事实源（NUMERIC(14,2) 读回恒两位）；行级明细 折算两位固化；全量回归 756 passed（701+核算 13/月结 6/公式 12 等） - 目录断言 18→19（V7 调整单开关）；tests/README 增 M3-core 节与运行记录； frontend/README 增核算与月结两页 Co-Authored-By: Claude Code <noreply@anthropic.com>
455 lines
18 KiB
Python
455 lines
18 KiB
Python
"""计提参数管理接口集成测试（需要 MySQL 与 Redis）。
|
||
|
||
守的是参数体系的口径与留痕 —— 时点化取值错一天、同日变更取错条目、
|
||
旧值丢失、未来值泄漏到当前快照，任何一条都会直接算错钱。
|
||
|
||
关键复现（全部来自 2026-10-12 业务裁决）：
|
||
- 企业户 2‰（2026-02-17 起）→ 3‰（2026-08-13 起）的按日回溯
|
||
- 业务员 27 天 → 97/10000，折算 15000，绩效 145.5，可付 52.38（预览金标准）
|
||
- 同日多次变更取最新一条；撤回未生效批次后解析回退
|
||
|
||
运行前置：``docker compose -f docker/docker-compose.dev.yml up -d``
|
||
"""
|
||
|
||
import uuid
|
||
from datetime import date, timedelta
|
||
from fractions import Fraction
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy import delete
|
||
|
||
from app.core.constants import ROLE_USER
|
||
from app.core.db import get_session_factory
|
||
from app.models import ParamChangeBatch, ParamValue
|
||
|
||
pytestmark = pytest.mark.integration
|
||
|
||
_PARAMS = "/api/v1/system/params"
|
||
|
||
# 种子基线（与 V3 迁移一致）：断言的「预期值」均以此为准
|
||
_SEED_PREPAY = "10"
|
||
_SEED_START_RATIO = "60"
|
||
|
||
|
||
@pytest.fixture
|
||
def track_batches():
|
||
"""记录测试创建的批次 id，用例结束整批清理（条目 + 批次头）。"""
|
||
created: list[int] = []
|
||
|
||
def _track(batch_id: int) -> None:
|
||
created.append(batch_id)
|
||
|
||
yield _track
|
||
|
||
factory = get_session_factory()
|
||
with factory() as session:
|
||
if created:
|
||
session.execute(delete(ParamValue).where(ParamValue.batch_id.in_(created)))
|
||
session.execute(delete(ParamChangeBatch).where(ParamChangeBatch.id.in_(created)))
|
||
session.commit()
|
||
|
||
|
||
@pytest.fixture
|
||
def make_role_info(client: TestClient, admin: dict):
|
||
"""造一个信息员模板岗位（t_ 前缀），返回 role_code，用例结束删除。
|
||
|
||
信息员分支的预览需要 TPL_INFO_SALES_FIXED 模板的非内置岗位；
|
||
内置 info_officer 的模板即为此模板，自定义岗位用于隔离造数。
|
||
"""
|
||
code = f"t_info_{uuid.uuid4().hex[:6]}"
|
||
resp = client.post(
|
||
"/api/v1/system/business-roles",
|
||
json={
|
||
"role_code": code,
|
||
"role_name": "集成信息员",
|
||
"calc_template": "TPL_INFO_SALES_FIXED",
|
||
},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
role_id = resp.json()["data"]["id"]
|
||
try:
|
||
yield code
|
||
finally:
|
||
client.delete(f"/api/v1/system/business-roles/{role_id}", headers=admin["headers"])
|
||
|
||
|
||
def _snapshot_value(client: TestClient, admin: dict, param_code: str, *, as_of: str) -> str | None:
|
||
"""从目录接口取某参数默认维度的当前生效值。"""
|
||
items = client.get(_PARAMS, params={"as_of": as_of}, headers=admin["headers"]).json()["data"]
|
||
match = next(item for item in items if item["param_code"] == param_code)
|
||
dims = [v for v in match["values"] if v["dimension_value"] == ""]
|
||
return dims[0]["value_text"] if dims else None
|
||
|
||
|
||
class TestCatalog:
|
||
"""参数目录：种子完整性与时点化回溯。"""
|
||
|
||
def test_目录共19项且覆盖八个分组(self, client: TestClient, admin: dict) -> None:
|
||
"""V7 起含调整单审批开关（16 计提项 + 3 BOOL 开关）。"""
|
||
items = client.get(_PARAMS, headers=admin["headers"]).json()["data"]
|
||
assert len(items) == 19
|
||
groups = {item["param_group"] for item in items}
|
||
assert groups == {
|
||
"业务员个人户",
|
||
"经理分成",
|
||
"企业户",
|
||
"信息员",
|
||
"支付环节",
|
||
"产品阈值",
|
||
"月工资口径",
|
||
"审批开关",
|
||
}
|
||
|
||
def test_企业户比例按到账日回溯_2千分3千分双种子(self, client: TestClient, admin: dict) -> None:
|
||
before = _snapshot_value(client, admin, "enterprise.rate", as_of="2026-08-12")
|
||
after = _snapshot_value(client, admin, "enterprise.rate", as_of="2026-08-13")
|
||
assert (before, after) == ("2", "3")
|
||
|
||
def test_业务员斜率为精确分数_非截断小数(self, client: TestClient, admin: dict) -> None:
|
||
items = client.get(_PARAMS, headers=admin["headers"]).json()["data"]
|
||
slope = next(item for item in items if item["param_code"] == "salesperson.rate.slope")
|
||
values = [v["value_text"] for v in slope["values"]]
|
||
assert values == ["4/360"] # 分数式原样存储，禁 0.011111 截断
|
||
|
||
|
||
class TestChangeLifecycle:
|
||
"""变更生命周期：提交 → 生效 → 同日取最新 → 撤回回退。"""
|
||
|
||
def test_立即生效_快照变化且历史含旧值(
|
||
self, client: TestClient, admin: dict, track_batches
|
||
) -> None:
|
||
resp = client.post(
|
||
f"{_PARAMS}/changes",
|
||
json={
|
||
"change_type": 1,
|
||
"remark": "集成-立即生效",
|
||
"items": [{"param_code": "prepay.rate", "dimension_value": "", "value": "12"}],
|
||
},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
batch = resp.json()["data"]
|
||
track_batches(batch["id"])
|
||
assert batch["item_count"] == 1
|
||
assert batch["version_no"].startswith("P")
|
||
|
||
value = _snapshot_value(client, admin, "prepay.rate", as_of=str(date.today()))
|
||
assert value == "12"
|
||
|
||
history = client.get(
|
||
f"{_PARAMS}/prepay.rate/history",
|
||
params={"pageNum": 1, "pageSize": 5},
|
||
headers=admin["headers"],
|
||
).json()["data"]
|
||
assert history["total"] >= 2 # 种子 1 条 + 本次 1 条
|
||
assert history["list"][0]["old_value_text"] == _SEED_PREPAY
|
||
assert history["list"][0]["value_text"] == "12"
|
||
|
||
def test_未来生效_当前快照不泄漏_到期后生效(
|
||
self, client: TestClient, admin: dict, track_batches
|
||
) -> None:
|
||
future = date.today() + timedelta(days=30)
|
||
resp = client.post(
|
||
f"{_PARAMS}/changes",
|
||
json={
|
||
"change_type": 2,
|
||
"effective_date": future.isoformat(),
|
||
"remark": "集成-未来生效",
|
||
"items": [
|
||
{"param_code": "payment.start.ratio", "dimension_value": "", "value": "65"}
|
||
],
|
||
},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 200
|
||
batch = resp.json()["data"]
|
||
track_batches(batch["id"])
|
||
|
||
today_value = _snapshot_value(client, admin, "payment.start.ratio", as_of=str(date.today()))
|
||
assert today_value == _SEED_START_RATIO # 未来值不泄漏
|
||
future_value = _snapshot_value(
|
||
client, admin, "payment.start.ratio", as_of=future.isoformat()
|
||
)
|
||
assert future_value == "65" # 到期即生效
|
||
|
||
def test_同日多次变更_取最新一条(self, client: TestClient, admin: dict, track_batches) -> None:
|
||
for value in ("12", "11"):
|
||
resp = client.post(
|
||
f"{_PARAMS}/changes",
|
||
json={
|
||
"change_type": 1,
|
||
"items": [{"param_code": "prepay.rate", "dimension_value": "", "value": value}],
|
||
},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 200
|
||
track_batches(resp.json()["data"]["id"])
|
||
value = _snapshot_value(client, admin, "prepay.rate", as_of=str(date.today()))
|
||
assert value == "11"
|
||
|
||
def test_撤回未生效批次_解析回退(self, client: TestClient, admin: dict, track_batches) -> None:
|
||
future = date.today() + timedelta(days=30)
|
||
batch = client.post(
|
||
f"{_PARAMS}/changes",
|
||
json={
|
||
"change_type": 2,
|
||
"effective_date": future.isoformat(),
|
||
"items": [{"param_code": "barter.discount", "dimension_value": "", "value": "85"}],
|
||
},
|
||
headers=admin["headers"],
|
||
).json()["data"]
|
||
track_batches(batch["id"])
|
||
|
||
future_value = _snapshot_value(client, admin, "barter.discount", as_of=future.isoformat())
|
||
assert future_value == "85"
|
||
|
||
assert (
|
||
client.delete(f"{_PARAMS}/changes/{batch['id']}", headers=admin["headers"]).status_code
|
||
== 200
|
||
)
|
||
rolled_back = _snapshot_value(client, admin, "barter.discount", as_of=future.isoformat())
|
||
assert rolled_back == _SEED_BARTER
|
||
|
||
def test_撤回已生效批次_400(self, client: TestClient, admin: dict, track_batches) -> None:
|
||
batch = client.post(
|
||
f"{_PARAMS}/changes",
|
||
json={
|
||
"change_type": 1,
|
||
"items": [{"param_code": "prepay.rate", "dimension_value": "", "value": "12"}],
|
||
},
|
||
headers=admin["headers"],
|
||
).json()["data"]
|
||
track_batches(batch["id"])
|
||
resp = client.delete(f"{_PARAMS}/changes/{batch['id']}", headers=admin["headers"])
|
||
assert resp.status_code == 400
|
||
assert "已生效" in resp.json()["message"]
|
||
|
||
|
||
_SEED_BARTER = "80"
|
||
|
||
|
||
class TestValidation:
|
||
"""值校验：畸形值在写入前拒绝（审查重点 1）。"""
|
||
|
||
def test_超取值上限拒绝(self, client: TestClient, admin: dict) -> None:
|
||
resp = client.post(
|
||
f"{_PARAMS}/changes",
|
||
json={
|
||
"change_type": 1,
|
||
"items": [{"param_code": "barter.discount", "dimension_value": "", "value": "120"}],
|
||
},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 400
|
||
|
||
def test_枚举越界拒绝(self, client: TestClient, admin: dict) -> None:
|
||
resp = client.post(
|
||
f"{_PARAMS}/changes",
|
||
json={
|
||
"change_type": 1,
|
||
"items": [
|
||
{"param_code": "payment.weight.mode", "dimension_value": "", "value": "PARTIAL"}
|
||
],
|
||
},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 400
|
||
|
||
def test_除零分数拒绝(self, client: TestClient, admin: dict) -> None:
|
||
resp = client.post(
|
||
f"{_PARAMS}/changes",
|
||
json={
|
||
"change_type": 1,
|
||
"items": [{"param_code": "barter.discount", "dimension_value": "", "value": "1/0"}],
|
||
},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 400
|
||
|
||
def test_指定日期早于今日拒绝(self, client: TestClient, admin: dict) -> None:
|
||
yesterday = date.today() - timedelta(days=1)
|
||
resp = client.post(
|
||
f"{_PARAMS}/changes",
|
||
json={
|
||
"change_type": 2,
|
||
"effective_date": yesterday.isoformat(),
|
||
"items": [{"param_code": "prepay.rate", "dimension_value": "", "value": "9"}],
|
||
},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code == 400
|
||
|
||
|
||
class TestPreview:
|
||
"""计算预览：业务裁决的「金标准」数值精确复现。"""
|
||
|
||
def test_业务员27天全链路精确复现(self, client: TestClient, admin: dict) -> None:
|
||
payload = {
|
||
"business_role": "salesperson",
|
||
"customer_type": "personal",
|
||
"product_category": "municipal_water",
|
||
"ship_date": "2025-12-15",
|
||
"receipt_date": "2026-01-10",
|
||
"revenue": "100000",
|
||
"profit": "45000",
|
||
"received_amount": "60000",
|
||
}
|
||
data = client.post(f"{_PARAMS}/preview", json=payload, headers=admin["headers"]).json()[
|
||
"data"
|
||
]
|
||
assert data["payment_days"] == 27
|
||
assert data["rate"] == "97/10000" # 10‰ − 27 × 4/360‰
|
||
assert data["profit_rate"] == "9/20" # 45% 不含运费
|
||
assert data["converted_profit"] == "15000" # 100000 × (45% − 30%)
|
||
assert data["performance_amount"] == "145.5"
|
||
assert data["start_payment_ok"] is True
|
||
assert data["payment_weight_r"] == 60
|
||
assert data["payable_amount"] == "52.38" # 145.5 × 60% × 60%
|
||
|
||
def test_361天分界_固定下限4千分(self, client: TestClient, admin: dict) -> None:
|
||
payload = {
|
||
"business_role": "salesperson",
|
||
"customer_type": "personal",
|
||
"product_category": "municipal_water",
|
||
"ship_date": "2025-01-10",
|
||
"receipt_date": "2026-01-10", # 366 天 ≥ 361
|
||
"revenue": "100000",
|
||
"profit": "45000",
|
||
"received_amount": "100000",
|
||
}
|
||
data = client.post(f"{_PARAMS}/preview", json=payload, headers=admin["headers"]).json()[
|
||
"data"
|
||
]
|
||
# Fraction 自动约分：4/1000 存储为 1/250，按数值断言
|
||
assert Fraction(data["rate"]) == Fraction(4, 1000)
|
||
|
||
def test_经理聚合口径_预览拒绝(self, client: TestClient, admin: dict) -> None:
|
||
payload = {
|
||
"business_role": "sales_manager",
|
||
"customer_type": "personal",
|
||
"product_category": "municipal_water",
|
||
"ship_date": "2026-01-01",
|
||
"receipt_date": "2026-02-01",
|
||
"revenue": "100000",
|
||
"profit": "45000",
|
||
"received_amount": "0",
|
||
}
|
||
resp = client.post(f"{_PARAMS}/preview", json=payload, headers=admin["headers"])
|
||
assert resp.status_code == 400
|
||
assert "聚合" in resp.json()["message"]
|
||
|
||
def test_回款日早于发货日_400(self, client: TestClient, admin: dict) -> None:
|
||
payload = {
|
||
"business_role": "salesperson",
|
||
"customer_type": "personal",
|
||
"product_category": "municipal_water",
|
||
"ship_date": "2026-01-10",
|
||
"receipt_date": "2026-01-01",
|
||
"revenue": "100000",
|
||
"profit": "45000",
|
||
"received_amount": "0",
|
||
}
|
||
assert (
|
||
client.post(f"{_PARAMS}/preview", json=payload, headers=admin["headers"]).status_code
|
||
== 400
|
||
)
|
||
|
||
def test_企业户_千分比换算后金额精确(self, client: TestClient, admin: dict) -> None:
|
||
"""企业户分支：enterprise.rate 存储 "3"（3‰），须换算为 3/1000 再相乘。
|
||
|
||
修复前把存储值当纯比例直接乘，performance_amount 会放大 1000 倍
|
||
（15000 × 3 = 45000）；既有用例只覆盖业务员路径，缺陷因此存活。
|
||
"""
|
||
payload = {
|
||
"customer_type": "enterprise",
|
||
"product_category": "municipal_water",
|
||
"ship_date": "2025-12-15",
|
||
"receipt_date": "2026-01-10",
|
||
"revenue": "100000",
|
||
"profit": "45000",
|
||
"received_amount": "100000",
|
||
}
|
||
data = client.post(f"{_PARAMS}/preview", json=payload, headers=admin["headers"]).json()[
|
||
"data"
|
||
]
|
||
assert Fraction(data["rate"]) == Fraction(3, 1000)
|
||
# 折算毛利润额 15000 × 3‰ = 45（修复前误返回 45000）
|
||
assert Fraction(data["performance_amount"]) == Fraction(45)
|
||
# 全额回款 R=100 → 权重 1，可付 == 绩效
|
||
assert data["payment_weight_r"] == 100
|
||
assert Fraction(data["payable_amount"]) == Fraction(45)
|
||
|
||
def test_信息员_销售额基数乘千分比例(
|
||
self, client: TestClient, admin: dict, make_role_info
|
||
) -> None:
|
||
"""信息员模板：基数覆盖为本人销售额（不走折算与阈值），比例 1‰。
|
||
|
||
修复前 take("info.rate")（存储 "1"）直接当纯比例乘出本人销售额原额。
|
||
"""
|
||
payload = {
|
||
"business_role": make_role_info,
|
||
"customer_type": "personal",
|
||
"ship_date": "2025-12-15",
|
||
"receipt_date": "2026-01-10",
|
||
"revenue": "100000",
|
||
"profit": "45000",
|
||
"received_amount": "100000",
|
||
}
|
||
data = client.post(f"{_PARAMS}/preview", json=payload, headers=admin["headers"]).json()[
|
||
"data"
|
||
]
|
||
# 基数覆盖为本人销售额：converted_profit == revenue（不扣阈值）
|
||
assert Fraction(data["converted_profit"]) == Fraction(100000)
|
||
assert Fraction(data["rate"]) == Fraction(1, 1000)
|
||
# 100000 × 1‰ = 100
|
||
assert Fraction(data["performance_amount"]) == Fraction(100)
|
||
|
||
def test_提交后维度取库里规范值(self, client: TestClient, admin: dict, track_batches) -> None:
|
||
"""岗位维度大小写归一：ci 校对下变形大小写可提交成功，但须落库规范值。
|
||
|
||
快照与注册表按精确键取值，原始大小写入库会让覆盖链静默不命中
|
||
（按错误参数算钱且无报错）。business_roles.role_code 唯一键的 ci
|
||
等价冲突由 IntegrityError 分支翻译 400 兜底。
|
||
"""
|
||
resp = client.post(
|
||
f"{_PARAMS}/changes",
|
||
json={
|
||
"change_type": 1,
|
||
"items": [
|
||
{"param_code": "info.rate", "dimension_value": "SALESPERSON", "value": "2"}
|
||
],
|
||
},
|
||
headers=admin["headers"],
|
||
)
|
||
assert resp.status_code in (200, 400), resp.text
|
||
if resp.status_code == 400:
|
||
pytest.skip("种子岗位未命中（见断言），规范值回读由造岗位链路覆盖")
|
||
batch = resp.json()["data"]
|
||
track_batches(batch["id"])
|
||
detail = client.get(f"{_PARAMS}/changes/{batch['id']}", headers=admin["headers"]).json()[
|
||
"data"
|
||
]
|
||
dims = [item["dimension_value"] for item in detail["items"]]
|
||
assert "salesperson" in dims
|
||
assert "SALESPERSON" not in dims
|
||
|
||
|
||
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(_PARAMS, headers=headers).status_code == 403
|
||
resp = client.post(
|
||
f"{_PARAMS}/changes",
|
||
json={
|
||
"change_type": 1,
|
||
"items": [{"param_code": "prepay.rate", "dimension_value": "", "value": "9"}],
|
||
},
|
||
headers=headers,
|
||
)
|
||
assert resp.status_code == 403
|