f75a9d534e
- tests/integration/factories.py：回款/发货两文件逐字重复的 _customer/_bitem/_bpayload/_create_business 上提（t_ 前缀与默认行 口径单点维护）；test_audit_retry 的持续冲突断言由 500 更新为 400「数据完整性冲突」——全局 IntegrityError handler（批 6）收敛 行为属预期改进，非回归 - param_service.submit_change 拆出 _normalize_items（校验段独立， 主函数降至 50 行约定内）；e2e 六个列表页 helper 收口 common.py （base_data 版为准：分组参数化、:text-is 精确匹配），两脚本变薄 - 全量回归 625 passed（基线 603 + 新增 22）；e2e 四链路（登录 8/ 系统管理 13/基础数据 13/业务 16）全过 - 测试 README 补 2026-10-15 运行记录与 10 条评审缺陷（预览千分比/ 审计假成功/字典序 MAX/无锁编辑/删产品 500/schema 口径/quantize/ 负天数外插/评级并发/大小写不对称） Co-Authored-By: Claude Code <noreply@anthropic.com>
351 lines
14 KiB
Python
351 lines
14 KiB
Python
"""发货管理接口集成测试（需要 MySQL 与 Redis）。
|
||
|
||
守的是发货单的行金额固化与双重汇总回写 —— 行金额（amount）与发货单总额
|
||
（total_amount）若经前端传值，业务级 total_ship_amount 与运费合计就是可
|
||
被篡改的源数据，M3 核算直接读业务冗余列；终态业务发货单禁写缺失会让核算
|
||
基线漂移；被回款批次指定的发货单若能直接删，指定它的回款就指向悬空批次，
|
||
回款天数无从派生。
|
||
|
||
运行前置：``docker compose -f docker/docker-compose.dev.yml up -d``
|
||
|
||
数据卫生：人员/公司/产品名一律 ``t_`` 前缀由 conftest 兜底清理；发货单
|
||
编号系统生成（SH 前缀），conftest 按编号前缀全量清；删除语义的用例自验
|
||
DELETE 端点。
|
||
"""
|
||
|
||
import re
|
||
import uuid
|
||
from datetime import date
|
||
from decimal import Decimal
|
||
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy import select
|
||
|
||
from app.core.constants import ACTION_CREATE, MODULE_SHIPMENT
|
||
from app.core.db import get_session_factory
|
||
from app.models import OperationLog
|
||
from tests.integration.factories import create_business as _create_business
|
||
|
||
pytestmark = pytest.mark.integration
|
||
|
||
_BUSINESSES = "/api/v1/business/businesses"
|
||
_SHIPMENTS = "/api/v1/business/shipments"
|
||
_RECEIPTS = "/api/v1/business/receipts"
|
||
_PERSONNEL = "/api/v1/base/personnel"
|
||
|
||
|
||
def _item(**overrides: object) -> dict:
|
||
"""发货明细行入参（默认：3 × 300.00 = 900）。"""
|
||
body: dict = {
|
||
"product_name": f"t_产品_{uuid.uuid4().hex[:8]}",
|
||
"quantity": "3",
|
||
"sale_price": "300.00",
|
||
}
|
||
body.update(overrides)
|
||
return body
|
||
|
||
|
||
def _spayload(business_id: int, **overrides: object) -> dict:
|
||
body: dict = {
|
||
"business_id": business_id,
|
||
"ship_date": str(date.today()),
|
||
"items": [_item()],
|
||
}
|
||
body.update(overrides)
|
||
return body
|
||
|
||
|
||
def _create(client: TestClient, headers: dict, business_id: int, **overrides: object) -> dict:
|
||
resp = client.post(_SHIPMENTS, json=_spayload(business_id, **overrides), headers=headers)
|
||
assert resp.status_code == 200, resp.text
|
||
return resp.json()["data"]
|
||
|
||
|
||
def _terminate(client: TestClient, headers: dict, business_id: int) -> None:
|
||
"""把业务流转到已终止（终态禁写场景的造数步骤）。"""
|
||
resp = client.put(
|
||
f"{_BUSINESSES}/{business_id}/status",
|
||
json={"status": 3, "reason": f"t_终止_{uuid.uuid4().hex[:6]}"},
|
||
headers=headers,
|
||
)
|
||
assert resp.status_code == 200, resp.text
|
||
|
||
|
||
def _logs_of(username: str) -> list[OperationLog]:
|
||
"""取某操作人的审计行（按 id 倒序，最新在前）。"""
|
||
factory = get_session_factory()
|
||
with factory() as session:
|
||
return list(
|
||
session.execute(
|
||
select(OperationLog)
|
||
.where(OperationLog.username == username)
|
||
.order_by(OperationLog.id.desc())
|
||
).scalars()
|
||
)
|
||
|
||
|
||
class TestAuth:
|
||
"""鉴权：业务数据属业务操作不要求管理员，但仍是登录后接口。"""
|
||
|
||
def test_未登录访问发货单列表_401(self, client: TestClient) -> None:
|
||
resp = client.get(_SHIPMENTS)
|
||
assert resp.status_code == 401, resp.text
|
||
|
||
|
||
class TestCreate:
|
||
"""创建：SH 编号契约、行校验、行金额固化与双重汇总回写 —— 业务级汇总
|
||
是 M3 起付判定的直接消费源，错写就是算错钱。"""
|
||
|
||
def test_创建发货单_行金额固化与双重汇总回写(self, client: TestClient, normal: dict) -> None:
|
||
headers = normal["headers"]
|
||
business = _create_business(client, headers)
|
||
data = _create(
|
||
client,
|
||
headers,
|
||
business["id"],
|
||
freight_amount="88.50",
|
||
items=[
|
||
_item(quantity="3", sale_price="300.00", amount="999.99"),
|
||
_item(quantity="2", sale_price="250.00"),
|
||
],
|
||
)
|
||
|
||
# 编号格式：SH{YYYYMMDD}-{4位当日序号}（日期取当日）
|
||
expected_prefix = f"SH{date.today():%Y%m%d}-"
|
||
assert re.fullmatch(r"SH\d{8}-\d{4}", data["shipment_no"]), data["shipment_no"]
|
||
assert data["shipment_no"].startswith(expected_prefix), data["shipment_no"]
|
||
|
||
# 行金额由服务层按 数量×单价 固化：schema 无 amount 入参，塞了也忽略
|
||
detail = client.get(f"{_SHIPMENTS}/{data['id']}", headers=headers).json()["data"]
|
||
amounts = [Decimal(i["amount"]) for i in detail["items"]]
|
||
assert amounts == [Decimal("900"), Decimal("500")], detail["items"]
|
||
assert Decimal(detail["total_amount"]) == Decimal("1400")
|
||
assert Decimal(detail["freight_amount"]) == Decimal("88.50")
|
||
|
||
# 业务级汇总回写（发货金额 / 运费合计）
|
||
business_detail = client.get(f"{_BUSINESSES}/{business['id']}", headers=headers).json()[
|
||
"data"
|
||
]
|
||
assert Decimal(business_detail["total_ship_amount"]) == Decimal("1400")
|
||
assert Decimal(business_detail["total_freight_amount"]) == Decimal("88.50")
|
||
|
||
def test_业务不存在_404(self, client: TestClient, normal: dict) -> None:
|
||
resp = client.post(_SHIPMENTS, json=_spayload(999999999), headers=normal["headers"])
|
||
assert resp.status_code == 404, resp.text
|
||
assert "业务不存在" in resp.json()["message"], resp.text
|
||
|
||
def test_终态业务禁新发货_400(self, client: TestClient, normal: dict) -> None:
|
||
headers = normal["headers"]
|
||
business = _create_business(client, headers)
|
||
_terminate(client, headers, business["id"])
|
||
resp = client.post(_SHIPMENTS, json=_spayload(business["id"]), headers=headers)
|
||
assert resp.status_code == 400, resp.text
|
||
assert "业务已完成或已终止，禁止新发货" in resp.json()["message"], resp.text
|
||
|
||
def test_明细为空_400(self, client: TestClient, normal: dict) -> None:
|
||
business = _create_business(client, normal["headers"])
|
||
resp = client.post(
|
||
_SHIPMENTS, json=_spayload(business["id"], items=[]), headers=normal["headers"]
|
||
)
|
||
assert resp.status_code == 400, resp.text
|
||
|
||
def test_明细行数量为0_400(self, client: TestClient, normal: dict) -> None:
|
||
business = _create_business(client, normal["headers"])
|
||
resp = client.post(
|
||
_SHIPMENTS,
|
||
json=_spayload(business["id"], items=[_item(quantity="0")]),
|
||
headers=normal["headers"],
|
||
)
|
||
assert resp.status_code == 400, resp.text
|
||
assert "数量" in resp.json()["message"], resp.text
|
||
|
||
def test_运费负数_400(self, client: TestClient, normal: dict) -> None:
|
||
"""运费是金额源数据：负值会让业务运费合计与报表直接错乱。"""
|
||
business = _create_business(client, normal["headers"])
|
||
resp = client.post(
|
||
_SHIPMENTS,
|
||
json=_spayload(business["id"], freight_amount="-8.00"),
|
||
headers=normal["headers"],
|
||
)
|
||
assert resp.status_code == 400, resp.text
|
||
assert "不能为负数" in resp.json()["message"], resp.text
|
||
|
||
def test_运费三位小数_400(self, client: TestClient, normal: dict) -> None:
|
||
business = _create_business(client, normal["headers"])
|
||
resp = client.post(
|
||
_SHIPMENTS,
|
||
json=_spayload(business["id"], freight_amount="8.333"),
|
||
headers=normal["headers"],
|
||
)
|
||
assert resp.status_code == 400, resp.text
|
||
assert "最多两位小数" in resp.json()["message"], resp.text
|
||
|
||
|
||
class TestEdit:
|
||
"""编辑（明细全量替换）与两处汇总同步 —— 业务级汇总漏算即漂移。"""
|
||
|
||
def test_编辑发货单_两处汇总同步(self, client: TestClient, normal: dict) -> None:
|
||
headers = normal["headers"]
|
||
business = _create_business(client, headers)
|
||
data = _create(
|
||
client,
|
||
headers,
|
||
business["id"],
|
||
freight_amount="50.00",
|
||
items=[_item(quantity="1", sale_price="1000.00")],
|
||
)
|
||
before = client.get(f"{_BUSINESSES}/{business['id']}", headers=headers).json()["data"]
|
||
assert Decimal(before["total_ship_amount"]) == Decimal("1000")
|
||
|
||
updated = client.put(
|
||
f"{_SHIPMENTS}/{data['id']}",
|
||
json={
|
||
"ship_date": str(date.today()),
|
||
"freight_amount": "12.00",
|
||
"items": [_item(quantity="2", sale_price="150.00")],
|
||
},
|
||
headers=headers,
|
||
)
|
||
assert updated.status_code == 200, updated.text
|
||
assert Decimal(updated.json()["data"]["total_amount"]) == Decimal("300")
|
||
|
||
# 业务级两处汇总随编辑同步
|
||
after = client.get(f"{_BUSINESSES}/{business['id']}", headers=headers).json()["data"]
|
||
assert Decimal(after["total_ship_amount"]) == Decimal("300")
|
||
assert Decimal(after["total_freight_amount"]) == Decimal("12")
|
||
|
||
detail = client.get(f"{_SHIPMENTS}/{data['id']}", headers=headers).json()["data"]
|
||
assert len(detail["items"]) == 1, "编辑是全量替换，旧行不得残留"
|
||
assert Decimal(detail["items"][0]["amount"]) == Decimal("300")
|
||
|
||
|
||
class TestDelete:
|
||
"""删除保护与汇总回退：被回款指定禁删；删除后业务级汇总必须回算。"""
|
||
|
||
def test_被回款指定禁删_解除后可删_汇总回退(self, client: TestClient, normal: dict) -> None:
|
||
headers = normal["headers"]
|
||
business = _create_business(client, headers)
|
||
data = _create(
|
||
client,
|
||
headers,
|
||
business["id"],
|
||
freight_amount="20.00",
|
||
items=[_item(quantity="1", sale_price="500.00")],
|
||
)
|
||
receipt = client.post(
|
||
_RECEIPTS,
|
||
json={
|
||
"business_id": business["id"],
|
||
"shipment_id": data["id"],
|
||
"receipt_date": str(date.today()),
|
||
"amount": "500.00",
|
||
},
|
||
headers=headers,
|
||
)
|
||
assert receipt.status_code == 200, receipt.text
|
||
|
||
blocked = client.delete(f"{_SHIPMENTS}/{data['id']}", headers=headers)
|
||
assert blocked.status_code == 400, blocked.text
|
||
assert "已被回款批次指定" in blocked.json()["message"], blocked.text
|
||
|
||
# 解除回款的批次指定（未核算回款可删）后发货单可删，业务级汇总回退
|
||
assert (
|
||
client.delete(
|
||
f"{_RECEIPTS}/{receipt.json()['data']['id']}", headers=headers
|
||
).status_code
|
||
== 200
|
||
)
|
||
freed = client.delete(f"{_SHIPMENTS}/{data['id']}", headers=headers)
|
||
assert freed.status_code == 200, freed.text
|
||
business_detail = client.get(f"{_BUSINESSES}/{business['id']}", headers=headers).json()[
|
||
"data"
|
||
]
|
||
assert Decimal(business_detail["total_ship_amount"]) == Decimal("0")
|
||
assert Decimal(business_detail["total_freight_amount"]) == Decimal("0")
|
||
|
||
def test_删除未被指定的发货单_业务级汇总回退(self, client: TestClient, normal: dict) -> None:
|
||
headers = normal["headers"]
|
||
business = _create_business(client, headers)
|
||
data = _create(
|
||
client,
|
||
headers,
|
||
business["id"],
|
||
freight_amount="100.00",
|
||
items=[_item(quantity="2", sale_price="1000.00")],
|
||
)
|
||
before = client.get(f"{_BUSINESSES}/{business['id']}", headers=headers).json()["data"]
|
||
assert Decimal(before["total_ship_amount"]) == Decimal("2000")
|
||
|
||
freed = client.delete(f"{_SHIPMENTS}/{data['id']}", headers=headers)
|
||
assert freed.status_code == 200, freed.text
|
||
assert client.get(f"{_SHIPMENTS}/{data['id']}", headers=headers).status_code == 404
|
||
after = client.get(f"{_BUSINESSES}/{business['id']}", headers=headers).json()["data"]
|
||
assert Decimal(after["total_ship_amount"]) == Decimal("0")
|
||
assert Decimal(after["total_freight_amount"]) == Decimal("0")
|
||
|
||
|
||
class TestTerminal:
|
||
"""终态业务的发货单禁改禁删 —— 发货的运费/金额是业务级汇总与 M3 核算
|
||
的源数据，终态后不得变动。"""
|
||
|
||
def test_终态业务发货单禁改_400(self, client: TestClient, normal: dict) -> None:
|
||
headers = normal["headers"]
|
||
business = _create_business(client, headers)
|
||
data = _create(client, headers, business["id"])
|
||
_terminate(client, headers, business["id"])
|
||
|
||
resp = client.put(
|
||
f"{_SHIPMENTS}/{data['id']}",
|
||
json={
|
||
"ship_date": str(date.today()),
|
||
"items": [_item(quantity="1", sale_price="1.00")],
|
||
},
|
||
headers=headers,
|
||
)
|
||
assert resp.status_code == 400, resp.text
|
||
assert "禁止编辑发货单" in resp.json()["message"], resp.text
|
||
|
||
def test_终态业务发货单禁删_400(self, client: TestClient, normal: dict) -> None:
|
||
headers = normal["headers"]
|
||
business = _create_business(client, headers)
|
||
data = _create(client, headers, business["id"])
|
||
_terminate(client, headers, business["id"])
|
||
|
||
resp = client.delete(f"{_SHIPMENTS}/{data['id']}", headers=headers)
|
||
assert resp.status_code == 400, resp.text
|
||
assert "禁止删除发货单" in resp.json()["message"], resp.text
|
||
|
||
|
||
class TestAudit:
|
||
"""写操作（含校验失败的）都在 operation_logs 留痕 —— 审计是验收项。"""
|
||
|
||
def test_新增发货单成功留审计(self, client: TestClient, normal: dict) -> None:
|
||
headers = normal["headers"]
|
||
business = _create_business(client, headers)
|
||
data = _create(client, headers, business["id"])
|
||
logs = [
|
||
log
|
||
for log in _logs_of(normal["username"])
|
||
if log.module == MODULE_SHIPMENT and log.action == ACTION_CREATE
|
||
]
|
||
assert logs, "新增发货单未留操作日志"
|
||
latest = logs[0]
|
||
assert latest.success == 1
|
||
assert "新增发货单" in latest.summary
|
||
assert data["shipment_no"] in latest.summary
|
||
|
||
def test_终态禁新发货失败留审计(self, client: TestClient, normal: dict) -> None:
|
||
headers = normal["headers"]
|
||
business = _create_business(client, headers)
|
||
_terminate(client, headers, business["id"])
|
||
resp = client.post(_SHIPMENTS, json=_spayload(business["id"]), headers=headers)
|
||
assert resp.status_code == 400, resp.text
|
||
failed = [
|
||
log
|
||
for log in _logs_of(normal["username"])
|
||
if log.module == MODULE_SHIPMENT and log.success == 0
|
||
]
|
||
assert failed, "失败操作未留痕"
|
||
assert "禁止新发货" in (failed[0].error_message or ""), failed[0]
|