group_xinghuo_jinrong/tests/test_convert_core.py

"""T-6 `convert_core_repository.apply_convert` 单测（开发计划 §6.1 DoD）。

sqlite 内存库（`conftest.sqlite_engine` 单一事实源建表），数据自建。

**金额一律由生产 `calc.py` / `fee.py` 算出，不手算**（自检第 13 问：
同一规则只留一个副本 —— 测试里手算一遍等于给公式造第二个副本，
生产改了测试还在"绿"，是假证据）。

覆盖：
1. 正常路径：2 条流水（R-b：`redeem`/`subscribe`、同组）+ N 条明细 + 两端持仓
2. 冲突路径：`remain_qty < 本次扣减` → `LotConflict`(409) 且**无残留**
3. 回滚路径：明细写入注入异常 → `core_trade`/`core_share_lot`/`core_holding` **全无残留**
4. 转出端归零**保留行**；转入端首次 INSERT、再次 UPDATE（累加）
5. R-a 并发首次建行：另一笔先落行 → 本笔 INSERT 撞 UNIQUE → 回退增量 UPDATE
   → 只有一行、金额为两笔之和、本笔不报错
6. 死锁 1213 自动重试（2026-09-10 用户拍板 · T-13 后补）：整事务重试、
   重试预算耗尽上抛、非 1213（1205）不重试
"""

from __future__ import annotations

from datetime import date, datetime
from decimal import ROUND_HALF_UP, Decimal

import pytest
from sqlalchemy import text
from sqlalchemy.exc import OperationalError

from app.gateway.convert_core_repository import (
    ConvertApplyInput,
    ConvertCoreRepository,
    LotCharge,
)
from app.service.convert.calc import (
    convert_amount,
    diff_fee,
    hold_days,
    in_qty,
    lot_amount,
    lot_fee,
    plan_lots,
)
from app.service.convert.errors import LotConflict
from app.service.convert.fee import pick_fee_rate
from app.service.convert.types import FeeRule, Lot

CUSTOMER = "CUST-T6"
PROD_OUT = "PROD-T6A"  # 转出（债券型，申购费率 0.0030）
PROD_IN = "PROD-T6B"  # 转入（股票型，申购费率 0.0080）
TRADE_AT = datetime(2026, 9, 4, 10, 0, 0)
TRADE_DATE = date(2026, 9, 4)
OUT_RATE = Decimal("0.0030")
IN_RATE = Decimal("0.0080")
IN_NAV = Decimal("0.9500")

# 赎回费五档（与 07-seed-fee-rule.sql、fee.py docstring 三处一致）
FEE_TIERS = [
    (0, 7, "0.0150"),
    (7, 30, "0.0100"),
    (30, 180, "0.0050"),
    (180, 365, "0.0025"),
    (365, None, "0.0000"),
]


def _dec(value: object, places: str = "0.01") -> Decimal:
    """DB 读回值（sqlite float / MySQL Decimal）→ 统一 2 位 Decimal 再比较。"""
    return Decimal(str(value)).quantize(Decimal(places), rounding=ROUND_HALF_UP)


