T-15 D28 部分成交专项:
- sqlite +1 终态单幂等 skipped 口径裁定(ConfirmConflict 仅限读后状态才变窗口)
- 真库 confirm +3 组 19 项(E2 占用释放双向 / G2 actual_qty=0 → rejected / H2 撤单后确认)→ 102/102
- CONVERT_STRESS Barrier(2) 窄窗争批:恰 1 confirmed + 1 rejected
- 突变实证防御纵深:④可用量复核 / ⑤plan_lots / 事务内哨兵三道防线拆任一道仍收敛 rejected
T-16 D29 资金流中转:
- _insert_cash_flows 全仓唯一写入点(out/redeem 费前 + in/subscribe 净转入,remark=convert:{gid} 同事务)
- test_confirm_writes_no_cash_flow 反转为恰 2 条 + 金额对账(T-7 登记闭环)
- 真库 apply A2 组 + B 组 0 残留 → 30/30;confirm 反转断言 → 103/103
- FK 坑(core_cash_flow.fk_cf_customer 挡客户删除):5 处清理段补删
T-17 D30 share_class + A/C 互转:
- _validate_products 纯新增拦截分支:两端 share_class 非空且不同才查 allow_ac_convert(任一端=1 两向放行)
- 口径裁定 2 条(R-12 字面偏差,待用户追认):NULL 不参与 A/C 判定 / 同类走一般转换规则
- 联网核实(用户铁律):A/C 互转需管理人开通(摩根 2025-11 公告)= 开关真实载体;持有期重新起算与本仓一致
- 种子 ⑥ 段(双开对/双关对照/同类对照);sqlite +4 → 842 passed / 8 skipped;真库 accept H 组 → 43/43
- 突变:分支失效 → closed 红,还原绿
文档:开发计划 T-15~T-17 DoD 全勾 + 执行记录 · 基线 842 · TODO/MEMORY/AGENTS/交接文档 v4.5 终态(第 5 步收官 → 下一步第 6 步集成测试)
557 lines
23 KiB
Python
557 lines
23 KiB
Python
"""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 # 未触发任何重试
|