"""审批单版本号并发防线（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]