diff --git a/backend/README.md b/backend/README.md index 16a2af4..e647574 100644 --- a/backend/README.md +++ b/backend/README.md @@ -231,6 +231,10 @@ bcrypt 5.0 起对超过 72 字节的密码**直接抛 ValueError**（此前是 - **详情内容来源三态**：`content_source=snapshot`（送审快照）/ `live`（V11 前的历史单体无快照，只读回退当前内容并显式告警）/ `missing`（引用对象不存在且无快照）。**禁止回填历史单快照**——回填等于伪造提交时内容。 - **二次分配方案 append-only**：`business_secondary_allocations` 行不物理删除，编辑全量替换与驳回清空只把旧行 `is_active=0`；唯一键为 `(business_id, version_no, personnel_id)`，同一版本内参与人唯一、跨版本可重复。`business_repository.list_allocations(active_only=True)` 是核算与详情回显的唯一取数口径；业务详情 `allocation_history` 返回全部版本供回溯。 - **月结提交要锁行**：`settlement_service.submit` 取月结行必须 `for_update=True`——并发提交会让两次 `submit_order` 算出同一个审批单 `version_no`，送审次序失真。 +- **审批单 `version_no` 有唯一键兜底（V12）**：`uk_approval_orders_order_type_ref_version (order_type, ref_id, version_no)`——同一对象的第 N 次送审在库里有唯一一行。ORM 侧约束名必须与迁移 DDL 严格一致，否则 autogenerate 误判。 +- **取版本号必须是「当前读」（`approval_service._next_version`）**，两个禁忌：(a) 不能用普通 `MAX` 查询——请求级鉴权与业务共用同一 DbSession，鉴权的普通 SELECT 已把 REPEATABLE READ 快照锚定在请求开始时刻，父行锁只串行化写入时刻、**不刷新快照**，MAX 会读到旧值；(b) 不能写成 `select(func.max(...)).with_for_update()`——MySQL 官方明确「Locking clauses are not permitted in queries with aggregate functions or GROUP BY」，聚合上的 FOR UPDATE 语义无保证。正解：锁读该 ref 的现有版本行（`select(version_no)...with_for_update()`），最大值 Python 侧算（同 `no_generator` 套路）。 +- **版本号冲突的收敛口径**：`submit_order` 内不 catch、不 rollback（红线）。create 路径由外层编号重试循环整体重来（`business_service._NO_RETRY`）；编辑/月结提交路径由审计上下文统一回滚，再把 1062 翻译为 400「提交冲突（版本号生成重复），请重试」（同 `param_service` 版本号冲突模板）。唯一键是最后一道防线，不是取号正确性的前提。 +- **迁移前先修存量**：V11 把存量行 `version_no` 一律置 1，而 V11 前每次重提都新建行——同一 `(order_type, ref_id)` 存在多行 version_no=1，直接加唯一键必报 1062。V12 先重排「组内有重复」的分组（按 id 升序 1..n），其余分组不动（全量重排会改掉前端已展示过的历史版本号）。 - **新增审批对象类型**须同时提供快照组装函数并接入 `_content_of` 的 live 回退分支，不要在各业务服务内联 JSON 结构。 ### 工作台口径（M4 ⑧，改动前先读） diff --git a/backend/alembic/versions/7f4f7b918bc4_v12_审批单版本号唯一键.py b/backend/alembic/versions/7f4f7b918bc4_v12_审批单版本号唯一键.py new file mode 100644 index 0000000..8c2f920 --- /dev/null +++ b/backend/alembic/versions/7f4f7b918bc4_v12_审批单版本号唯一键.py @@ -0,0 +1,71 @@ +"""v12 审批单版本号唯一键 + +Revision ID: 7f4f7b918bc4 +Revises: f3a8c2d61b47 +Create Date: 2026-10-21 10:15:00.000000 + +给 ``approval_orders`` 加 ``(order_type, ref_id, version_no)`` 唯一键，兜住 +「同一引用对象的第几次送审」在并发下算重。 + +**加键前必须先重排存量**：V11 迁移用 ``ADD COLUMN version_no INT NOT NULL +DEFAULT 1`` 把存量行一律置 1，而 V11 之前的每次重提都是**新建行**（撤回 +只改旧单 status，行不删）。于是任何有 V11 前审批历史的库里，同一 +``(order_type, ref_id)`` 存在多行且 ``version_no`` 全为 1——直接 +``ADD CONSTRAINT`` 必报 1062，迁移会停在半途。 + +重排只针对「组内有重复」的分组（``COUNT(*) > COUNT(DISTINCT version_no)``）， +按 id 升序重排为 1..n；其余分组一个字节都不动——全量重排会把前端已展示 +过的历史版本号（``V{{ version_no }}``）改掉，属于用户可见的历史数据篡改。 +""" + +from collections.abc import Sequence + +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = "7f4f7b918bc4" +down_revision: str | Sequence[str] | None = "f3a8c2d61b47" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + +_UK = "uk_approval_orders_order_type_ref_version" + + +def upgrade() -> None: + """先重排存量重复的版本号，再加唯一键。""" + # 重排：只动「组内有重复版本号」的分组，按 id 升序编 1..n。 + # 用派生表内联而非临时表——同一张临时表被 JOIN 两侧引用会报 1137 + # "Can't reopen table"。 + op.execute( + "UPDATE approval_orders ao " + "JOIN (SELECT id, ROW_NUMBER() OVER (" + " PARTITION BY order_type, ref_id ORDER BY id) AS rn " + " FROM approval_orders " + " WHERE (order_type, ref_id) IN (" + " SELECT order_type, ref_id FROM approval_orders " + " GROUP BY order_type, ref_id " + " HAVING COUNT(*) > COUNT(DISTINCT version_no))) t " + "ON ao.id = t.id " + "SET ao.version_no = t.rn" + ) + # 幂等 guard：MySQL DDL 隐式提交，若加键成功而版本号未落盘， + # 重跑会报 1091（键已存在）。写法沿用 V9 先例。 + op.execute( + "SET @uk_exists := (SELECT COUNT(*) FROM information_schema.TABLE_CONSTRAINTS " + "WHERE CONSTRAINT_SCHEMA = DATABASE() AND TABLE_NAME = 'approval_orders' " + f"AND CONSTRAINT_NAME = '{_UK}')" + ) + op.execute( + "SET @ddl := IF(@uk_exists = 0, " + f"'ALTER TABLE approval_orders ADD CONSTRAINT {_UK} " + "UNIQUE (order_type, ref_id, version_no)', 'DO 0')" + ) + op.execute("PREPARE stmt FROM @ddl") + op.execute("EXECUTE stmt") + op.execute("DEALLOCATE PREPARE stmt") + + +def downgrade() -> None: + """删唯一键。唯一键本质是索引，DROP INDEX 不校验数据，存量是否已有 + 重复 version_no 与删索引无关；重排结果也不回滚（版本号保持 1..n 连续）。""" + op.execute(f"ALTER TABLE approval_orders DROP INDEX {_UK}") diff --git a/backend/app/models/approval.py b/backend/app/models/approval.py index 40261d4..cf68105 100644 --- a/backend/app/models/approval.py +++ b/backend/app/models/approval.py @@ -15,7 +15,16 @@ ref_id 为**固化无外键**（同 business_status_logs 先例）：order_type= from datetime import datetime -from sqlalchemy import JSON, DateTime, ForeignKey, Index, Integer, String, text +from sqlalchemy import ( + JSON, + DateTime, + ForeignKey, + Index, + Integer, + String, + UniqueConstraint, + text, +) from sqlalchemy.orm import Mapped, mapped_column from app.models.base import BIGINT_UNSIGNED, TINYINT_TYPE, Base, TimestampMixin, mysql_table_args @@ -54,6 +63,14 @@ class ApprovalOrder(Base, TimestampMixin): __tablename__ = "approval_orders" __table_args__ = mysql_table_args( "审批单表（营销后勤提交→财务复核→领导审核；开关关闭时自动通过留痕）", + # 同一引用对象的第 N 次送审在库里有唯一一行：并发提单算重版本号时 + # 由 DB 兜住（应用层取号走锁读，见 approval_service._next_version） + UniqueConstraint( + "order_type", + "ref_id", + "version_no", + name="uk_approval_orders_order_type_ref_version", + ), Index("idx_approval_orders_type_status", "order_type", "status"), # 版本号按 (order_type, ref_id) 递增，须有对应索引避免全表 max 扫描 Index("idx_approval_orders_type_ref", "order_type", "ref_id"), @@ -82,7 +99,8 @@ class ApprovalOrder(Base, TimestampMixin): Integer, nullable=False, server_default="1", - comment="提交版本号（同一对象类型+引用对象从1递增；重提产生新版本，旧单快照不覆盖）", + comment="提交版本号（同一对象类型+引用对象从1递增；重提产生新版本，旧单快照不覆盖；" + "同一对象内唯一，由uk_approval_orders_order_type_ref_version强制）", ) snapshot_json: Mapped[dict | None] = mapped_column( JSON, diff --git a/backend/app/services/approval_service.py b/backend/app/services/approval_service.py index b8f3e29..08b60a3 100644 --- a/backend/app/services/approval_service.py +++ b/backend/app/services/approval_service.py @@ -296,11 +296,11 @@ def act( operator_name=operator.username, ) ) - # 驳回=方案未获批准：清空该业务的二次分配行，防止被否决的方案以 - # 「当前生效行」身份留在表里、被 M3 核算快照算进工资；方案事实由 - # 审批单留痕与操作日志 request_data 承载（分摊语义落定后如需保留 - # 历史版本，在 ④ 设计时演进）。审批单的清理由调用方（act 内联） - # 委托 business_repository 完成 —— service 层互调、不触 commit。 + # 驳回=方案未获批准：把该业务的生效方案行置历史（is_active=0， + # **不删行**——V11 起 append-only 版本化），防止被否决的方案以 + # 「当前生效行」身份被 M3 核算快照算进工资；历史方案仍可在业务 + # 详情 allocation_history 回溯。委托 business_repository 完成 + # —— service 层互调、不触 commit。 if rejected and order.order_type == 1: from app.repositories import business_repository @@ -414,18 +414,34 @@ def _withdraw_pending(session: Session, *, order_type: int, ref_id: int, operato def _next_version(session: Session, order_type: int, ref_id: int) -> int: - """同一对象类型+引用对象的下一个提交版本号（从 1 递增）。""" - return ( - int( - session.execute( - select(func.coalesce(func.max(ApprovalOrder.version_no), 0)).where( - ApprovalOrder.order_type == order_type, - ApprovalOrder.ref_id == ref_id, - ) - ).scalar_one() + """同一对象类型+引用对象的下一个提交版本号（从 1 递增）。 + + **必须用锁读（当前读），不能用普通 MAX 查询**：普通 SELECT 在 + REPEATABLE READ 下是快照读，读到的是本事务首次一致读的快照；请求级 + 鉴权（deps.get_current_user 与业务共用同一 DbSession）那次 SELECT 已把 + 快照锚定在请求开始时刻，此后即使父行锁等到并发事务提交，MAX 仍停在 + 旧快照上——两个请求会算出同一个 version_no。锁读是当前读，永远读最新 + 已提交数据。 + + **不能写成 select(func.max(...)).with_for_update()**：MySQL 官方明确 + 「Locking clauses are not permitted in queries with aggregate functions + or GROUP BY」，聚合结果上的 FOR UPDATE 语义无保证。改锁「该 ref 的现有 + 版本行」，最大值在 Python 侧算——同 no_generator 的既有套路。只取 + version_no 单列，不把 snapshot_json 拉进内存。 + """ + versions = ( + session.execute( + select(ApprovalOrder.version_no) + .where( + ApprovalOrder.order_type == order_type, + ApprovalOrder.ref_id == ref_id, + ) + .with_for_update() ) - + 1 + .scalars() + .all() ) + return (max(versions) if versions else 0) + 1 def allocation_content(session: Session, business: Business) -> dict: diff --git a/backend/app/services/business_service.py b/backend/app/services/business_service.py index edaa614..a13b2fa 100644 --- a/backend/app/services/business_service.py +++ b/backend/app/services/business_service.py @@ -347,15 +347,14 @@ def _build_allocation_rows( return rows -def _plan_changed( - old_rows: list[SecondaryAllocation], new_rows: list[SecondaryAllocation] -) -> bool: +def _plan_changed(old_rows: list[SecondaryAllocation], new_rows: list[SecondaryAllocation]) -> bool: """方案内容（参与人 + 比例）是否真的变了。 比较口径是**集合**（人员 ID 与其比例的组合），与提交顺序无关——送审 快照同样按库内生效行的顺序生成，若按顺序比较，用户只是把行拖了个 序就会被判成「变了」，进而无谓地新增版本与重提审批单。 """ + def _key(row: SecondaryAllocation) -> tuple[int, Decimal]: return (row.personnel_id, Decimal(row.ratio).quantize(Decimal("0.01"))) @@ -564,14 +563,22 @@ def update_business( ref_id=business_id, snapshot=snapshot, ): - approval_service.submit_order( - session, - order_type=APPROVAL_TYPE_SECONDARY_ALLOCATION, - ref_id=business_id, - refer_no=business.business_no, - submitter=operator, - snapshot=snapshot, - ) + try: + approval_service.submit_order( + session, + order_type=APPROVAL_TYPE_SECONDARY_ALLOCATION, + ref_id=business_id, + refer_no=business.business_no, + submitter=operator, + snapshot=snapshot, + ) + except IntegrityError as exc: + # version_no 唯一键的并发冲突（取号是锁读后此项理论不可达， + # 留作 DB 兜底）：审计上下文会统一回滚，这里只把 1062 + # 翻译成可重试的业务提示，其余完整性错误原样上抛 + if _is_duplicate_entry(exc): + raise BadRequestError("提交冲突（版本号生成重复），请重试") from exc + raise else: approval_service.withdraw_pending_orders( session, diff --git a/backend/app/services/settlement_service.py b/backend/app/services/settlement_service.py index 50fed9a..8a531ed 100644 --- a/backend/app/services/settlement_service.py +++ b/backend/app/services/settlement_service.py @@ -13,11 +13,13 @@ from datetime import date from decimal import ROUND_HALF_UP, Decimal from sqlalchemy import select +from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import Session from app.core.constants import ( MODULE_SETTLEMENT, ) +from app.core.db_errors import is_duplicate_entry from app.core.exceptions import BadRequestError, NotFoundError from app.models import ( MonthlySettlement, @@ -254,16 +256,24 @@ def submit( if settlement.status not in (SETTLEMENT_STATUS_DRAFT, SETTLEMENT_STATUS_REJECTED): raise BadRequestError("仅草稿或已驳回状态可提交") - order = approval_service.submit_order( - session, - order_type=_SETTLEMENT_ORDER_TYPE, - ref_id=settlement.id, - refer_no=f"JS{settlement.settlement_month.replace('-', '')}", - submitter=operator, - # 月结明细在生成时已固化；提交时快照让复核者看到的是送审版本， - # 而不是审批期间若发生重生成后的内容 - snapshot=approval_service.settlement_content(session, settlement.id), - ) + try: + order = approval_service.submit_order( + session, + order_type=_SETTLEMENT_ORDER_TYPE, + ref_id=settlement.id, + refer_no=f"JS{settlement.settlement_month.replace('-', '')}", + submitter=operator, + # 月结明细在生成时已固化；提交时快照让复核者看到的是送审版本， + # 而不是审批期间若发生重生成后的内容 + snapshot=approval_service.settlement_content(session, settlement.id), + ) + except IntegrityError as exc: + # version_no 唯一键的并发冲突（提交前已锁月结行，此项理论不可达， + # 留作 DB 兜底）：审计上下文会统一回滚，这里只把 1062 翻译成 + # 可重试的业务提示，其余完整性错误原样上抛 + if is_duplicate_entry(exc): + raise BadRequestError("提交冲突（版本号生成重复），请重试") from exc + raise settlement.approval_order_no = order.order_no if order.status == SETTLEMENT_STATUS_APPROVED: mark_locked(session, settlement, operator=operator) diff --git a/backend/tests/README.md b/backend/tests/README.md index d66bf22..7b96360 100644 --- a/backend/tests/README.md +++ b/backend/tests/README.md @@ -169,7 +169,14 @@ uv run pytest tests/integration -q # 仅集成 | `tests/integration/test_secondary_allocation.py`（扩） | **同一会话内旧方案行随置历史同步刷新**（Core UPDATE 的会话同步策略退化为不刷新时，行对象会停留在 is_active=1 的过期内存值） | +1 | | `tests/integration/test_approval_flow.py`（扩） | **送审内容未变不重提**：方案与金额都没动 → 方案不新增版本、审批单不重提（旧单仍待财务复核）；**只改金额** → 方案版本不变但审批单 version_no+1 且新快照金额为改后值；**改方案比例** → 方案版本+1、审批单 version_no+1 且旧单撤回；**无在途单时保存仍提新单**（保守规则）；`pending_matches_snapshot` 四态（一致/不一致/无在途单/快照为 None） | +5 | -运行：`uv run pytest tests/integration/test_approval_flow.py tests/integration/test_secondary_allocation.py tests/integration/test_schema_sync.py -q` → **37 passed**（2026-10-20）；`tests/integration/test_settlements.py` → **16 passed**；`tests/integration/test_calc.py` → **20 passed**；`tests/unit` → **525 passed**（含新增 `test_approval_helpers.py` 10 例）；**全量 `uv run pytest -q` → 891 passed（10:23）**。ruff、前端 `pnpm type-check` 与 `pnpm build` 均绿。 +### 审批单版本号并发防线（2026-10-21 完成：V12 唯一键 + 锁读取号） + +| 文件 | 守什么 | 用例数 | +| --- | --- | --- | +| `tests/integration/test_approval_version.py`（新） | **取号读最新已提交值**（两会话法：s1 锚定快照 + s2 提交 version_no=2 → s1 必须取到 3，守住「快照读 vs 当前读」根因）；**取号查询带 FOR UPDATE 且不含聚合**（事件钩子抓真实 SQL——MySQL 官方不允许聚合函数携带锁子句）；**唯一键生效**（重复 (order_type, ref_id, version_no) 直插撞 1062）；**编辑路径冲突收敛为 400** 且业务数据不写坏（monkeypatch `_next_version` 恒返回 1）；**正常路径版本号仍递增**（1→2→3） | +5 | +| `tests/unit/test_models.py`（扩） | 审批单唯一键覆盖 (order_type, ref_id, version_no) 三列且约束名与迁移 DDL 一致 | +1 | + +运行（2026-10-22，容器恢复后）：`tests/integration/test_approval_version.py` + `test_dashboard.py` → **16 passed**；全量 `uv run pytest -q` → **897 passed（12:02）**；`uv run ruff check .` 绿、前端 `pnpm type-check` 绿。首次全量跑出两处**测试代码**适配（非产品缺陷）：① 工作台 `_insert_approval` 用同一 `(order_type=2, ref_id=0)` 连插两张单，撞 V12 新唯一键 → 每张单改给不同 `ref_id`；② 新用例跨模块引用 `switched_on` 夹具（pytest 夹具只在定义模块与 conftest 可见，报 fixture not found）→ 用 `_set_switch` 本地重建。 ## 三、运行记录 @@ -217,6 +224,9 @@ uv run pytest tests/integration -q # 仅集成 | 2026-10-20 | 每次保存带方案的业务都刷出版本与审批留痕 | 只改备注、方案一字未动，也会「置历史+新增版本+撤回旧单+提新单」，方案历史与审批留痕被噪音淹没，两个 `version_no` 语义混同 | 判据统一为**送审内容**：方案未变不新增版本行；在途单送审内容一致则不重提（`pending_matches_snapshot`，只比 `data`） | | 2026-10-20 | 三处列表查询整列拉快照 | 审批单列表、工作台待办、审批超时预警都用 `select(ApprovalOrder)` 取全列，快照 JSON 可达上百 KB，翻页与工作台负担随快照体积线性放大 | 三处加 `defer(ApprovalOrder.snapshot_json)`（沿用操作日志列表排除 `request_data` 的既有模式） | | 2026-10-20 | 月结提交未锁行 | 并发/双击提交同一月结时两次 `_next_version` 可能算出同一 `version_no`，送审次序失真 | `settlement_repository.get_by_id` 加 `for_update` 参数，提交路径锁行 | +| 2026-10-21 | 审批单 `version_no` 无唯一约束 | 应用层取号即使正确，并发下 DB 层无兜底；且唯一键是「一个对象至多一张在途单」红线之外版本语义的最后防线 | V12 迁移加 `uk_approval_orders_order_type_ref_version (order_type, ref_id, version_no)`；加键前先重排存量重复分组（V11 把存量一律置 1 而 V11 前重提是新建行，直接加键必报 1062） | +| 2026-10-21 | `_next_version` 是快照读 | 请求级鉴权与业务共用同一 DbSession，鉴权的普通 SELECT 已锚定 REPEATABLE READ 快照——父行锁只串行化写入时刻、**不刷新快照**，并发提单的 MAX 停在请求开始时刻 → 版本号重复 | 取号改锁读（`select(version_no)...with_for_update()`，当前读），最大值 Python 侧算；**不能用聚合 + FOR UPDATE**——MySQL 官方不允许聚合函数携带锁子句 | +| 2026-10-21 | `act()` 驳回清空方案的注释与行为相反 | 注释写「清空该业务的二次分配行…如需保留历史版本，在 ④ 设计时演进」，而 V11 已实现 append-only（只置 `is_active=0` 行不删）——误导后来者按旧语义改代码 | 注释改为「置历史不删行，历史方案可回溯」 | | 2026-10-19 | 前端报表五处状态缺陷 | 打印超限判定恒假（静默截断）、部门下钻 query 未读取、回款多桶共用一个 drill 状态（面板串行）、summary 在请求序号守卫前写入（合计串号）、e2e 全页行数选择器必失败 | 分别修正判定/读取 query/按桶独立状态+数组参数/`onData` 回调/限定主桶表；补 e2e 多桶展开、下钻、截断、部门打印用例 | | 2026-10-19 | 工作台权限收紧后守卫仍回退 `/dashboard` | 普通用户登录/直连受限页被反复重定向到同一受限页 → **无法进入系统**（导航被判定无限重定向） | `router/access.ts` 统一准入与落地页解析（工作台 → 业务数据 → 回款）；登录回跳经 `canAccessPath` 校验；菜单/面包屑复用同一判定 | | 2026-10-19 | 部门汇总应发只计一次全公司 5% | 月结每位有聚合行的经理各计一次 5%，签字稿少计 N-1 份；**导出/打印金额小于月结应发** | 团队行增 `has_aggregate/aggregate_amount`（优先冻结聚合行）；对账改按聚合额求和，新增「聚合行舍入差」吸收分列 q2 尾差；补「与 `total_payable` 对拍」用例 | diff --git a/backend/tests/integration/test_approval_version.py b/backend/tests/integration/test_approval_version.py new file mode 100644 index 0000000..b5281d4 --- /dev/null +++ b/backend/tests/integration/test_approval_version.py @@ -0,0 +1,183 @@ +"""审批单版本号并发防线（V12 唯一键 + 锁读取号）。 + +守的是： +- **取号必须读最新已提交值**：请求级鉴权与业务共用同一 DbSession，鉴权 + 的普通 SELECT 已把 REPEATABLE READ 快照锚定在请求开始时刻；父行锁只 + 串行化写入时刻、不刷新快照——取号若是快照读，并发提单会算出同一个 + version_no（两会话法做行为级回归锁，不依赖真多线程） +- **取号查询带 FOR UPDATE 且不用聚合**：MySQL 官方不允许聚合函数携带 + 锁子句（Locking clauses are not permitted in queries with aggregate + functions or GROUP BY），聚合上的 FOR UPDATE 语义无保证 +- **DB 层唯一键生效**：即使应用层取号缺陷被绕过，重复 + (order_type, ref_id, version_no) 也必须在库上被拒（1062） +- **冲突收敛为 400**：编辑路径撞唯一键 → 审计上下文回滚 → 业务提示 + 可重试，业务数据不得写坏 +- **正常路径版本号仍递增**：锁读改写不破坏既有 1→2→3 行为 +""" + +import uuid + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy import event, text + +from app.core.db import get_engine, get_session_factory +from app.core.db_errors import is_duplicate_entry +from app.models import ApprovalOrder +from tests.integration.factories import business_payload +from tests.integration.test_approval_flow import ( + _BUSINESSES, + _create_with_plan, + _latest_order, + _person, + _set_switch, +) + +pytestmark = pytest.mark.integration + + +@pytest.fixture +def switched_on(client: TestClient, admin: dict) -> None: + """把二次分配审批开关置 1（用例结束由 conftest 恢复种子基线）。 + + pytest 夹具只在定义它的模块与 conftest 中可见，跨模块直接引用会报 + ``fixture 'switched_on' not found``；这里用 ``_set_switch`` 本地重建。 + """ + _set_switch(client, admin["headers"], "1") + + +class TestNextVersionLockRead: + """取号查询的锁读口径（真实 SQL 形态 + 行为级快照回归）。""" + + def test_取号查询带FOR_UPDATE_不含聚合(self, client: TestClient, admin: dict) -> None: + """抓 _next_version 真实发出的 SQL（比本地重造语句再编译更贴真）： + 必须含 FOR UPDATE、不含 MAX( —— 聚合上的锁子句 MySQL 不保证语义。""" + + captured: list[str] = [] + + def _spy(_conn, _cursor, statement, *_rest): + if "approval_orders" in statement and "version_no" in statement: + captured.append(statement) + + engine = get_engine() + event.listen(engine, "before_cursor_execute", _spy) + try: + p = _person(client, admin["headers"]) + _create_with_plan(client, admin["headers"], p["id"]) # 触发一次真实提单 + finally: + event.remove(engine, "before_cursor_execute", _spy) + + assert captured, "未捕获到取号查询" + assert any("FOR UPDATE" in sql.upper() for sql in captured), captured + assert not any("MAX(" in sql.upper() for sql in captured), captured + + def test_版本号读最新已提交值_不受事务旧快照影响(self) -> None: + """两会话法（无需线程）：s1 先锚定快照 → s2 提交新版本 → + s1 取号必须读到 3 而非旧快照里的 2——守住「快照读 vs 当前读」根因。""" + from app.services import approval_service + + factory = get_session_factory() + with factory() as s1: + # 锚定快照（模拟请求级鉴权读先于业务写发生） + s1.execute(text("SELECT order_no FROM approval_orders LIMIT 1")).all() + + with factory() as s2: + order = ApprovalOrder( + order_no=f"AP-{uuid.uuid4().hex[:12]}", + order_type=1, + ref_id=987654321, + refer_no="YW-SNAPSHOT-TEST", + version_no=2, + status=5, + submitter_name="t_snapshot_probe", + ) + s2.add(order) + s2.commit() + + # s1 的普通读仍停在旧快照（该分组在快照里不存在） + stale = s1.execute( + text( + "SELECT COALESCE(MAX(version_no),0) FROM approval_orders " + "WHERE order_type=1 AND ref_id=987654321" + ) + ).scalar_one() + assert stale == 0, "前置失效：快照读应停在锚定时刻，先修隔离级别再跑本用例" + + assert approval_service._next_version(s1, 1, 987654321) == 3, ( + "取号必须读到最新已提交值（2+1=3），读到 2 说明退化为快照读" + ) + s1.rollback() + + def test_唯一键生效_重复版本号被拒(self) -> None: + """重复 (order_type, ref_id, version_no) 直插必须撞 1062—— + 应用层取号即使被绕过，DB 也是最后一道防线。""" + from sqlalchemy.exc import IntegrityError + + factory = get_session_factory() + with factory() as s: + common = { + "order_type": 1, + "ref_id": 987654322, + "refer_no": "YW-UK-TEST", + "status": 5, + "submitter_name": "t_uk_probe", + } + s.add(ApprovalOrder(order_no=f"AP-{uuid.uuid4().hex[:12]}", version_no=1, **common)) + s.flush() + s.add(ApprovalOrder(order_no=f"AP-{uuid.uuid4().hex[:12]}", version_no=1, **common)) + with pytest.raises(IntegrityError) as exc_info: + s.flush() + assert is_duplicate_entry(exc_info.value), "应为唯一键冲突（1062）而非其他完整性错误" + s.rollback() + + def test_正常路径版本号仍递增(self, client: TestClient, admin: dict, switched_on) -> None: + """常规编辑两次：送审版本号 1→2→3——锁读改写不得破坏既有行为。""" + p = _person(client, admin["headers"]) + data = _create_with_plan(client, admin["headers"], p["id"]) + + first = _latest_order(client, admin["headers"], data["business_no"]) + assert first["version_no"] == 1 + + for ratio in ("60.00", "70.00"): + detail = client.get(f"{_BUSINESSES}/{data['id']}", headers=admin["headers"]).json()[ + "data" + ] + payload = business_payload( + detail["customer_id"], + secondary_allocations=[{"personnel_id": p["id"], "ratio": ratio}], + ) + resp = client.put(f"{_BUSINESSES}/{data['id']}", json=payload, headers=admin["headers"]) + assert resp.status_code == 200, resp.text + + third = _latest_order(client, admin["headers"], data["business_no"]) + assert third["version_no"] == 3 + + +class TestConflictConvergence: + """编辑路径撞版本号唯一键 → 收敛为可重试的 400，业务数据不写坏。""" + + def test_编辑路径版本号冲突_收敛为400( + self, client: TestClient, admin: dict, switched_on, monkeypatch + ) -> None: + from app.services import approval_service + + p = _person(client, admin["headers"]) + data = _create_with_plan(client, admin["headers"], p["id"]) + before = client.get(f"{_BUSINESSES}/{data['id']}", headers=admin["headers"]).json()["data"] + + # 模拟取号读到过期快照：恒返回 1（库中 V1 已存在）→ 撞唯一键 + monkeypatch.setattr(approval_service, "_next_version", lambda *a, **k: 1) + + payload = business_payload( + before["customer_id"], + secondary_allocations=[{"personnel_id": p["id"], "ratio": "80.00"}], + ) + resp = client.put(f"{_BUSINESSES}/{data['id']}", json=payload, headers=admin["headers"]) + assert resp.status_code == 400, resp.text + assert "版本号生成重复" in resp.json()["message"] + + monkeypatch.undo() + # 业务数据不得写坏：整个事务回滚，方案仍是 50.00（V1 生效行） + after = client.get(f"{_BUSINESSES}/{data['id']}", headers=admin["headers"]).json()["data"] + assert [a["ratio"] for a in after["secondary_allocations"]] == ["50.00"] + assert [h["version_no"] for h in after["allocation_history"]] == [1] diff --git a/backend/tests/integration/test_dashboard.py b/backend/tests/integration/test_dashboard.py index 67e6d47..eb7b509 100644 --- a/backend/tests/integration/test_dashboard.py +++ b/backend/tests/integration/test_dashboard.py @@ -250,7 +250,13 @@ def leader(make_user, auth_headers) -> dict: def _insert_approval(status: int) -> None: - """直插在途审批单（避免依赖两次流转，专注工作台过滤口径）。""" + """直插在途审批单（避免依赖两次流转，专注工作台过滤口径）。 + + ``ref_id`` 仅作对象分组锚点、本用例不校验：**每张单必须给不同的 + ref_id**——V12 的 ``uk_approval_orders_order_type_ref_version`` 要求 + (order_type, ref_id, version_no) 唯一，同一假对象连插两张 version_no=1 + 会撞 1062。 + """ from app.core.db import get_session_factory from app.models import ApprovalOrder @@ -260,7 +266,7 @@ def _insert_approval(status: int) -> None: ApprovalOrder( order_no=f"AP{uuid.uuid4().hex[:12]}", order_type=2, - ref_id=0, + ref_id=status, refer_no="t_工作台", status=status, submitter_name="t_admin", diff --git a/backend/tests/unit/test_models.py b/backend/tests/unit/test_models.py index 51f06d9..056e06f 100644 --- a/backend/tests/unit/test_models.py +++ b/backend/tests/unit/test_models.py @@ -96,6 +96,20 @@ class TestNamingConvention: for fk in table.foreign_keys: assert fk.constraint.name.startswith(f"fk_{table.name}_"), fk.constraint.name + def test_审批单版本号唯一键覆盖三列(self) -> None: + """并发提单算重版本号时由 DB 兜底（应用层取号走锁读，见 + approval_service._next_version）；约束名必须与迁移 DDL 严格一致， + 否则 alembic autogenerate 会误判成「约束缺失/多余」。""" + table = Base.metadata.tables["approval_orders"] + uks = [c for c in table.constraints if c.__class__.__name__ == "UniqueConstraint"] + hit = [ + c + for c in uks + if {col.name for col in c.columns} == {"order_type", "ref_id", "version_no"} + ] + assert hit, "审批单缺 (order_type, ref_id, version_no) 唯一键" + assert hit[0].name == "uk_approval_orders_order_type_ref_version" + class TestComments: """规范红线要求中文 COMMENT。""" diff --git a/db/init/init-dev.sql b/db/init/init-dev.sql index 66ab2c8..40dde34 100644 --- a/db/init/init-dev.sql +++ b/db/init/init-dev.sql @@ -722,6 +722,7 @@ CREATE TABLE approval_orders ( update_time DATETIME NOT NULL COMMENT '更新时间' DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, CONSTRAINT pk_approval_orders PRIMARY KEY (id), CONSTRAINT uk_approval_orders_order_no UNIQUE (order_no), + CONSTRAINT uk_approval_orders_order_type_ref_version UNIQUE (order_type, ref_id, version_no), CONSTRAINT fk_approval_orders_submitter_id FOREIGN KEY(submitter_id) REFERENCES users (id) ON DELETE SET NULL )CHARSET=utf8mb4 ENGINE=InnoDB COMMENT='审批单表（营销后勤提交→财务复核→领导审核；开关关闭时自动通过留痕）' COLLATE utf8mb4_general_ci; @@ -890,4 +891,4 @@ CREATE TABLE alembic_version ( CONSTRAINT alembic_version_pkc PRIMARY KEY (version_num) )CHARSET=utf8mb4 ENGINE=InnoDB COLLATE utf8mb4_general_ci; -INSERT INTO alembic_version (version_num) VALUES ('f3a8c2d61b47'); +INSERT INTO alembic_version (version_num) VALUES ('7f4f7b918bc4'); diff --git a/db/migration/V12__approval_order_version_unique.sql b/db/migration/V12__approval_order_version_unique.sql new file mode 100644 index 0000000..a745fa9 --- /dev/null +++ b/db/migration/V12__approval_order_version_unique.sql @@ -0,0 +1,16 @@ +-- Running upgrade f3a8c2d61b47 -> 7f4f7b918bc4 + +UPDATE approval_orders ao JOIN (SELECT id, ROW_NUMBER() OVER ( PARTITION BY order_type, ref_id ORDER BY id) AS rn FROM approval_orders WHERE (order_type, ref_id) IN ( SELECT order_type, ref_id FROM approval_orders GROUP BY order_type, ref_id HAVING COUNT(*) > COUNT(DISTINCT version_no))) t ON ao.id = t.id SET ao.version_no = t.rn; + +SET @uk_exists := (SELECT COUNT(*) FROM information_schema.TABLE_CONSTRAINTS WHERE CONSTRAINT_SCHEMA = DATABASE() AND TABLE_NAME = 'approval_orders' AND CONSTRAINT_NAME = 'uk_approval_orders_order_type_ref_version'); + +SET @ddl := IF(@uk_exists = 0, 'ALTER TABLE approval_orders ADD CONSTRAINT uk_approval_orders_order_type_ref_version UNIQUE (order_type, ref_id, version_no)', 'DO 0'); + +PREPARE stmt FROM @ddl; + +EXECUTE stmt; + +DEALLOCATE PREPARE stmt; + +UPDATE alembic_version SET version_num='7f4f7b918bc4' WHERE alembic_version.version_num = 'f3a8c2d61b47'; +