# ── 种子 ────────────────────────────────────────────────────────────
def _seed_base(engine) -> None:
    with engine.begin() as conn:
        conn.execute(
            text("INSERT INTO core_customer (customer_id, display_name) VALUES (:c, 'T6')"),
            {"c": CUSTOMER},
        )
        conn.execute(
            text(
                "INSERT INTO core_product (product_id, product_name, min_risk_code, "
                "product_type, can_subscribe, can_redeem, subscribe_fee_rate) "
                "VALUES (:p, '转出基金', 'R2', 'bond', 1, 1, :r)"
            ),
            {"p": PROD_OUT, "r": float(OUT_RATE)},
        )
        conn.execute(
            text(
                "INSERT INTO core_product (product_id, product_name, min_risk_code, "
                "product_type, can_subscribe, can_redeem, subscribe_fee_rate) "
                "VALUES (:p, '转入基金', 'R4', 'stock', 1, 1, :r)"
            ),
            {"p": PROD_IN, "r": float(IN_RATE)},
        )
        for mh, mh_max, rate in FEE_TIERS:
            conn.execute(
                text(
                    "INSERT INTO core_fee_rule (product_id, fee_type, min_hold_days, "
                    "max_hold_days, rate) VALUES (:p, 'redeem', :mh, :mh_max, :rate)"
                ),
                {"p": PROD_OUT, "mh": mh, "mh_max": mh_max, "rate": float(rate)},
            )
        conn.execute(
            text(
                "INSERT INTO core_product_nav (product_id, nav, daily_chg_pct, nav_date) "
                "VALUES (:p, :nav, 0, :d)"
            ),
            {"p": PROD_IN, "nav": float(IN_NAV), "d": TRADE_DATE},
        )


def _seed_lot(engine, lot_id: str, qty: str, nav: str, confirmed_at: datetime) -> None:
    with engine.begin() as conn:
        conn.execute(
            text(
                "INSERT INTO core_share_lot (lot_id, customer_id, product_id, qty, "
                "remain_qty, nav, confirmed_at) VALUES (:l, :c, :p, :q, :q, :nav, :cat)"
            ),
            {
                "l": lot_id, "c": CUSTOMER, "p": PROD_OUT,
                "q": float(qty), "nav": float(nav), "cat": confirmed_at,
            },
        )


def _seed_holding(engine, pid: str, qty: str, cost: str, mv: str) -> None:
    with engine.begin() as conn:
        conn.execute(
            text(
                "INSERT INTO core_holding (customer_id, product_id, qty, cost_amount, "
                "market_value, pnl_pct, as_of) VALUES (:c, :p, :q, :cost, :mv, 0, :d)"
            ),
            {
                "c": CUSTOMER, "p": pid, "q": float(qty),
                "cost": float(cost), "mv": float(mv), "d": TRADE_DATE,
            },
        )


def _read_lots(engine) -> list[Lot]:
    from app.repository.core_ro import CoreReadOnlyRepository

    return [
        Lot.from_row(r)
        for r in CoreReadOnlyRepository(engine=engine).list_share_lots(CUSTOMER, PROD_OUT)
    ]


def _read_rules(engine) -> list[FeeRule]:
    from app.repository.core_ro import CoreReadOnlyRepository

    return [
        FeeRule.from_row(r)
        for r in CoreReadOnlyRepository(engine=engine).get_redeem_fee_rules(PROD_OUT)
    ]


# ── 折算（全部走生产纯函数，与 T-7 的调用姿势一致）──────────────────────
def _build_input(
    engine,
    *,
    group_id: str,
    requested: str,
    in_lot_id: str = "LOT-IN-1",
    out_trade_id: str | None = None,
    in_trade_id: str | None = None,
) -> ConvertApplyInput:
    """跑一遍「选批 → 逐批计费 → 折算」，构造阶段一入参。"""
    lots = _read_lots(engine)
    rules = _read_rules(engine)
    plan = plan_lots(lots, Decimal(requested))

    charges: list[LotCharge] = []
    for alloc in plan.allocations:
        days = hold_days(TRADE_DATE, alloc.confirmed_at)
        rate = pick_fee_rate(rules, days, product_id=PROD_OUT)
        amount = lot_amount(alloc.qty, alloc.nav)
        charges.append(
            LotCharge(
                lot_id=alloc.lot_id,
                qty=alloc.qty,
                hold_days=days,
                amount=amount,
                fee_rate=rate,
                fee_amount=lot_fee(amount, rate),
                nav=alloc.nav,
                nav_date=TRADE_DATE,
            )
        )

    out_amount = sum((c.amount for c in charges), Decimal("0"))
    redeem_fee = sum((c.fee_amount for c in charges), Decimal("0"))
    conv = convert_amount(out_amount, redeem_fee)
    gap = diff_fee(conv, OUT_RATE, IN_RATE, "amount_diff")
    in_amount = conv - gap
    shares = in_qty(in_amount, IN_NAV)

    gid = group_id
    return ConvertApplyInput(
        convert_group_id=gid,
        out_trade_id=out_trade_id or f"{gid}-OUT",
        in_trade_id=in_trade_id or f"{gid}-IN",
        customer_id=CUSTOMER,
        from_product_id=PROD_OUT,
        to_product_id=PROD_IN,
        traded_at=TRADE_AT,
        out_qty=plan.actual_qty,
        out_amount=out_amount,
        in_qty=shares,
        in_amount=in_amount,
        in_nav=IN_NAV,
        in_nav_date=TRADE_DATE,
        in_lot_id=in_lot_id,
        in_confirmed_at=datetime(2026, 9, 5, 10, 0, 0),  # T+1（D14）
        charges=tuple(charges),
    )


