108bf87225
补齐源快照、资料预演、正式导入及工资核算的依赖链路。 新增版本化输入、逐段 FIFO、生成锁和结构化当月调整， 并提供默认关闭的周期采集、有限重试及 Windows 托管。 原始客户资料、私有配置及首页改动不纳入本次提交。 BREAKING CHANGE: 升级前需应用新增迁移，并通过 PAYROLL_MAKER_USER_ID 绑定实际工资制作人，否则业务只读。 旧单批次核算默认关闭；历史兼容需显式启用 PAYROLL_ALLOW_LEGACY_CALCULATION。生产启用和真实金额待验收。
184 lines
8.2 KiB
Python
184 lines
8.2 KiB
Python
"""审批单版本号并发防线（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.mark.usefixtures("legacy_calculation")]
|
||
|
||
|
||
@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]
|