test(business): 业务数据管理模块集成用例（70 例）与两处缺陷修复

- 覆盖：编号格式与序号递增/状态机终态拦截/三处汇总冗余一致性/跨业务批次
  400/已核算禁改禁删/毛利润金标准与防除零/asOf 回溯/评级重叠/审计双记录
- 缺陷修复①：change_status 留痕 old_status 取到赋值后属性（时间线失真 3→3）
- 缺陷修复②：freight_amount 漏金额校验（负数/三位小数可落库污染运费合计）
- conftest 扩展业务六表清理（编号前缀 + FK 逆序收口）与评级种子基线恢复
- 全量回归 603 passed（单元 381 / 集成 222）

Co-Authored-By: Claude Code <noreply@anthropic.com>

backend/app/schemas/shipment.py

@@ -95,6 +95,12 @@ class ShipmentCreateIn(BaseModel):
     items: list[ShipmentItemIn] = Field(min_length=1, description="发货明细行，至少 1 行")
     remark: str | None = Field(default=None, max_length=255)
 
+    @field_validator("freight_amount")
+    @classmethod
+    def _check_freight(cls, value: str | None) -> str | None:
+        """运费同样是金额源数据：复用 _check_amount 拒负数/多小数位/超容量。"""
+        return _check_amount(value, "运费金额", AMOUNT_MAX)
+
 
 class ShipmentUpdateIn(BaseModel):
     """编辑发货单（明细全量替换）。
@@ -112,6 +118,12 @@ class ShipmentUpdateIn(BaseModel):
     )
     remark: str | None = Field(default=None, max_length=255)
 
+    @field_validator("freight_amount")
+    @classmethod
+    def _check_freight(cls, value: str | None) -> str | None:
+        """运费同样是金额源数据：复用 _check_amount 拒负数/多小数位/超容量。"""
+        return _check_amount(value, "运费金额", AMOUNT_MAX)
+
 
 class ShipmentQuery(PageQuery):
     """发货单列表查询条件（字段名蛇形直出，与前端 api 封装同名对应）。

backend/app/services/business_service.py