def _rows(engine, sql: str, **params) -> list[dict]:
    with engine.connect() as conn:
        return [dict(r) for r in conn.execute(text(sql), params).mappings()]


def _holding(engine, pid: str) -> dict | None:
    rows = _rows(
        engine,
        "SELECT * FROM core_holding WHERE customer_id = :c AND product_id = :p",
        c=CUSTOMER, p=pid,
    )
    return rows[0] if rows else None


# ── 1. 正常路径 ─────────────────────────────────────────────────────
def test_apply_convert_happy_path_writes_trades_details_and_holdings(sqlite_engine):
    _seed_base(sqlite_engine)
    # 持有 34 天（30–180 档 0.0050）与 3 天（<7 档 0.0150）两个批次
    _seed_lot(sqlite_engine, "LOT-A1", "100", "1.0300", datetime(2026, 8, 1, 10, 0, 0))
    _seed_lot(sqlite_engine, "LOT-A2", "50", "1.0000", datetime(2026, 9, 1, 10, 0, 0))
    _seed_holding(sqlite_engine, PROD_OUT, "150", "150.00", "154.50")

    req = _build_input(sqlite_engine, group_id="CNV-T6-1", requested="120")
    ConvertCoreRepository(engine=sqlite_engine).apply_convert(req)

    # ① 两条流水：R-b 必须是 redeem / subscribe（**不是 convert**），同组
    trades = _rows(
        sqlite_engine,
        "SELECT * FROM core_trade WHERE convert_group_id = :g ORDER BY trade_type",
        g="CNV-T6-1",
    )
    assert len(trades) == 2
    assert [t["trade_type"] for t in trades] == ["redeem", "subscribe"]
    assert {t["product_id"] for t in trades} == {PROD_OUT, PROD_IN}
    out_trade = next(t for t in trades if t["trade_type"] == "redeem")
    in_trade = next(t for t in trades if t["trade_type"] == "subscribe")
    assert _dec(out_trade["amount"]) == _dec(req.out_amount)
    assert _dec(in_trade["amount"]) == _dec(req.in_amount)
    assert _dec(out_trade["qty"]) == _dec(req.out_qty)
    assert _dec(in_trade["qty"]) == _dec(req.in_qty)

    # ② 批次：LOT-A1 扣满归零但**保留行**，LOT-A2 剩 30，转入新批次已建
    remain = {
        r["lot_id"]: _dec(r["remain_qty"])
        for r in _rows(
            sqlite_engine,
            "SELECT lot_id, remain_qty FROM core_share_lot WHERE customer_id = :c",
            c=CUSTOMER,
        )
    }
    assert remain["LOT-A1"] == Decimal("0.00")  # 归零保留行（D9/P2）
    assert remain["LOT-A2"] == Decimal("30.00")
    assert remain["LOT-IN-1"] == _dec(req.in_qty)
    new_lot = _rows(
        sqlite_engine,
        "SELECT * FROM core_share_lot WHERE lot_id = 'LOT-IN-1'",
    )[0]
    assert new_lot["product_id"] == PROD_IN
    assert _dec(new_lot["nav"]) == _dec(IN_NAV)
    assert str(new_lot["confirmed_at"])[:10] == "2026-09-05"  # T+1（D14）
    assert new_lot["source_trade_id"] == req.in_trade_id

    # ③ 明细：每批一行，含 nav / nav_date（D6 补偿源）
    details = _rows(
        sqlite_engine,
        "SELECT * FROM core_convert_lot_detail WHERE convert_group_id = :g ORDER BY lot_id",
        g="CNV-T6-1",
    )
    assert [d["lot_id"] for d in details] == ["LOT-A1", "LOT-A2"]
    assert [d["hold_days"] for d in details] == [34, 3]
    assert [_dec(d["fee_rate"], "0.0001") for d in details] == [
        Decimal("0.0050"), Decimal("0.0150")
    ]

    # ④ 两端持仓：转出等比例结转成本、转入按净值建仓
    out_h = _holding(sqlite_engine, PROD_OUT)
    assert _dec(out_h["qty"]) == Decimal("30.00")  # 150 − 120
    assert _dec(out_h["cost_amount"]) == Decimal("30.00")  # 150 × (1 − 120/150)
    assert _dec(out_h["market_value"]) == Decimal("30.90")  # 154.50 × 0.2
    assert _dec(out_h["pnl_pct"], "0.0001") == Decimal("0.0300")
    assert str(out_h["as_of"])[:10] == "2026-09-04"

    in_h = _holding(sqlite_engine, PROD_IN)
    assert _dec(in_h["qty"]) == _dec(req.in_qty)
    assert _dec(in_h["cost_amount"]) == _dec(req.in_amount)


