From ca4cc72eea9259d485139143d2998023a10129ed Mon Sep 17 00:00:00 2001 From: liuwenxin <1009266660@qq.com> Date: Tue, 13 Oct 2026 13:39:09 +0800 Subject: [PATCH] =?UTF-8?q?test(business):=20=E4=B8=9A=E5=8A=A1=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E7=AE=A1=E7=90=86=E6=A8=A1=E5=9D=97=E9=9B=86=E6=88=90?= =?UTF-8?q?=E7=94=A8=E4=BE=8B=EF=BC=8870=20=E4=BE=8B=EF=BC=89=E4=B8=8E?= =?UTF-8?q?=E4=B8=A4=E5=A4=84=E7=BC=BA=E9=99=B7=E4=BF=AE=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 覆盖：编号格式与序号递增/状态机终态拦截/三处汇总冗余一致性/跨业务批次 400/已核算禁改禁删/毛利润金标准与防除零/asOf 回溯/评级重叠/审计双记录 - 缺陷修复①：change_status 留痕 old_status 取到赋值后属性（时间线失真 3→3） - 缺陷修复②：freight_amount 漏金额校验（负数/三位小数可落库污染运费合计） - conftest 扩展业务六表清理（编号前缀 + FK 逆序收口）与评级种子基线恢复 - 全量回归 603 passed（单元 381 / 集成 222） Co-Authored-By: Claude Code --- backend/app/schemas/shipment.py | 12 + backend/app/services/business_service.py | 7 +- backend/tests/README.md | 21 +- backend/tests/integration/conftest.py | 61 ++ backend/tests/integration/test_businesses.py | 811 +++++++++++++++++++ backend/tests/integration/test_ratings.py | 294 +++++++ backend/tests/integration/test_receipts.py | 422 ++++++++++ backend/tests/integration/test_shipments.py | 394 +++++++++ 8 files changed, 2017 insertions(+), 5 deletions(-) create mode 100644 backend/tests/integration/test_businesses.py create mode 100644 backend/tests/integration/test_ratings.py create mode 100644 backend/tests/integration/test_receipts.py create mode 100644 backend/tests/integration/test_shipments.py diff --git a/backend/app/schemas/shipment.py b/backend/app/schemas/shipment.py index 61253f1..9ce2377 100644 --- a/backend/app/schemas/shipment.py +++ b/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 封装同名对应）。 diff --git a/backend/app/services/business_service.py b/backend/app/services/business_service.py index 0c92e1d..4a49f2d 100644 --- a/backend/app/services/business_service.py +++ b/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 diff --git a/backend/tests/README.md b/backend/tests/README.md index 8cbd5d3..5620916 100644 --- a/backend/tests/README.md +++ b/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 补测 | -| 业务数据管理 | 进行中 | 集成用例在 T11 统一编写（28 项裁决见 `docs/问题与解决方案对照表_v2.xlsx`） | +| 基础数据管理 | ✅ 已完成 | 2026-10-13 完成（48 例，见上表）；「被业务数据引用」的删除保护已随 T11 的业务/发货禁删用例覆盖 | +| 业务数据管理 | ✅ 后端已完成 | 2026-10-13 完成（70 例，见上表）；前端页面与绩效核算联动（M3 台账核销回写 settlement_status）待 M3 补测 | | 绩效工资核算 | 未开始 | 需求已澄清；金额精度需专项用例（`Fraction` 与 1/90 日斜率已在 M1 公式层覆盖） | | 报表统计 | 未开始 | 依赖报表口径确认 | diff --git a/backend/tests/integration/conftest.py b/backend/tests/integration/conftest.py index 6323f94..52e3b6b 100644 --- a/backend/tests/integration/conftest.py +++ b/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 自引用、清子表）→ 产品 diff --git a/backend/tests/integration/test_businesses.py b/backend/tests/integration/test_businesses.py new file mode 100644 index 0000000..6c5b423 --- /dev/null +++ b/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] diff --git a/backend/tests/integration/test_ratings.py b/backend/tests/integration/test_ratings.py new file mode 100644 index 0000000..2f598c1 --- /dev/null +++ b/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] diff --git a/backend/tests/integration/test_receipts.py b/backend/tests/integration/test_receipts.py new file mode 100644 index 0000000..5d6e47e --- /dev/null +++ b/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] diff --git a/backend/tests/integration/test_shipments.py b/backend/tests/integration/test_shipments.py new file mode 100644 index 0000000..d9db666 --- /dev/null +++ b/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]