@@ -511,12 +511,15 @@ def change_status(
         if business.status == payload.status:
             raise BadRequestError("状态未变化")
 
+        # 变更前状态须在赋值前固化：ORM 属性赋值后 old_status 就取不到旧值
+        # （留痕会记成 old==new 失真，时间线无法回溯）
+        old_status = business.status
         business.status = payload.status
         business_repository.add_status_log(
             session,
             BusinessStatusLog(
                 business_id=business_id,
-                old_status=business.status,
+                old_status=old_status,
                 new_status=payload.status,
                 change_reason=payload.reason,
                 operator_id=operator.id,
@@ -528,7 +531,7 @@ def change_status(
         logger.info(
             "业务状态变更 | business_id=%s | %s -> %s",
             business_id,
-            business.status,
+            old_status,
             payload.status,
         )
         return out

backend/tests/README.md

@@ -1,6 +1,6 @@
 # 后端测试
 
-> AI摘要：本项目后端的测试用例清单与运行记录。按「基础设施层随实现即时测、业务模块整模块完成时统一测」分层组织。当前 301 个用例全绿（单元 223 / 集成 78）。
+> AI摘要：本项目后端的测试用例清单与运行记录。按「基础设施层随实现即时测、业务模块整模块完成时统一测」分层组织。当前 603 个用例全绿（单元 381 / 集成 222）。
 
 ## 一、测试约定
 
@@ -102,6 +102,17 @@ uv run pytest tests/integration -q   # 仅集成
 
 运行：`uv run pytest tests/integration/test_personnel.py tests/integration/test_products.py tests/integration/test_customers.py -q` → **48 passed**（2026-10-13，连跑两次幂等）。
 
+### 业务数据管理（2026-10-13 完成）
+
+| 文件 | 守什么 | 用例数 |
+| --- | --- | --- |
+| `tests/integration/test_businesses.py` | 未登录 401；编号 `YW{YYYYMMDD}-{4位}` 与当日序号递增；空明细/数量 0/非法类别/单价非数字 400；**销售额服务层固化**（schema 无 sale_amount 入参，塞了忽略）；特殊业务必填说明；编辑全量替换后**三处冗余一致**（明细 SUM == 业务冗余列 == DB SUM）；状态机（终止无原因 400 → 留痕 old=1/new=3 → 终态再变更 400）；归属变更主表+留痕同生（old/new owner 与 assistant、owner_type=2）、归属未变化 400、创建时归属人员已离职 400；有发货/回款禁换客户禁删；空业务删除连带清理明细/状态留痕/业务归属留痕；毛利润视图金标准（销售 10000/成本 30 → 率 997/1000、市政 30% → 折算 6970；**销售额 0 全 0 防除零**；**成本缺失行**派生为 null 不计类别汇总；**asOf 按业务日期回溯旧阈值**）；审计成功/失败双记录 | 25 |
+| `tests/integration/test_shipments.py` | 未登录 401；编号 `SH{YYYYMMDD}-{4位}`；业务不存在 404；终态业务禁新发货 400；空明细/数量 0/运费负数/运费三位小数 400；**行金额与发货单总额服务层固化**；业务 total_ship/total_freight 回写一致；编辑后两处汇总同步；被回款指定禁删（解除后可删）；删除后业务级汇总回退；终态业务发货单禁改禁删；审计双记录 | 15 |
+| `tests/integration/test_receipts.py` | 未登录 401；批次 `HK{YYYYMMDD}-{4位}`；未指定批次可登记且 payment_days 为 null；**指定批次派生「发货日首日计第 1 天」**（2026-10-05 → 2026-11-02 = 29 天，编辑清指定后回 null）；业务不存在 404 / 终态禁新增 400；金额负数/三位小数 400；**跨业务指定批次 400**（「发货批次不属于该业务，无法指定」，编辑路同校验）；合计回写与删除回退；**已核算（改库 settlement_status=1）禁改禁删**；settlement_status 筛选 0/1 两路；审计双记录 | 16 |
+| `tests/integration/test_ratings.py` | 未登录 401；种子四等级全量按 sort_order 升序（A 90-100/B 75-89/C 60-74/D 0-59）；区间倒挂/超上限/负下限 → 400「分值区间不合法」；名称唯一 → 400「该等级名称已存在」；**新增与编辑两路重叠 400 点名冲突等级**；**边界相邻不算重叠正例**（删 C 后新增 60-74 / 把 D 编辑为 0-74，74 与 75 相邻放行）；编辑改名改区间回显；删除后判定表更新、二次删除 404；审计双记录。种子删改由 conftest 兜底恢复 | 14 |
+
+运行：`uv run pytest tests/integration/test_{businesses,shipments,receipts,ratings}.py -q` → **70 passed**（2026-10-13，连跑两次幂等）；全量回归 `uv run pytest -q` → **603 passed**（单元 381 / 集成 222）。
+
 ## 三、运行记录
 
 | 日期 | 范围 | 用例数 | 结果 | 备注 |
@@ -112,6 +123,8 @@ uv run pytest tests/integration -q   # 仅集成
 | 2026-10-10 | 全部 | 282 | ✅ 通过（103.81s） | **系统管理模块（用户/角色/日志）完成**后统一编写该模块套件（新增 91 例：单元 48 / 集成 43）。含操作日志审计机制专项；**e2e 发现并修复 `replace_roles` 唯一键缺陷**（见四、缺陷表）；集成用例需容器在跑 |
 | 2026-10-10 | 全部 | 301 | ✅ 通过（87.40s） | **code-review 缺陷修复后全量回归**（新增/改造 19 例）。密码结构化脱敏、时区归一、日志详情两段式各有专项；集成套件因密码哈希模块级预计算提速（63s→42s） |
 | 2026-10-13 | 基础数据集成三文件 | 48 | ✅ 通过（27.60s） | **基础数据管理模块完成**后统一编写（人员 20/产品 14/客户 14）。审计上下文内外的校验位置差异为本次新发现的缺陷类别（见下） |
+| 2026-10-13 | 业务数据集成四文件 | 70 | ✅ 通过（54.73s，连跑两次幂等） | **业务数据管理（后端接口）完成**后统一编写（业务 25/发货 15/回款 16/评级 14）。全量回归 603 例通过 |
+| 2026-10-13 | 全部 | 603 | ✅ 通过（5:17） | **T11 全量回归**（单元 381 / 集成 222），含本模块 70 例 |
 
 ## 四、测试发现并修复的缺陷
 
@@ -132,13 +145,15 @@ uv run pytest tests/integration -q   # 仅集成
 | 2026-10-10 | 时间筛选未做时区归一 | 混用 aware/naive 抛 TypeError 穿透成 **500**；带时区输入被驱动静默丢弃偏移（查询窗口偏 8 小时且无提示） | `LogQueryBase` 加 field_validator：aware → `astimezone()` 本地 naive、naive 原样；补 4 例专项 |
 | 2026-10-10 | 审计时序两处缺陷 | ① 失败路径 rollback 后读 operator 触发额外查询，账号并发被删时 ObjectDeletedError 被兜底吞掉、失败审计静默丢失；② 成功/失败两分支 8 个实参重复展开，且 role/user service 在 commit 之后仍做查询与响应构造——后置查询一旦抛错，「已提交成功」会被记成失败 | operator 标量进入时固化；审计链 DTO 化（`audit` 4 参 / `_safe_write` 2 参 / `add` 2 参）；service 的 commit 后查询与响应构造全部前置 |
 | 2026-10-13 | 基础数据三 service 的**前置校验在审计上下文之外** | 校验失败（岗位停用/经理自引用/分类不合法等 400）完全不留操作日志，违背「成功与失败都记」的既定原则——与 2026-10-10 审计时序缺陷同源，新写的 service 又犯了（各自独立实现时口径漂移） | personnel/product/customer 三 service 的值域与引用校验统一移入 `audit_service.audit` 上下文内；集成用例锁定（审计成功与失败双记录） |
+| 2026-10-13 | `change_status` 状态留痕的 old_status 取到**赋值后**的 ORM 属性 | 时间线记成 `old==new`（如 1→3 的变更记成 3→3）——「变更前状态」无法回溯，状态流转追溯失效。由集成用例断言 `(old,new)==(1,3)` 抓住 | 赋值前固化 `old_status = business.status` 再流转；应用日志一并修正；回归用例锁定 |
+| 2026-10-13 | 发货 `freight_amount` 漏做金额校验 | 负数/三位小数/超容量的运费直接落库（schema 对 quantity/sale_price 都复用了 `_check_amount`，唯独运费没套用），业务运费合计与报表随之错乱 | shipments.py schema 的 Create/Update 加 `field_validator("freight_amount")`（同套口径，文案「运费金额不能为负数 / 最多两位小数」）；集成用例锁定 |
 
 ## 五、待补用例
 
 | 模块 | 状态 | 说明 |
 | --- | --- | --- |
-| 基础数据管理 | ✅ 已完成 | 2026-10-13 完成（48 例，见上表）；「被业务数据引用」的删除保护在业务数据表就位后于 T11 补测 |
+| 基础数据管理 | ✅ 已完成 | 2026-10-13 完成（48 例，见上表）；「被业务数据引用」的删除保护已随 T11 的业务/发货禁删用例覆盖 |
-| 业务数据管理 | 进行中 | 集成用例在 T11 统一编写（28 项裁决见 `docs/问题与解决方案对照表_v2.xlsx`） |
+| 业务数据管理 | ✅ 后端已完成 | 2026-10-13 完成（70 例，见上表）；前端页面与绩效核算联动（M3 台账核销回写 settlement_status）待 M3 补测 |
 | 绩效工资核算 | 未开始 | 需求已澄清；金额精度需专项用例（`Fraction` 与 1/90 日斜率已在 M1 公式层覆盖） |
 | 报表统计 | 未开始 | 依赖报表口径确认 |
 

backend/tests/integration/conftest.py

@@ -12,9 +12,15 @@
   消失，必须显式清理，否则残留会影响后续用例对留痕条数的断言
 - 操作日志的清理同时按 username 与 user_id —— 账号被删除后其 ``user_id``
   会被 SET NULL，只能靠冗余的 username 追溯，清理逻辑必须覆盖两种状态
+- M2 业务数据（业务/发货/回款）单据编号由系统生成（YW/SH/HK 前缀 + 当日
+  序号），**没有 t_ 前缀可用**，兜底清理按编号前缀全量清（三类单据表目前
+  只有测试会写）；外键均为 RESTRICT，删除顺序不能倒，且必须在客户/人员
+  清理之前（FK 会挡住删客户/删人员）；评级判定表则恢复种子基线 ——
+  评级用例会删改种子构造重叠/相邻场景，恢复保证跨文件幂等
 """
 
 import uuid
+from decimal import Decimal
 
 import pytest
 from sqlalchemy import delete, select, update
@@ -23,18 +29,35 @@ from app.core.constants import ROLE_ADMIN, ROLE_USER
 from app.core.db import get_session_factory
 from app.core.security import create_access_token, hash_password
 from app.models import (
+    Business,
+    BusinessItem,
     BusinessRole,
+    BusinessStatusLog,
     Customer,
+    CustomerRatingGrade,
     OperationLog,
     OwnershipChangeLog,
     Personnel,
     PersonnelPlan,
     PersonnelRegion,
     Product,
+    Receipt,
     Role,
+    Shipment,
+    ShipmentItem,
     User,
     UserRole,
 )
+from app.models.customer import OWNER_TYPE_BUSINESS
+
+# 评级判定表的种子基线（与 db/migration/V5 一致）：判定按 sort_order 升序
+# 取首个 min_score ≤ 总分的等级，种子铺满 0-100
+_RATING_SEEDS = [
+    ("A", Decimal("90"), Decimal("100"), 1, "优秀"),
+    ("B", Decimal("75"), Decimal("89"), 2, "良好"),
+    ("C", Decimal("60"), Decimal("74"), 3, "合格"),
+    ("D", Decimal("0"), Decimal("59"), 4, "待改进"),
+]
 
 TEST_PASSWORD = "TestPwd@2026"
 # 模块级预计算一次：bcrypt（rounds=12）单次约 0.2–0.4 秒，逐用例重复哈希
@@ -69,6 +92,44 @@ def _purge_test_data() -> None:
             session.execute(delete(UserRole).where(UserRole.role_id.in_(role_ids)))
             session.execute(delete(Role).where(Role.id.in_(role_ids)))
 
+        # ---- M2 业务数据（T11）：业务/发货/回款三类单据 ----
+        # 编号系统生成、无 t_ 前缀可用，按业务编号前缀全量清（三类单据表目前
+        # 只有测试会写）。外键均为 RESTRICT，删除顺序不能倒：回款 → 发货
+        # 明细 → 发货单 → 业务明细 → 状态留痕 → 业务维度归属留痕（无外键，
+        # 须显式清）→ 业务主表；且必须在下方客户/人员清理之前，否则 FK 挡住
+        # 删客户/删人员（businesses.customer_id / .owner_personnel_id RESTRICT）
+        business_ids = list(
+            session.execute(select(Business.id).where(Business.business_no.like("YW%"))).scalars()
+        )
+        shipment_ids = list(
+            session.execute(select(Shipment.id).where(Shipment.shipment_no.like("SH%"))).scalars()
+        )
+        if business_ids:
+            session.execute(delete(Receipt).where(Receipt.business_id.in_(business_ids)))
+        if shipment_ids:
+            session.execute(delete(ShipmentItem).where(ShipmentItem.shipment_id.in_(shipment_ids)))
+            session.execute(delete(Shipment).where(Shipment.id.in_(shipment_ids)))
+        if business_ids:
+            session.execute(delete(BusinessItem).where(BusinessItem.business_id.in_(business_ids)))
+            session.execute(
+                delete(BusinessStatusLog).where(BusinessStatusLog.business_id.in_(business_ids))
+            )
+            session.execute(
+                delete(OwnershipChangeLog).where(
+                    OwnershipChangeLog.owner_type == OWNER_TYPE_BUSINESS,
+                    OwnershipChangeLog.business_id.in_(business_ids),
+                )
+            )
+            session.execute(delete(Business).where(Business.id.in_(business_ids)))
+
+        # 评级判定表恢复种子基线：评级用例会删改种子构造重叠/相邻场景，
+        # 兜底恢复保证每个用例开始时都是纯种子（判定表无下游引用，全删全插安全）
+        session.execute(delete(CustomerRatingGrade))
+        session.add_all(
+            CustomerRatingGrade(grade_name=n, min_score=lo, max_score=hi, sort_order=so, remark=rm)
+            for n, lo, hi, so, rm in _RATING_SEEDS
+        )
+
         # ---- M2 基础数据：客户/归属留痕/人员/产品/测试岗位 ----
         # 外键均为 RESTRICT，删除顺序不能倒：留痕无外键先按 customer_id 清，
         # 再删客户（子表）→ 人员（先置空 manager_id 自引用、清子表）→ 产品

backend/tests/integration/test_businesses.py

@@ -0,0 +1,811 @@
+"""业务管理接口集成测试（需要 MySQL 与 Redis）。
+
+守的是业务单的五处汇总冗余一致性与状态机/归属留痕 —— 明细/发货/回款任何
+写路径漏算汇总，业务行的 total_* 就会与源数据漂移，M3 起付判定直接读冗余
+列，漂移就是算错钱；状态留痕的「变更前状态」若取到赋值后的值（old==new
+失真）或归属变更拆成「主表已换、留痕缺失」，事后追溯无据；编辑端点若能
+改归属，就绕过留痕产生「改了归属却无留痕」的旁路。毛利润视图为只读派生
+值，断言锁定金标准口径（销售额 0 防除零、成本缺失行不进类别汇总）。
+
+运行前置：``docker compose -f docker/docker-compose.dev.yml up -d``
+
+数据卫生：人员/公司/产品名一律 ``t_`` 前缀由 conftest 兜底清理；业务/
+发货/回款编号系统生成（YW/SH/HK 前缀），conftest 按编号前缀全量清；
+删除语义的用例自验 DELETE 端点。
+"""
+
+import re
+import uuid
+from datetime import date
+from decimal import Decimal
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy import delete, select
+
+from app.core.constants import (
+    ACTION_CREATE,
+    BUSINESS_ROLE_SALESPERSON,
+    MODULE_BUSINESS,
+    PRODUCT_CATEGORY_MUNICIPAL_WATER,
+    PRODUCT_CATEGORY_RAIL_TRANSIT,
+)
+from app.core.db import get_session_factory
+from app.models import (
+    Business,
+    BusinessItem,
+    BusinessRole,
+    OperationLog,
+    OwnershipChangeLog,
+    ParamChangeBatch,
+    ParamValue,
+)
+from app.models.business import BUSINESS_STATUS_ACTIVE
+from app.models.customer import OWNER_TYPE_BUSINESS
+
+pytestmark = pytest.mark.integration
+
+_BUSINESSES = "/api/v1/business/businesses"
+_PERSONNEL = "/api/v1/base/personnel"
+_CUSTOMERS = "/api/v1/base/customers"
+_PARAMS = "/api/v1/system/params"
+
+
+def _company() -> str:
+    return f"t_公司_{uuid.uuid4().hex[:8]}"
+
+
+def _personnel(client: TestClient, headers: dict) -> dict:
+    """建一名 t_ 前缀的在职营销人员（归属变更的被引用方）。"""
+    factory = get_session_factory()
+    with factory() as session:
+        role_id = int(
+            session.execute(
+                select(BusinessRole.id).where(
+                    BusinessRole.role_code == BUSINESS_ROLE_SALESPERSON,
+                    BusinessRole.status == 1,
+                )
+            ).scalar_one()
+        )
+    resp = client.post(
+        _PERSONNEL,
+        json={
+            "personnel_name": f"t_人员_{uuid.uuid4().hex[:8]}",
+            "business_role_id": role_id,
+        },
+        headers=headers,
+    )
+    assert resp.status_code == 200, resp.text
+    return resp.json()["data"]
+
+
+def _customer(client: TestClient, headers: dict) -> dict:
+    """建一名 t_ 前缀测试客户（业务的必选引用）。"""
+    resp = client.post(
+        _CUSTOMERS,
+        json={
+            "company_name": _company(),
+            "owner_name": "张老板",
+            "customer_type": "personal",
+            "address": "本溪市平山区",
+        },
+        headers=headers,
+    )
+    assert resp.status_code == 200, resp.text
+    return resp.json()["data"]
+
+
+def _item(**overrides: object) -> dict:
+    """业务明细行入参（默认：数量 100 × 单价 100.00 = 销售额 10000，成本 30）。"""
+    body: dict = {
+        "product_name": f"t_产品_{uuid.uuid4().hex[:8]}",
+        "product_category": PRODUCT_CATEGORY_MUNICIPAL_WATER,
+        "quantity": "100",
+        "sale_price": "100.00",
+        "cost_amount": "30.00",
+    }
+    body.update(overrides)
+    return body
+
+
+def _payload(customer_id: int, **overrides: object) -> dict:
+    body: dict = {
+        "customer_id": customer_id,
+        "business_date": str(date.today()),
+        "items": [_item()],
+    }
+    body.update(overrides)
+    return body
+
+
+def _create(client: TestClient, headers: dict, customer_id: int, **overrides: object) -> dict:
+    resp = client.post(_BUSINESSES, json=_payload(customer_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()
+        )
+
+
+@pytest.fixture
+def threshold_batch(client: TestClient, admin: dict):
+    """提交「立即生效」的产品阈值参数变更批次，用例结束整批清理。
+
+    走参数接口的真实路径（system params 要求管理员）；清理按 batch_id 删
+    条目与批次头（test_param_management.track_batches 同款）。
+    """
+    created: list[int] = []
+
+    def _factory(dimension_value: str, value: str) -> dict:
+        resp = client.post(
+            f"{_PARAMS}/changes",
+            json={
+                "change_type": 1,
+                "remark": "t_毛利润阈值变更",
+                "items": [
+                    {
+                        "param_code": "product.threshold",
+                        "dimension_value": dimension_value,
+                        "value": value,
+                    }
+                ],
+            },
+            headers=admin["headers"],
+        )
+        assert resp.status_code == 200, resp.text
+        batch = resp.json()["data"]
+        created.append(batch["id"])
+        return batch
+
+    yield _factory
+
+    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()
+
+
+class TestAuth:
+    """鉴权：业务数据属业务操作不要求管理员，但仍是登录后接口。"""
+
+    def test_未登录访问业务列表_401(self, client: TestClient) -> None:
+        resp = client.get(_BUSINESSES)
+        assert resp.status_code == 401, resp.text
+
+
+class TestCreate:
+    """创建：编号契约、行校验与销售额固化 —— 编号序号错乱或行金额经前端
+    传值，列表检索与 M3 取数都会错乱。"""
+
+    def test_创建业务_编号格式与当日序号递增(self, client: TestClient, normal: dict) -> None:
+        headers = normal["headers"]
+        customer = _customer(client, headers)
+        first = _create(client, headers, customer["id"])
+        second = _create(client, headers, customer["id"])
+
+        # 编号格式：YW{YYYYMMDD}-{4位当日序号}（日期取当日，序号连续 +1）
+        expected_prefix = f"YW{date.today():%Y%m%d}-"
+        for no in (first["business_no"], second["business_no"]):
+            assert re.fullmatch(r"YW\d{8}-\d{4}", no), no
+            assert no.startswith(expected_prefix), (no, expected_prefix)
+        assert int(second["business_no"].rsplit("-", 1)[1]) == (
+            int(first["business_no"].rsplit("-", 1)[1]) + 1
+        ), (first["business_no"], second["business_no"])
+
+        # 回包契约：默认执行中、客户名直出
+        assert first["status"] == BUSINESS_STATUS_ACTIVE
+        assert first["customer_name"] == customer["company_name"]
+
+    def test_空明细_400(self, client: TestClient, normal: dict) -> None:
+        customer = _customer(client, normal["headers"])
+        resp = client.post(
+            _BUSINESSES, json=_payload(customer["id"], items=[]), headers=normal["headers"]
+        )
+        assert resp.status_code == 400, resp.text
+
+    def test_明细行数量为0_400(self, client: TestClient, normal: dict) -> None:
+        customer = _customer(client, normal["headers"])
+        resp = client.post(
+            _BUSINESSES,
+            json=_payload(customer["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:
+        """类别是折算阈值的行级锚点：脏类别取不到阈值参数。"""
+        customer = _customer(client, normal["headers"])
+        resp = client.post(
+            _BUSINESSES,
+            json=_payload(customer["id"], items=[_item(product_category="water")]),
+            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:
+        customer = _customer(client, normal["headers"])
+        resp = client.post(
+            _BUSINESSES,
+            json=_payload(customer["id"], items=[_item(sale_price="abc")]),
+            headers=normal["headers"],
+        )
+        assert resp.status_code == 400, resp.text
+        assert "须为有效数字" in resp.json()["message"], resp.text
+
+    def test_销售额由服务层固化_入参sale_amount不生效(
+        self, client: TestClient, normal: dict
+    ) -> None:
+        """schema 无 sale_amount 入参，塞了也必须被忽略 —— 行金额只能由
+        数量×单价计算，否则前端传值直接改业务销售额。"""
+        headers = normal["headers"]
+        customer = _customer(client, headers)
+        item = _item(quantity="3", sale_price="12.34", sale_amount="88888.00")
+        data = _create(client, headers, customer["id"], items=[item])
+
+        detail = client.get(f"{_BUSINESSES}/{data['id']}", headers=headers).json()["data"]
+        assert Decimal(detail["items"][0]["sale_amount"]) == Decimal("37.02")
+        assert Decimal(detail["total_sale_amount"]) == Decimal("37.02")
+
+    def test_特殊业务缺说明_400(self, client: TestClient, normal: dict) -> None:
+        customer = _customer(client, normal["headers"])
+        resp = client.post(
+            _BUSINESSES,
+            json=_payload(customer["id"], is_special=1),
+            headers=normal["headers"],
+        )
+        assert resp.status_code == 400, resp.text
+        assert "特殊业务必须填写说明" in resp.json()["message"], resp.text
+
+    def test_特殊业务有说明创建成功_回显说明(self, client: TestClient, normal: dict) -> None:
+        headers = normal["headers"]
+        customer = _customer(client, headers)
+        data = _create(client, headers, customer["id"], is_special=1, special_note="公司参与结算")
+        assert data["is_special"] == 1
+        assert data["special_note"] == "公司参与结算"
+
+
+class TestEdit:
+    """编辑：全量替换明细并单点重算汇总 —— 三处冗余（行 SUM / 业务冗余列 /
+    数据库 SUM）不一致就说明重算有旁路。"""
+
+    def test_编辑全量替换明细_三处汇总一致(self, client: TestClient, normal: dict) -> None:
+        headers = normal["headers"]
+        customer = _customer(client, headers)
+        data = _create(
+            client,
+            headers,
+            customer["id"],
+            items=[
+                _item(quantity="100", sale_price="100.00", cost_amount="30.00"),
+                _item(quantity="50", sale_price="200.00", cost_amount="40.00"),
+            ],
+        )
+        before = client.get(f"{_BUSINESSES}/{data['id']}", headers=headers).json()["data"]
+        assert Decimal(before["total_sale_amount"]) == Decimal("20000")
+        assert Decimal(before["total_cost_amount"]) == Decimal("70")
+
+        updated = client.put(
+            f"{_BUSINESSES}/{data['id']}",
+            json=_payload(
+                customer["id"],
+                items=[_item(quantity="200", sale_price="50.00", cost_amount="30.00")],
+            ),
+            headers=headers,
+        )
+        assert updated.status_code == 200, updated.text
+        assert Decimal(updated.json()["data"]["total_sale_amount"]) == Decimal("10000")
+
+        detail = client.get(f"{_BUSINESSES}/{data['id']}", headers=headers).json()["data"]
+        assert len(detail["items"]) == 1, "编辑是全量替换，旧行不得残留"
+        item_sum_sale = sum(Decimal(i["sale_amount"]) for i in detail["items"])
+        item_sum_cost = sum(Decimal(i["cost_amount"]) for i in detail["items"] if i["cost_amount"])
+        assert item_sum_sale == Decimal(detail["total_sale_amount"])
+        assert item_sum_cost == Decimal(detail["total_cost_amount"])
+
+        # 数据库侧再核一次 SUM（接口回包来自内存计算，两处同源才可靠）
+        factory = get_session_factory()
+        with factory() as session:
+            db_business = session.get(Business, data["id"])
+            db_sale = (
+                session.execute(
+                    select(BusinessItem.sale_amount).where(BusinessItem.business_id == data["id"])
+                )
+                .scalars()
+                .all()
+            )
+        assert sum(db_sale) == db_business.total_sale_amount == Decimal("10000.00")
+
+    def test_编辑端点塞owner不生效_归属唯一走ownership(
+        self, client: TestClient, normal: dict
+    ) -> None:
+        """归属刻意不出编辑端点：否则「改归属无留痕」的旁路破坏 append-only。"""
+        headers = normal["headers"]
+        customer = _customer(client, headers)
+        owner = _personnel(client, headers)
+        someone = _personnel(client, headers)
+        data = _create(client, headers, customer["id"], owner_personnel_id=owner["id"])
+
+        resp = client.put(
+            f"{_BUSINESSES}/{data['id']}",
+            json=_payload(customer["id"], owner_personnel_id=someone["id"]),
+            headers=headers,
+        )
+        assert resp.status_code == 200, resp.text
+        assert resp.json()["data"]["owner_personnel_id"] == owner["id"], resp.text
+
+        logs = client.get(f"{_BUSINESSES}/{data['id']}/ownership-logs", headers=headers).json()[
+            "data"
+        ]
+        assert logs["total"] == 0, "编辑端点改归属却无留痕"
+
+
+class TestStatus:
+    """状态机：终态不可逆是终态单据禁写的前提；留痕从创建首条记起。"""
+
+    def test_终止留痕_终态禁再变更(self, client: TestClient, normal: dict) -> None:
+        headers = normal["headers"]
+        customer = _customer(client, headers)
+        data = _create(client, headers, customer["id"])
+
+        # 终止必填原因（schema 层校验，400）
+        no_reason = client.put(
+            f"{_BUSINESSES}/{data['id']}/status", json={"status": 3}, headers=headers
+        )
+        assert no_reason.status_code == 400, no_reason.text
+        assert "终止业务必须填写原因" in no_reason.json()["message"], no_reason.text
+
+        done = client.put(
+            f"{_BUSINESSES}/{data['id']}/status",
+            json={"status": 3, "reason": "客户取消订单"},
+            headers=headers,
+        )
+        assert done.status_code == 200, done.text
+        assert done.json()["data"]["status"] == 3
+
+        # 终态不可再变更
+        again = client.put(
+            f"{_BUSINESSES}/{data['id']}/status",
+            json={"status": 2, "reason": "强行翻转"},
+            headers=headers,
+        )
+        assert again.status_code == 400, again.text
+        assert "不可再变更状态" in again.json()["message"], again.text
+
+        logs = client.get(f"{_BUSINESSES}/{data['id']}/status-logs", headers=headers).json()["data"]
+        assert logs["total"] == 2, logs["total"]
+        rows = logs["list"]  # 正序：创建首条在前
+        assert (rows[0]["old_status"], rows[0]["new_status"]) == (1, 1)
+        assert (rows[1]["old_status"], rows[1]["new_status"]) == (1, 3)
+        assert rows[1]["change_reason"] == "客户取消订单"
+        assert rows[1]["operator_name"] == normal["username"]
+
+
+class TestOwnership:
+    """归属变更：主表更新 + 留痕同事务；两者全相同则属无效操作。"""
+
+    def test_归属变更_主表更新与留痕同生(self, client: TestClient, normal: dict) -> None:
+        headers = normal["headers"]
+        customer = _customer(client, headers)
+        old_owner = _personnel(client, headers)
+        data = _create(client, headers, customer["id"], owner_personnel_id=old_owner["id"])
+
+        new_owner = _personnel(client, headers)
+        new_assistant = _personnel(client, headers)
+        resp = client.put(
+            f"{_BUSINESSES}/{data['id']}/ownership",
+            json={
+                "new_owner_personnel_id": new_owner["id"],
+                "new_assistant_personnel_id": new_assistant["id"],
+                "reason": "人员调动",
+            },
+            headers=headers,
+        )
+        assert resp.status_code == 200, resp.text
+        body = resp.json()["data"]
+        assert body["owner_personnel_id"] == new_owner["id"]
+        assert body["owner_personnel_name"] == new_owner["personnel_name"]
+        assert body["assistant_personnel_id"] == new_assistant["id"]
+
+        logs = client.get(f"{_BUSINESSES}/{data['id']}/ownership-logs", headers=headers).json()[
+            "data"
+        ]
+        assert logs["total"] == 1, "首次归属变更只产生一条留痕"
+        row = logs["list"][0]
+        assert row["old_owner_personnel_id"] == old_owner["id"]
+        assert row["new_owner_personnel_id"] == new_owner["id"]
+        assert row["old_assistant_personnel_id"] is None
+        assert row["new_assistant_personnel_id"] == new_assistant["id"]
+        assert row["old_owner_name"] == old_owner["personnel_name"]
+        assert row["new_owner_name"] == new_owner["personnel_name"]
+        assert row["change_reason"] == "人员调动"
+        assert row["operator_name"] == normal["username"]
+
+        # owner_type 由 DB 固化断言（响应 DTO 不含该列；2=业务归属）
+        factory = get_session_factory()
+        with factory() as session:
+            log = session.execute(
+                select(OwnershipChangeLog).where(OwnershipChangeLog.business_id == data["id"])
+            ).scalar_one()
+        assert log.owner_type == OWNER_TYPE_BUSINESS, (
+            f"owner_type 应为业务归属(2)：{log.owner_type}"
+        )
+
+    def test_归属未变化_400(self, client: TestClient, normal: dict) -> None:
+        headers = normal["headers"]
+        customer = _customer(client, headers)
+        owner = _personnel(client, headers)
+        data = _create(client, headers, customer["id"], owner_personnel_id=owner["id"])
+        resp = client.put(
+            f"{_BUSINESSES}/{data['id']}/ownership",
+            json={"new_owner_personnel_id": owner["id"], "reason": "无变化"},
+            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"]
+        quit_owner = _personnel(client, headers)
+        off = client.put(
+            f"{_PERSONNEL}/{quit_owner['id']}/status", json={"status": 0}, headers=headers
+        )
+        assert off.status_code == 200, off.text
+        customer = _customer(client, headers)
+        resp = client.post(
+            _BUSINESSES,
+            json=_payload(customer["id"], owner_personnel_id=quit_owner["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:
+        headers = normal["headers"]
+        customer = _customer(client, headers)
+        data = _create(client, headers, customer["id"])
+        resp = client.put(
+            f"{_BUSINESSES}/{data['id']}/ownership",
+            json={"new_owner_personnel_id": 999999999, "reason": "填错人"},
+            headers=headers,
+        )
+        assert resp.status_code == 400, resp.text
+        assert "负责人员不存在或已离职" in resp.json()["message"], resp.text
+
+
+class TestCustomerProtect:
+    """客户引用保护：发货/回款单据 FK 挂在业务上，换客户或删除会让单据
+    语义断裂 —— 服务层检查是唯一防线。"""
+
+    def test_有发货或回款时换客户_400(self, client: TestClient, normal: dict) -> None:
+        headers = normal["headers"]
+        customer = _customer(client, headers)
+        other = _customer(client, headers)
+        data = _create(client, headers, customer["id"])
+
+        # 有发货记录：禁换
+        ship = client.post(
+            "/api/v1/business/shipments",
+            json={
+                "business_id": data["id"],
+                "ship_date": str(date.today()),
+                "items": [_item(product_name="t_货", quantity="1", sale_price="10.00")],
+            },
+            headers=headers,
+        )
+        assert ship.status_code == 200, ship.text
+        blocked_ship = client.put(
+            f"{_BUSINESSES}/{data['id']}",
+            json=_payload(other["id"]),
+            headers=headers,
+        )
+        assert blocked_ship.status_code == 400, blocked_ship.text
+        assert "已有发货或回款记录" in blocked_ship.json()["message"], blocked_ship.text
+
+        # 解除发货后仍有回款：禁换（同一路径 has_docs）
+        client.delete(f"/api/v1/business/shipments/{ship.json()['data']['id']}", headers=headers)
+        receipt = client.post(
+            "/api/v1/business/receipts",
+            json={
+                "business_id": data["id"],
+                "receipt_date": str(date.today()),
+                "amount": "100.00",
+            },
+            headers=headers,
+        )
+        assert receipt.status_code == 200, receipt.text
+        blocked_receipt = client.put(
+            f"{_BUSINESSES}/{data['id']}",
+            json=_payload(other["id"]),
+            headers=headers,
+        )
+        assert blocked_receipt.status_code == 400, blocked_receipt.text
+        assert "已有发货或回款记录" in blocked_receipt.json()["message"], blocked_receipt.text
+
+    def test_有发货或回款禁删_400(self, client: TestClient, normal: dict) -> None:
+        headers = normal["headers"]
+        customer = _customer(client, headers)
+        data = _create(client, headers, customer["id"])
+
+        ship = client.post(
+            "/api/v1/business/shipments",
+            json={
+                "business_id": data["id"],
+                "ship_date": str(date.today()),
+                "items": [_item(product_name="t_货", quantity="1", sale_price="10.00")],
+            },
+            headers=headers,
+        )
+        assert ship.status_code == 200, ship.text
+        blocked_ship = client.delete(f"{_BUSINESSES}/{data['id']}", headers=headers)
+        assert blocked_ship.status_code == 400, blocked_ship.text
+        assert "请改为已终止而非删除" in blocked_ship.json()["message"], blocked_ship.text
+
+        # 解除发货后仍有回款：禁删（文案同一路径）
+        client.delete(f"/api/v1/business/shipments/{ship.json()['data']['id']}", headers=headers)
+        receipt = client.post(
+            "/api/v1/business/receipts",
+            json={
+                "business_id": data["id"],
+                "receipt_date": str(date.today()),
+                "amount": "100.00",
+            },
+            headers=headers,
+        )
+        assert receipt.status_code == 200, receipt.text
+        blocked_receipt = client.delete(f"{_BUSINESSES}/{data['id']}", headers=headers)
+        assert blocked_receipt.status_code == 400, blocked_receipt.text
+        assert "请改为已终止而非删除" in blocked_receipt.json()["message"], blocked_receipt.text
+
+    def test_空业务删除成功_连带清理明细与留痕(self, client: TestClient, normal: dict) -> None:
+        headers = normal["headers"]
+        customer = _customer(client, headers)
+        owner = _personnel(client, headers)
+        new_owner = _personnel(client, headers)
+        data = _create(
+            client,
+            headers,
+            customer["id"],
+            owner_personnel_id=owner["id"],
+            items=[_item(quantity="1", sale_price="5.00")],
+        )
+        # 先造一次归属留痕，删除时须连带清理（无外键固化表）
+        moved = client.put(
+            f"{_BUSINESSES}/{data['id']}/ownership",
+            json={"new_owner_personnel_id": new_owner["id"], "reason": "调动"},
+            headers=headers,
+        )
+        assert moved.status_code == 200, moved.text
+
+        deleted = client.delete(f"{_BUSINESSES}/{data['id']}", headers=headers)
+        assert deleted.status_code == 200, deleted.text
+        assert client.get(f"{_BUSINESSES}/{data['id']}", headers=headers).status_code == 404
+
+        factory = get_session_factory()
+        with factory() as session:
+            assert session.get(Business, data["id"]) is None
+            assert (
+                session.execute(select(BusinessItem).where(BusinessItem.business_id == data["id"]))
+                .scalars()
+                .first()
+                is None
+            )
+            # 状态留痕（business_status_logs）与归属留痕（owner_type=2）均无
+            # 外键固化，删除时必须显式清
+            assert (
+                session.execute(
+                    select(OwnershipChangeLog).where(
+                        OwnershipChangeLog.business_id == data["id"],
+                        OwnershipChangeLog.owner_type == OWNER_TYPE_BUSINESS,
+                    )
+                )
+                .scalars()
+                .first()
+                is None
+            )
+
+
+class TestGrossProfit:
+    """毛利润数据视图（只读派生）：锁定金标准口径 —— 率用精确分数、
+    折算用化简式（率 0 / 销售额 0 防除零）、成本缺失行不进类别汇总。"""
+
+    def test_毛利润基准值_率与折算锁定精确分数(self, client: TestClient, normal: dict) -> None:
+        """销售 10000 / 成本 30 → 利润 9970、率 997/1000、市政阈值 30% →
+        折算 10000 ×(997/1000 − 30/100) = 6970。"""
+        headers = normal["headers"]
+        customer = _customer(client, headers)
+        data = _create(
+            client,
+            headers,
+            customer["id"],
+            items=[_item(quantity="1", sale_price="10000.00", cost_amount="30.00")],
+        )
+        resp = client.get(f"{_BUSINESSES}/{data['id']}/gross-profit", headers=headers)
+        assert resp.status_code == 200, resp.text
+        view = resp.json()["data"]
+        assert view["as_of"] == str(date.today())
+
+        row = view["rows"][0]
+        assert Decimal(row["sale_amount"]) == Decimal("10000")
+        assert Decimal(row["cost_amount"]) == Decimal("30")
+        assert Decimal(row["profit_amount"]) == Decimal("9970")
+        assert row["profit_rate"] == "997/1000", "率须精确分数串（约分后），不得截断"
+        assert row["threshold"] == "30"
+        assert Decimal(row["converted_profit"]) == Decimal("6970")
+        assert row["missing_reason"] is None
+        assert view["missing_count"] == 0
+
+        cats = view["categories"]
+        assert [c["product_category"] for c in cats] == ["municipal_water"]
+        assert Decimal(cats[0]["sale_amount"]) == Decimal("10000")
+        assert Decimal(cats[0]["profit_amount"]) == Decimal("9970")
+        assert cats[0]["profit_rate"] == "997/1000"
+        assert Decimal(cats[0]["converted_profit"]) == Decimal("6970")
+
+    def test_毛利润_销售额0_全0不抛错_防除零(self, client: TestClient, normal: dict) -> None:
+        """单价空 → 行销售额 0：率分母为 0 时必须返回 0 而非抛除零异常，
+        否则录入期（补价前）打开毛利润视图就是 500。"""
+        headers = normal["headers"]
+        customer = _customer(client, headers)
+        data = _create(
+            client,
+            headers,
+            customer["id"],
+            items=[_item(quantity="5", sale_price=None, cost_amount="0.00")],
+        )
+        resp = client.get(f"{_BUSINESSES}/{data['id']}/gross-profit", headers=headers)
+        assert resp.status_code == 200, resp.text
+        view = resp.json()["data"]
+        row = view["rows"][0]
+        assert Decimal(row["sale_amount"]) == Decimal("0")
+        assert Decimal(row["profit_amount"]) == Decimal("0")
+        assert row["profit_rate"] == "0"
+        assert Decimal(row["converted_profit"]) == Decimal("0")
+        assert view["missing_count"] == 0
+
+        cats = view["categories"]
+        assert [c["product_category"] for c in cats] == ["municipal_water"]
+        assert Decimal(cats[0]["sale_amount"]) == Decimal("0")
+        assert Decimal(cats[0]["converted_profit"]) == Decimal("0")
+
+    def test_毛利润_成本缺失行_派生为空不计入类别汇总(
+        self, client: TestClient, normal: dict
+    ) -> None:
+        headers = normal["headers"]
+        customer = _customer(client, headers)
+        data = _create(
+            client,
+            headers,
+            customer["id"],
+            items=[
+                _item(
+                    product_name="t_产品_市政",
+                    quantity="1",
+                    sale_price="5000.00",
+                    cost_amount="100.00",
+                ),
+                _item(
+                    product_name="t_产品_轨交",
+                    product_category=PRODUCT_CATEGORY_RAIL_TRANSIT,
+                    quantity="2",
+                    sale_price=None,
+                    cost_amount=None,
+                ),
+            ],
+        )
+        resp = client.get(f"{_BUSINESSES}/{data['id']}/gross-profit", headers=headers)
+        assert resp.status_code == 200, resp.text
+        view = resp.json()["data"]
+        assert view["missing_count"] == 1
+
+        rows = view["rows"]
+        assert rows[1]["cost_amount"] is None
+        assert rows[1]["profit_amount"] is None
+        assert rows[1]["profit_rate"] is None
+        assert rows[1]["converted_profit"] is None
+        assert "成本缺失" in (rows[1]["missing_reason"] or ""), rows[1]
+        assert rows[1]["threshold"] == "20"
+        # 可计算行正常派生
+        assert Decimal(rows[0]["profit_amount"]) == Decimal("4900")
+        assert rows[0]["profit_rate"] == "49/50"
+
+        cats = view["categories"]
+        assert [c["product_category"] for c in cats] == ["municipal_water"], "缺失行不得进类别汇总"
+        assert Decimal(cats[0]["sale_amount"]) == Decimal("5000")
+        assert Decimal(cats[0]["cost_amount"]) == Decimal("100")
+        assert Decimal(cats[0]["profit_amount"]) == Decimal("4900")
+        assert cats[0]["profit_rate"] == "49/50"
+        assert Decimal(cats[0]["converted_profit"]) == Decimal("3400")
+
+    def test_毛利润asOf按业务日期取参数时点值_旧日期取旧值(
+        self, client: TestClient, normal: dict, threshold_batch
+    ) -> None:
+        """改参数后按旧业务日期取值必须回溯旧阈值 —— 落库即漂移的「假快照」
+        靠 asOf 回溯参数快照，M3 台账才是法定固化点。"""
+        headers = normal["headers"]
+        customer = _customer(client, headers)
+        data = _create(
+            client,
+            headers,
+            customer["id"],
+            items=[_item(quantity="1", sale_price="10000.00", cost_amount="30.00")],
+        )
+        threshold_batch("municipal_water", "25")  # 立即生效（今日）
+
+        today = client.get(
+            f"{_BUSINESSES}/{data['id']}/gross-profit",
+            params={"as_of": str(date.today())},
+            headers=headers,
+        )
+        assert today.status_code == 200, today.text
+        row_today = today.json()["data"]["rows"][0]
+        assert row_today["threshold"] == "25"
+        # 25%：10000 ×(997/1000 − 25/100) = 7470
+        assert Decimal(row_today["converted_profit"]) == Decimal("7470")
+
+        past = client.get(
+            f"{_BUSINESSES}/{data['id']}/gross-profit",
+            params={"as_of": "2026-02-17"},
+            headers=headers,
+        )
+        assert past.status_code == 200, past.text
+        row_past = past.json()["data"]["rows"][0]
+        assert row_past["threshold"] == "30", "旧业务日期须回溯变更前的阈值"
+        assert Decimal(row_past["converted_profit"]) == Decimal("6970")
+
+
+class TestAudit:
+    """写操作（含校验失败的）都在 operation_logs 留痕 —— 审计是验收项。"""
+
+    def test_新增业务成功留审计(self, client: TestClient, normal: dict) -> None:
+        headers = normal["headers"]
+        customer = _customer(client, headers)
+        _create(client, headers, customer["id"])
+        logs = [
+            log
+            for log in _logs_of(normal["username"])
+            if log.module == MODULE_BUSINESS and log.action == ACTION_CREATE
+        ]
+        assert logs, "新增业务未留操作日志"
+        latest = logs[0]
+        assert latest.success == 1
+        assert "创建业务" in latest.summary
+
+    def test_归属人员不存在失败留审计(self, client: TestClient, normal: dict) -> None:
+        customer = _customer(client, normal["headers"])
+        resp = client.post(
+            _BUSINESSES,
+            json=_payload(customer["id"], owner_personnel_id=999999999),
+            headers=normal["headers"],
+        )
+        assert resp.status_code == 400, resp.text
+        failed = [
+            log
+            for log in _logs_of(normal["username"])
+            if log.module == MODULE_BUSINESS and log.success == 0
+        ]
+        assert failed, "失败操作未留痕"
+        assert "负责人员不存在或已离职" in (failed[0].error_message or ""), failed[0]

backend/tests/integration/test_ratings.py

@@ -0,0 +1,294 @@
+"""客户评级等级配置接口集成测试（需要 MySQL 与 Redis）。
+
+守的是评级**判定表**的两条硬约束 —— ① 区间互不重叠：重叠会让同一总分
+命中多个等级（或两段都跳过），判定表失效，新增与编辑两路都必须校验；
+② 分值域 0≤min≤max≤100：越界/倒挂的配置直接算错等级。边界相邻不算重叠
+（前段 max=74、后段 min=75 相邻）是种子铺满 0-100 的铺法，要有正例锁住；
+「已存在」的名称唯一在先查与 1062 两路都要收敛成 400。
+
+评级用例会**删改种子**（删除 C、编辑 D 的区间构造相邻/重叠场景），conftest
+兜底在每个用例后把判定表恢复为种子基线（全删全插，判定表无下游引用），
+因此种子断言用例（列表全量返回）始终在纯种子上运行。
+
+运行前置：``docker compose -f docker/docker-compose.dev.yml up -d``
+
+数据卫生：测试等级名一律 ``t_`` 前缀；种子（A/B/C/D）的删改由 conftest
+恢复，删除语义的用例自验 DELETE 端点。
+"""
+
+import uuid
+from decimal import Decimal
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy import select
+
+from app.core.constants import ACTION_CREATE, MODULE_RATING
+from app.core.db import get_session_factory
+from app.models import OperationLog
+
+pytestmark = pytest.mark.integration
+
+_GRADES = "/api/v1/business/rating-grades"
+
+# 种子基线（与 V5 迁移一致）：分数字符串经 DB round-trip 后一位小数（Numeric(5,1)）
+_SEED = [
+    ("A", "90.0", "100.0", 1),
+    ("B", "75.0", "89.0", 2),
+    ("C", "60.0", "74.0", 3),
+    ("D", "0.0", "59.0", 4),
+]
+
+
+def _tname() -> str:
+    return f"t_等级_{uuid.uuid4().hex[:6]}"
+
+
+def _gpayload(grade_name: str, min_score: str, max_score: str, **overrides: object) -> dict:
+    body: dict = {
+        "grade_name": grade_name,
+        "min_score": min_score,
+        "max_score": max_score,
+        "sort_order": 99,
+        "remark": "t_评级测试",
+    }
+    body.update(overrides)
+    return body
+
+
+def _seed_id(client: TestClient, headers: dict, grade_name: str) -> int:
+    """从判定表列表取种子等级 id（种子由迁移初始化，主键不可假造）。"""
+    rows = client.get(_GRADES, headers=headers).json()["data"]
+    match = next(row for row in rows if row["grade_name"] == grade_name)
+    return match["id"]
+
+
+def _drop_seed(client: TestClient, headers: dict, grade_name: str) -> None:
+    """删掉种子等级，为测试等级清出分值空隙（种子由 conftest 兜底恢复）。
+
+    种子铺满 0-100 —— 凡要创建/编辑测试等级的用例，必须先清出与既有等级
+    完全分离的空隙，否则重叠约束（设计如此）一律 400。
+    """
+    resp = client.delete(f"{_GRADES}/{_seed_id(client, headers, grade_name)}", headers=headers)
+    assert resp.status_code == 200, resp.text
+
+
+def _names(client: TestClient, headers: dict) -> list[str]:
+    return [row["grade_name"] for row in client.get(_GRADES, headers=headers).json()["data"]]
+
+
+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(_GRADES)
+        assert resp.status_code == 401, resp.text
+
+
+class TestList:
+    """列表直出种子基线：判定表是前端评级计算器的唯一数据源。"""
+
+    def test_种子四等级全量返回_按sort_order升序(self, client: TestClient, normal: dict) -> None:
+        resp = client.get(_GRADES, headers=normal["headers"])
+        assert resp.status_code == 200, resp.text
+        rows = resp.json()["data"]
+        assert [(r["grade_name"], r["sort_order"]) for r in rows] == [
+            (name, order) for name, _, _, order in _SEED
+        ], rows
+        # 分值红线：str 出参且与种子精确相等（Decimal 比较，不经 float）
+        for row, (_, min_score, max_score, _) in zip(rows, _SEED, strict=True):
+            assert Decimal(row["min_score"]) == Decimal(min_score)
+            assert Decimal(row["max_score"]) == Decimal(max_score)
+
+
+class TestCreate:
+    """新增：区间合法（0≤min≤max≤100）、互不重叠、名称唯一 —— 三条硬约束
+    任一缺失都会算错等级。"""
+
+    def test_删除种子后新增相邻区间_成功且判定表更新(
+        self, client: TestClient, normal: dict
+    ) -> None:
+        """种子铺满 0-100，任何新区间都与某条重叠 —— 先删 C（60-74），
+        新增 60-74 与 D(0-59)/B(75-89) **边界相邻不算重叠**（正例）。"""
+        headers = normal["headers"]
+        client.delete(f"{_GRADES}/{_seed_id(client, headers, 'C')}", headers=headers)
+        resp = client.post(
+            _GRADES, json=_gpayload("t_E", "60", "74", sort_order=5), headers=headers
+        )
+        assert resp.status_code == 200, resp.text
+        body = resp.json()["data"]
+        assert Decimal(body["min_score"]) == Decimal("60")
+        assert Decimal(body["max_score"]) == Decimal("74")
+        assert body["sort_order"] == 5
+
+        rows = client.get(_GRADES, headers=headers).json()["data"]
+        assert _names(client, headers) == ["A", "B", "D", "t_E"], rows
+
+    def test_区间倒挂_400(self, client: TestClient, normal: dict) -> None:
+        resp = client.post(_GRADES, json=_gpayload(_tname(), "80", "10"), 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:
+        resp = client.post(
+            _GRADES, json=_gpayload(_tname(), "95", "150"), 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:
+        resp = client.post(_GRADES, json=_gpayload(_tname(), "-5", "10"), 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:
+        """先删 D 清出空隙：重叠校验在名称唯一校验之前，不清空隙会先撞
+        「与等级 D 重叠」而走不到名称校验。"""
+        headers = normal["headers"]
+        _drop_seed(client, headers, "D")
+        resp = client.post(_GRADES, json=_gpayload("A", "10", "20"), 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:
+        """85-95 与种子 A(90-100) 闭区间相交 → 400 并点名 A，前端据此定位行。"""
+        resp = client.post(_GRADES, json=_gpayload(_tname(), "85", "95"), headers=normal["headers"])
+        assert resp.status_code == 400, resp.text
+        assert "分值区间与等级 A 重叠" in resp.json()["message"], resp.text
+        # 失败不落库：判定表仍是 4 条种子
+        assert _names(client, normal["headers"]) == ["A", "B", "C", "D"]
+
+
+class TestEdit:
+    """编辑以修改后的区间对全部既有等级（排除自身）两两相交 —— 编辑路与
+    新增路共用重叠约束，两处都要拦。"""
+
+    def test_编辑与其他等级重叠_400点名冲突等级(self, client: TestClient, normal: dict) -> None:
+        """把 B 编辑为 60-65：与 C(60-74) 相交 → 点名 C；与 D(0-59)/A 均分离。"""
+        headers = normal["headers"]
+        b_id = _seed_id(client, headers, "B")
+        resp = client.put(
+            f"{_GRADES}/{b_id}",
+            json=_gpayload("B", "60", "65", sort_order=2),
+            headers=headers,
+        )
+        assert resp.status_code == 400, resp.text
+        assert "分值区间与等级 C 重叠" in resp.json()["message"], resp.text
+        # 失败不落库：种子保持原样
+        assert _names(client, headers) == ["A", "B", "C", "D"]
+
+    def test_编辑相邻区间_成功且判定表更新(self, client: TestClient, normal: dict) -> None:
+        """删除 C 后把 D 编辑为 0-74：与 B(75-89) **边界相邻不算重叠**
+        （74 与 75 相邻），这是种子铺法的正例。"""
+        headers = normal["headers"]
+        client.delete(f"{_GRADES}/{_seed_id(client, headers, 'C')}", headers=headers)
+        d_id = _seed_id(client, headers, "D")
+        resp = client.put(
+            f"{_GRADES}/{d_id}",
+            json=_gpayload("D", "0", "74", sort_order=4),
+            headers=headers,
+        )
+        assert resp.status_code == 200, resp.text
+
+        rows = client.get(_GRADES, headers=headers).json()["data"]
+        assert [(r["grade_name"], r["sort_order"]) for r in rows] == [("A", 1), ("B", 2), ("D", 4)]
+        assert Decimal(rows[2]["max_score"]) == Decimal("74"), rows[2]
+
+    def test_编辑回显改名与分数(self, client: TestClient, normal: dict) -> None:
+        headers = normal["headers"]
+        _drop_seed(client, headers, "D")  # 清出 0-59 空隙供测试等级落脚
+        grade_name = _tname()
+        created = client.post(
+            _GRADES, json=_gpayload(grade_name, "10", "20", sort_order=50), headers=headers
+        )
+        assert created.status_code == 200, created.text
+        grade_id = created.json()["data"]["id"]
+
+        new_name = _tname()
+        updated = client.put(
+            f"{_GRADES}/{grade_id}",
+            json=_gpayload(new_name, "15", "25", sort_order=60, remark="t_改"),
+            headers=headers,
+        )
+        assert updated.status_code == 200, updated.text
+        body = updated.json()["data"]
+        assert body["grade_name"] == new_name
+        assert Decimal(body["min_score"]) == Decimal("15")
+        assert Decimal(body["max_score"]) == Decimal("25")
+        assert body["sort_order"] == 60
+
+        rows = client.get(_GRADES, headers=headers).json()["data"]
+        match = next(r for r in rows if r["id"] == grade_id)
+        assert match["grade_name"] == new_name
+        assert Decimal(match["min_score"]) == Decimal("15")
+
+
+class TestDelete:
+    """删除允许清空（评级是纯计算器、评分记录不落库）—— 删除后判定表
+    立即反映，二次删除 404。"""
+
+    def test_删除等级后判定表更新_二次删除404(self, client: TestClient, normal: dict) -> None:
+        headers = normal["headers"]
+        _drop_seed(client, headers, "D")  # 清出 0-59 空隙供测试等级落脚
+        grade_name = _tname()
+        created = client.post(
+            _GRADES, json=_gpayload(grade_name, "30.5", "40.5", sort_order=50), headers=headers
+        )
+        assert created.status_code == 200, created.text
+        grade_id = created.json()["data"]["id"]
+        assert grade_name in _names(client, headers)
+
+        assert client.delete(f"{_GRADES}/{grade_id}", headers=headers).status_code == 200
+        assert grade_name not in _names(client, headers)
+
+        again = client.delete(f"{_GRADES}/{grade_id}", headers=headers)
+        assert again.status_code == 404, again.text
+        assert "评级等级不存在" in again.json()["message"], again.text
+
+
+class TestAudit:
+    """写操作（含校验失败的）都在 operation_logs 留痕 —— 审计是验收项。"""
+
+    def test_新增等级成功留审计(self, client: TestClient, normal: dict) -> None:
+        headers = normal["headers"]
+        _drop_seed(client, headers, "D")  # 清出 0-59 空隙供测试等级落脚
+        grade_name = _tname()
+        resp = client.post(
+            _GRADES,
+            json=_gpayload(grade_name, "30.5", "40.5", sort_order=50),
+            headers=headers,
+        )
+        assert resp.status_code == 200, resp.text
+        logs = [
+            log
+            for log in _logs_of(normal["username"])
+            if log.module == MODULE_RATING and log.action == ACTION_CREATE
+        ]
+        assert logs, "新增评级等级未留操作日志"
+        latest = logs[0]
+        assert latest.success == 1
+        assert grade_name in latest.summary
+
+    def test_重叠失败留审计(self, client: TestClient, normal: dict) -> None:
+        resp = client.post(_GRADES, json=_gpayload(_tname(), "85", "95"), headers=normal["headers"])
+        assert resp.status_code == 400, resp.text
+        failed = [
+            log
+            for log in _logs_of(normal["username"])
+            if log.module == MODULE_RATING and log.success == 0
+        ]
+        assert failed, "失败操作未留痕"
+        assert "分值区间与等级 A 重叠" in (failed[0].error_message or ""), failed[0]

backend/tests/integration/test_receipts.py

@@ -0,0 +1,422 @@
+"""回款管理接口集成测试（需要 MySQL 与 Redis）。
+
+守的是回款的数据完整性三道线 —— ① 发货批次同业务强校验：跨业务指定会让
+回款天数与台账归属漂移；② settlement_status 是 M3 核算防重锚点（需规
+2.3.3「每笔回款仅核算一次」），已核算禁改禁删缺失会让台账与源数据漂移；
+③ total_received_amount 回写是 M3 起付判定的直接消费源，删行漏算即漂移。
+payment_days 为只读派生值，锁定「发货日首日计第 1 天」口径。
+
+运行前置：``docker compose -f docker/docker-compose.dev.yml up -d``
+
+数据卫生：人员/公司/产品名一律 ``t_`` 前缀由 conftest 兜底清理；回款批次
+编号系统生成（HK 前缀），conftest 按编号前缀全量清；「已核算」的造数走
+直接改库（M2 只写 0、置位是 M3 职责），M3 核算置位语义不绕开。
+"""
+
+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_RECEIPT
+from app.core.db import get_session_factory
+from app.models import OperationLog, Receipt
+from app.models.receipt import SETTLEMENT_DONE
+
+pytestmark = pytest.mark.integration
+
+_BUSINESSES = "/api/v1/business/businesses"
+_SHIPMENTS = "/api/v1/business/shipments"
+_RECEIPTS = "/api/v1/business/receipts"
+_CUSTOMERS = "/api/v1/base/customers"
+
+
+def _customer(client: TestClient, headers: dict) -> dict:
+    resp = client.post(
+        _CUSTOMERS,
+        json={
+            "company_name": f"t_公司_{uuid.uuid4().hex[:8]}",
+            "owner_name": "张老板",
+            "customer_type": "personal",
+        },
+        headers=headers,
+    )
+    assert resp.status_code == 200, resp.text
+    return resp.json()["data"]
+
+
+def _bitem(**overrides: object) -> dict:
+    """业务明细行入参（发货创建的门槛：业务须有明细 ≥1 行）。"""
+    body: dict = {
+        "product_name": f"t_产品_{uuid.uuid4().hex[:8]}",
+        "product_category": "municipal_water",
+        "quantity": "100",
+        "sale_price": "100.00",
+    }
+    body.update(overrides)
+    return body
+
+
+def _bpayload(customer_id: int, **overrides: object) -> dict:
+    body: dict = {
+        "customer_id": customer_id,
+        "business_date": str(date.today()),
+        "items": [_bitem()],
+    }
+    body.update(overrides)
+    return body
+
+
+def _create_business(client: TestClient, headers: dict) -> dict:
+    """建一名 t_ 客户并挂一笔默认业务（回款/发货单挂在业务上）。"""
+    customer = _customer(client, headers)
+    resp = client.post(_BUSINESSES, json=_bpayload(customer["id"]), headers=headers)
+    assert resp.status_code == 200, resp.text
+    return resp.json()["data"]
+
+
+def _sitem(**overrides: object) -> dict:
+    body: dict = {
+        "product_name": f"t_产品_{uuid.uuid4().hex[:8]}",
+        "quantity": "1",
+        "sale_price": "500.00",
+    }
+    body.update(overrides)
+    return body
+
+
+def _spayload(business_id: int, ship_date: str, **overrides: object) -> dict:
+    body: dict = {
+        "business_id": business_id,
+        "ship_date": ship_date,
+        "items": [_sitem()],
+    }
+    body.update(overrides)
+    return body
+
+
+def _create_shipment(client: TestClient, headers: dict, business_id: int, ship_date: str) -> dict:
+    resp = client.post(_SHIPMENTS, json=_spayload(business_id, ship_date), headers=headers)
+    assert resp.status_code == 200, resp.text
+    return resp.json()["data"]
+
+
+def _rpayload(business_id: int, **overrides: object) -> dict:
+    body: dict = {
+        "business_id": business_id,
+        "receipt_date": str(date.today()),
+        "amount": "1200.00",
+    }
+    body.update(overrides)
+    return body
+
+
+def _create(client: TestClient, headers: dict, business_id: int, **overrides: object) -> dict:
+    resp = client.post(_RECEIPTS, json=_rpayload(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 _settle(receipt_id: int) -> None:
+    """已核算造数：M2 只写 0，置 1 由 M3 核算负责 —— 测试直接改库模拟置位，
+    不绕开「M2 不置位」的实现约定。"""
+    factory = get_session_factory()
+    with factory() as session:
+        row = session.get(Receipt, receipt_id)
+        assert row is not None, "回款行不存在"
+        row.settlement_status = SETTLEMENT_DONE
+        session.commit()
+
+
+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(_RECEIPTS)
+        assert resp.status_code == 401, resp.text
+
+
+class TestCreate:
+    """创建：HK 批次契约、金额红线、回款天数派生与业务合计回写。"""
+
+    def test_新增回款_批次号格式_未指定批次_无派生(self, client: TestClient, normal: dict) -> None:
+        headers = normal["headers"]
+        business = _create_business(client, headers)
+        data = _create(client, headers, business["id"])
+
+        # 批次号格式：HK{YYYYMMDD}-{4位当日序号}（日期取当日）
+        expected_prefix = f"HK{date.today():%Y%m%d}-"
+        assert re.fullmatch(r"HK\d{8}-\d{4}", data["receipt_batch_no"]), data["receipt_batch_no"]
+        assert data["receipt_batch_no"].startswith(expected_prefix), data["receipt_batch_no"]
+
+        # 未指定批次（shipment_id 空）：批次可暂不指定，回款天数无从派生
+        assert data["shipment_id"] is None
+        assert data["shipment_no"] is None
+        assert data["payment_days"] is None
+        # M2 只写 0：置 1 由 M3 核算负责
+        assert data["settlement_status"] == 0
+        assert Decimal(data["amount"]) == Decimal("1200")
+
+    def test_指定批次_回款天数发货日首日计第1天(self, client: TestClient, normal: dict) -> None:
+        """发货 2026-10-05 / 回款 2026-11-02 → 29 天（首日计第 1 天，
+        需规「26 天」示例已作废）。"""
+        headers = normal["headers"]
+        business = _create_business(client, headers)
+        shipment = _create_shipment(client, headers, business["id"], "2026-10-05")
+        data = _create(
+            client,
+            headers,
+            business["id"],
+            shipment_id=shipment["id"],
+            receipt_date="2026-11-02",
+            amount="500.00",
+        )
+        assert data["shipment_id"] == shipment["id"]
+        assert data["shipment_no"] == shipment["shipment_no"]
+        assert data["payment_days"] == 29, data
+
+    def test_新增回款_业务合计回写_删除后回退(self, client: TestClient, normal: dict) -> None:
+        headers = normal["headers"]
+        business = _create_business(client, headers)
+        data = _create(client, headers, business["id"])
+
+        after = client.get(f"{_BUSINESSES}/{business['id']}", headers=headers).json()["data"]
+        assert Decimal(after["total_received_amount"]) == Decimal("1200")
+
+        assert client.delete(f"{_RECEIPTS}/{data['id']}", headers=headers).status_code == 200
+        rollback = client.get(f"{_BUSINESSES}/{business['id']}", headers=headers).json()["data"]
+        assert Decimal(rollback["total_received_amount"]) == Decimal("0")
+
+    def test_业务不存在_404(self, client: TestClient, normal: dict) -> None:
+        resp = client.post(_RECEIPTS, json=_rpayload(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(_RECEIPTS, json=_rpayload(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(
+            _RECEIPTS, json=_rpayload(business["id"], amount="-1"), 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(
+            _RECEIPTS,
+            json=_rpayload(business["id"], amount="12.345"),
+            headers=normal["headers"],
+        )
+        assert resp.status_code == 400, resp.text
+        assert "最多两位小数" in resp.json()["message"], resp.text
+
+
+class TestCrossBusiness:
+    """发货批次同业务强校验 —— 两类失败（不存在/跨业务）统一文案，防响应
+    差异探测发货单存在性。"""
+
+    def test_跨业务指定发货批次_400(self, client: TestClient, normal: dict) -> None:
+        headers = normal["headers"]
+        business1 = _create_business(client, headers)
+        shipment = _create_shipment(client, headers, business1["id"], str(date.today()))
+        business2 = _create_business(client, headers)
+
+        resp = client.post(
+            _RECEIPTS,
+            json=_rpayload(business2["id"], shipment_id=shipment["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:
+        """编辑路径（全量替换）同样强校验，无「未变即免检」旁路。"""
+        headers = normal["headers"]
+        business1 = _create_business(client, headers)
+        shipment = _create_shipment(client, headers, business1["id"], str(date.today()))
+        business2 = _create_business(client, headers)
+        data = _create(client, headers, business2["id"])
+
+        resp = client.put(
+            f"{_RECEIPTS}/{data['id']}",
+            json={
+                "shipment_id": shipment["id"],
+                "receipt_date": str(date.today()),
+                "amount": "1200.00",
+            },
+            headers=headers,
+        )
+        assert resp.status_code == 400, resp.text
+        assert "发货批次不属于该业务，无法指定" in resp.json()["message"], resp.text
+
+    def test_编辑回款_全量替换_派生值实时刷新(self, client: TestClient, normal: dict) -> None:
+        headers = normal["headers"]
+        business = _create_business(client, headers)
+        shipment = _create_shipment(client, headers, business["id"], "2026-10-05")
+        data = _create(client, headers, business["id"])
+        assert data["payment_days"] is None
+
+        specified = client.put(
+            f"{_RECEIPTS}/{data['id']}",
+            json={
+                "shipment_id": shipment["id"],
+                "receipt_date": "2026-11-02",
+                "amount": "300.00",
+            },
+            headers=headers,
+        )
+        assert specified.status_code == 200, specified.text
+        body = specified.json()["data"]
+        assert body["payment_days"] == 29, "指定批次后回款天数按发货日首日派生"
+        assert body["shipment_no"] == shipment["shipment_no"]
+
+        cleared = client.put(
+            f"{_RECEIPTS}/{data['id']}",
+            json={
+                "shipment_id": None,
+                "receipt_date": "2026-11-02",
+                "amount": "300.00",
+            },
+            headers=headers,
+        )
+        assert cleared.status_code == 200, cleared.text
+        assert cleared.json()["data"]["payment_days"] is None, "清除指定后派生值回 null"
+
+
+class TestSettled:
+    """已核算回款禁改禁删（M3 核算防重锚点）—— 台账固化后源数据禁漂移。"""
+
+    def test_已核算禁改_400(self, client: TestClient, normal: dict) -> None:
+        headers = normal["headers"]
+        business = _create_business(client, headers)
+        data = _create(client, headers, business["id"])
+        _settle(data["id"])
+
+        resp = client.put(
+            f"{_RECEIPTS}/{data['id']}",
+            json={
+                "shipment_id": None,
+                "receipt_date": str(date.today()),
+                "amount": "9999.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"])
+        _settle(data["id"])
+
+        resp = client.delete(f"{_RECEIPTS}/{data['id']}", headers=headers)
+        assert resp.status_code == 400, resp.text
+        assert "已核算回款禁止删除" in resp.json()["message"], resp.text
+
+
+class TestFilter:
+    """核算状态筛选（M3 核算消费的入口口径）。"""
+
+    def test_settlement_status筛选_0只回未核算_1只回已核算(
+        self, client: TestClient, normal: dict
+    ) -> None:
+        headers = normal["headers"]
+        business = _create_business(client, headers)
+        pending = _create(client, headers, business["id"], amount="500.00")
+        settled = _create(client, headers, business["id"], amount="700.00")
+        _settle(settled["id"])
+
+        resp0 = client.get(
+            _RECEIPTS,
+            params={"business_id": business["id"], "settlement_status": 0},
+            headers=headers,
+        )
+        assert resp0.status_code == 200, resp0.text
+        rows0 = resp0.json()["data"]["list"]
+        assert [r["id"] for r in rows0] == [pending["id"]], rows0
+        assert rows0[0]["settlement_status"] == 0
+
+        resp1 = client.get(
+            _RECEIPTS,
+            params={"business_id": business["id"], "settlement_status": 1},
+            headers=headers,
+        )
+        assert resp1.status_code == 200, resp1.text
+        rows1 = resp1.json()["data"]["list"]
+        assert [r["id"] for r in rows1] == [settled["id"]], rows1
+
+
+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_RECEIPT and log.action == ACTION_CREATE
+        ]
+        assert logs, "新增回款未留操作日志"
+        latest = logs[0]
+        assert latest.success == 1
+        assert "新增回款" in latest.summary
+        assert data["receipt_batch_no"] in latest.summary
+
+    def test_跨业务指定失败留审计(self, client: TestClient, normal: dict) -> None:
+        headers = normal["headers"]
+        business1 = _create_business(client, headers)
+        shipment = _create_shipment(client, headers, business1["id"], str(date.today()))
+        business2 = _create_business(client, headers)
+        resp = client.post(
+            _RECEIPTS,
+            json=_rpayload(business2["id"], shipment_id=shipment["id"]),
+            headers=headers,
+        )
+        assert resp.status_code == 400, resp.text
+        failed = [
+            log
+            for log in _logs_of(normal["username"])
+            if log.module == MODULE_RECEIPT and log.success == 0
+        ]
+        assert failed, "失败操作未留痕"
+        assert "发货批次不属于该业务" in (failed[0].error_message or ""), failed[0]

backend/tests/integration/test_shipments.py

@@ -0,0 +1,394 @@
+"""发货管理接口集成测试（需要 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
+
+pytestmark = pytest.mark.integration
+
+_BUSINESSES = "/api/v1/business/businesses"
+_SHIPMENTS = "/api/v1/business/shipments"
+_RECEIPTS = "/api/v1/business/receipts"
+_PERSONNEL = "/api/v1/base/personnel"
+_CUSTOMERS = "/api/v1/base/customers"
+
+
+def _customer(client: TestClient, headers: dict) -> dict:
+    resp = client.post(
+        _CUSTOMERS,
+        json={
+            "company_name": f"t_公司_{uuid.uuid4().hex[:8]}",
+            "owner_name": "张老板",
+            "customer_type": "personal",
+        },
+        headers=headers,
+    )
+    assert resp.status_code == 200, resp.text
+    return resp.json()["data"]
+
+
+def _bitem(**overrides: object) -> dict:
+    """业务明细行入参（发货创建的门槛：业务须有明细 ≥1 行）。"""
+    body: dict = {
+        "product_name": f"t_产品_{uuid.uuid4().hex[:8]}",
+        "product_category": "municipal_water",
+        "quantity": "100",
+        "sale_price": "100.00",
+    }
+    body.update(overrides)
+    return body
+
+
+def _bpayload(customer_id: int, **overrides: object) -> dict:
+    body: dict = {
+        "customer_id": customer_id,
+        "business_date": str(date.today()),
+        "items": [_bitem()],
+    }
+    body.update(overrides)
+    return body
+
+
+def _create_business(client: TestClient, headers: dict) -> dict:
+    """建一名 t_ 客户并挂一笔默认业务（业务须先存在，发货单挂在业务上）。"""
+    customer = _customer(client, headers)
+    resp = client.post(_BUSINESSES, json=_bpayload(customer["id"]), headers=headers)
+    assert resp.status_code == 200, resp.text
+    return resp.json()["data"]
+
+
+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]