# ── 2. 冲突路径（并发哨兵）───────────────────────────────────────────
def test_apply_convert_raises_lot_conflict_and_leaves_nothing(sqlite_engine):
    _seed_base(sqlite_engine)
    _seed_lot(sqlite_engine, "LOT-A1", "100", "1.0300", datetime(2026, 8, 1, 10, 0, 0))
    _seed_holding(sqlite_engine, PROD_OUT, "100", "100.00", "103.00")

    # 申请 100 拿到正常入参，再把扣减量改大到超过剩余 → 条件 UPDATE rowcount=0
    req = _build_input(sqlite_engine, group_id="CNV-T6-C", requested="100")
    bigger = tuple(
        LotCharge(
            lot_id=c.lot_id, qty=Decimal("120"), hold_days=c.hold_days,
            amount=c.amount, fee_rate=c.fee_rate, fee_amount=c.fee_amount,
            nav=c.nav, nav_date=c.nav_date,
        )
        for c in req.charges
    )
    req = ConvertApplyInput(**{**req.__dict__, "charges": bigger})

    with pytest.raises(LotConflict) as exc:
        ConvertCoreRepository(engine=sqlite_engine).apply_convert(req)
    assert exc.value.status_code == 409

    # 冲突即回滚：流水、明细、资金流都是 0 行
    assert _rows(sqlite_engine, "SELECT * FROM core_trade") == []
    assert _rows(sqlite_engine, "SELECT * FROM core_convert_lot_detail") == []
    assert _rows(sqlite_engine, "SELECT * FROM core_cash_flow") == []


