PayrollManagementSystem/backend/tests/integration/test_approval_version.py

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