# ── 3. 回滚路径（同事务的硬证据）──────────────────────────────────────
def test_apply_convert_rolls_back_everything_when_detail_insert_fails(
    sqlite_engine, monkeypatch,
):
    _seed_base(sqlite_engine)
    _seed_lot(sqlite_engine, "LOT-A1", "100", "1.0300", datetime(2026, 8, 1, 10, 0, 0))
    _seed_lot(sqlite_engine, "LOT-A2", "50", "1.0000", datetime(2026, 9, 1, 10, 0, 0))
    _seed_holding(sqlite_engine, PROD_OUT, "150", "150.00", "154.50")

    def _boom(self, conn, req):  # noqa: ANN001
        raise RuntimeError("注入故障：明细写入失败")

    monkeypatch.setattr(ConvertCoreRepository, "_insert_lot_details", _boom)
    req = _build_input(sqlite_engine, group_id="CNV-T6-R", requested="120")

    with pytest.raises(RuntimeError):
        ConvertCoreRepository(engine=sqlite_engine).apply_convert(req)

    # 全部无残留 = 阶段一确实在同一个事务里（架构 §5.1）
    assert _rows(sqlite_engine, "SELECT * FROM core_trade") == []
    assert _rows(sqlite_engine, "SELECT * FROM core_convert_lot_detail") == []
    # T-16（R-11）：cash_flow 与流水同事务 → 注入故障点之后的回滚同样零残留
    assert _rows(sqlite_engine, "SELECT * FROM core_cash_flow") == []
    assert _dec(
        _rows(
            sqlite_engine,
            "SELECT remain_qty FROM core_share_lot WHERE lot_id = 'LOT-A1'",
        )[0]["remain_qty"]
    ) == Decimal("100.00")
    assert _dec(_holding(sqlite_engine, PROD_OUT)["qty"]) == Decimal("150.00")
    assert _rows(
        sqlite_engine,
        "SELECT * FROM core_share_lot WHERE lot_id = 'LOT-IN-1'",
    ) == []
    assert _holding(sqlite_engine, PROD_IN) is None


# ── 4. 转出归零保留行 / 转入端首次 INSERT 与再次 UPDATE ────────────────
def test_second_convert_zeroes_out_holding_and_accumulates_in_holding(sqlite_engine):
    _seed_base(sqlite_engine)
    _seed_lot(sqlite_engine, "LOT-A1", "100", "1.0300", datetime(2026, 8, 1, 10, 0, 0))
    _seed_lot(sqlite_engine, "LOT-A2", "50", "1.0000", datetime(2026, 9, 1, 10, 0, 0))
    _seed_holding(sqlite_engine, PROD_OUT, "150", "150.00", "154.50")
    repo = ConvertCoreRepository(engine=sqlite_engine)

    first = _build_input(sqlite_engine, group_id="CNV-T6-F", requested="120")
    repo.apply_convert(first)
    assert _dec(_holding(sqlite_engine, PROD_OUT)["qty"]) == Decimal("30.00")

    # 第二次：转出剩余 30（全转）→ 归零保留行；转入端已存在 → 累加而非覆盖
    second = _build_input(
        sqlite_engine, group_id="CNV-T6-S", requested="30", in_lot_id="LOT-IN-2",
    )
    repo.apply_convert(second)

    out_h = _holding(sqlite_engine, PROD_OUT)
    assert out_h is not None  # 归零**保留行**，不删（D9/P2，`xh_core_rw` 也无 DELETE）
    assert _dec(out_h["qty"]) == Decimal("0.00")

    in_h = _holding(sqlite_engine, PROD_IN)
    assert _dec(in_h["qty"]) == _dec(first.in_qty + second.in_qty)
    assert _dec(in_h["cost_amount"]) == _dec(first.in_amount + second.in_amount)
    # 持仓行数仍为 1（UNIQUE(customer_id, product_id)）
    assert len(
        _rows(
            sqlite_engine,
            "SELECT * FROM core_holding WHERE customer_id = :c",
            c=CUSTOMER,
        )
    ) == 2  # 转出 + 转入各一行


# ── 5. R-a 并发首次建行（IntegrityError → 回退增量 UPDATE）─────────────
def test_concurrent_first_insert_falls_back_to_incremental_update(
    sqlite_engine, monkeypatch,
):
    """两笔并发首次转入同一 (客户, 产品)：只有一行、金额为两笔之和、两笔都成功。

    模拟方式：注入点选在 `_insert_holding_row`（**UPDATE 之后、INSERT 之前**）——
    本笔的 UPDATE 已命中 0 行，此刻另一笔并发转换提交了同一 (客户, 产品) 的行，
    于是本笔 INSERT 撞 `UNIQUE(customer_id, product_id)` → 回退增量 UPDATE。
    回退若写成"绝对值 UPDATE"就会覆盖另一笔，故本用例是 R-a 的硬证据。
    """
    _seed_base(sqlite_engine)
    _seed_lot(sqlite_engine, "LOT-A1", "100", "1.0300", datetime(2026, 8, 1, 10, 0, 0))
    _seed_holding(sqlite_engine, PROD_OUT, "100", "100.00", "103.00")

    original = ConvertCoreRepository._insert_holding_row
    concurrent_qty = Decimal("50")
    concurrent_cost = Decimal("50")
    hit = {"fallback": False}  # 记录回退分支是否真的被走到（防止用例假绿）

    def _with_concurrent_row(self, conn, insert_sql, update_sql, params):  # noqa: ANN001
        if params["pid"] == PROD_IN:
            # 模拟另一笔并发转换已提交：本事务内可见，INSERT 必然撞 UNIQUE
            conn.execute(
                text(
                    "INSERT INTO core_holding (customer_id, product_id, qty, "
                    "cost_amount, market_value, pnl_pct, as_of) "
                    "VALUES (:c, :p, :q, :cost, :mv, 0, :d)"
                ),
                {
                    "c": params["cid"], "p": PROD_IN,
                    "q": str(concurrent_qty), "cost": str(concurrent_cost),
                    "mv": str(concurrent_cost), "d": params["as_of"],
                },
            )
            hit["fallback"] = True
        original(self, conn, insert_sql, update_sql, params)

    monkeypatch.setattr(
        ConvertCoreRepository, "_insert_holding_row", _with_concurrent_row
    )
    req = _build_input(sqlite_engine, group_id="CNV-T6-X", requested="100")
    ConvertCoreRepository(engine=sqlite_engine).apply_convert(req)  # 不得抛 5xx

    assert hit["fallback"] is True  # 注入点确实被走到，否则本用例无意义
    rows = _rows(
        sqlite_engine,
        "SELECT * FROM core_holding WHERE customer_id = :c AND product_id = :p",
        c=CUSTOMER, p=PROD_IN,
    )
    assert len(rows) == 1  # 只有一行
    assert _dec(rows[0]["qty"]) == _dec(concurrent_qty + req.in_qty)
    assert _dec(rows[0]["cost_amount"]) == _dec(concurrent_cost + req.in_amount)


# ── 6. 死锁 1213 自动重试（2026-09-10 用户拍板 · T-13 后补）────────────
def _mysql_error(code: int, msg: str) -> OperationalError:
    """构造带 MySQL 错误码的 `OperationalError`（orig.args[0] = code，仿 pymysql）。"""
    return OperationalError("stmt", {}, Exception(code, msg))


def test_apply_convert_retries_on_deadlock_and_commits(sqlite_engine, monkeypatch):
    """前 2 次注入 1213 死锁 → 第 3 次成功：**整事务重试**且最终落库。

    断言四层：① 成功不抛；② 事务体被调 3 次（1 + `convert_deadlock_retry_max`
    默认 2）；③ 退避两次且指数递增（base·2^(n-1) + [0, base) 抖动，第 2 次恒大于
    第 1 次）；④ 重试走完后数据真正落库 —— 证明重试的是「整个事务」而非空转。
    """
    _seed_base(sqlite_engine)
    _seed_lot(sqlite_engine, "LOT-A1", "100", "1.0300", datetime(2026, 8, 1, 10, 0, 0))
    _seed_lot(sqlite_engine, "LOT-A2", "50", "1.0000", datetime(2026, 9, 1, 10, 0, 0))
    _seed_holding(sqlite_engine, PROD_OUT, "150", "150.00", "154.50")

    original = ConvertCoreRepository._apply_convert_once
    calls = {"n": 0}

    def _flaky_once(self, req):  # noqa: ANN001
        calls["n"] += 1
        if calls["n"] <= 2:
            raise _mysql_error(1213, "Deadlock found when trying to get lock")
        return original(self, req)

    monkeypatch.setattr(ConvertCoreRepository, "_apply_convert_once", _flaky_once)

    delays: list[float] = []
    monkeypatch.setattr(
        "app.gateway.convert_core_repository.time.sleep", lambda s: delays.append(s)
    )

    req = _build_input(sqlite_engine, group_id="CNV-T6-DL", requested="120")
    ConvertCoreRepository(engine=sqlite_engine).apply_convert(req)  # 不得抛

    assert calls["n"] == 3  # 首跑 1 次 + 重试 2 次（默认预算）
    assert len(delays) == 2
    assert 0 < delays[0] < delays[1]  # 指数退避 + 抖动
    # 重试后真正落库（与 happy path 同口径的关键结果）
    assert _dec(_holding(sqlite_engine, PROD_OUT)["qty"]) == Decimal("30.00")
    assert len(
        _rows(
            sqlite_engine,
            "SELECT * FROM core_trade WHERE convert_group_id = 'CNV-T6-DL'",
        )
    ) == 2


def test_apply_convert_gives_up_after_retry_budget(sqlite_engine, monkeypatch):
    """一直 1213 → 按 1+max_retries 次耗尽预算后**原样上抛**，且不留任何残留。"""
    _seed_base(sqlite_engine)
    _seed_lot(sqlite_engine, "LOT-A1", "100", "1.0300", datetime(2026, 8, 1, 10, 0, 0))

    calls = {"n": 0}

    def _always_deadlock(self, req):  # noqa: ANN001
        calls["n"] += 1
        raise _mysql_error(1213, "Deadlock found when trying to get lock")

    monkeypatch.setattr(ConvertCoreRepository, "_apply_convert_once", _always_deadlock)
    monkeypatch.setattr(
        "app.gateway.convert_core_repository.time.sleep", lambda s: None
    )

    req = _build_input(sqlite_engine, group_id="CNV-T6-DX", requested="100")
    with pytest.raises(OperationalError):
        ConvertCoreRepository(engine=sqlite_engine).apply_convert(req)
    assert calls["n"] == 3  # 1 + convert_deadlock_retry_max(默认 2)
    assert _rows(sqlite_engine, "SELECT * FROM core_trade") == []  # 零残留


def test_apply_convert_does_not_retry_non_deadlock_errors(sqlite_engine, monkeypatch):
    """1205（锁等待超时）不属于死锁 → **不重试**，第 1 次即上抛。

    理由：1205 的重试语义受 `innodb_rollback_on_timeout` 影响（官方口径为
    「默认重试语句」，回滚粒度依配置而定），与 1213 的「重试整个事务」不通用。
    """
    _seed_base(sqlite_engine)
    _seed_lot(sqlite_engine, "LOT-A1", "100", "1.0300", datetime(2026, 8, 1, 10, 0, 0))

    calls = {"n": 0}

    def _lock_timeout(self, req):  # noqa: ANN001
        calls["n"] += 1
        raise _mysql_error(1205, "Lock wait timeout exceeded")

    monkeypatch.setattr(ConvertCoreRepository, "_apply_convert_once", _lock_timeout)
    monkeypatch.setattr(
        "app.gateway.convert_core_repository.time.sleep", lambda s: None
    )

    req = _build_input(sqlite_engine, group_id="CNV-T6-T5", requested="50")
    with pytest.raises(OperationalError):
        ConvertCoreRepository(engine=sqlite_engine).apply_convert(req)
    assert calls["n"] == 1  # 未触发任何重试