"""T-2 基金转换纯函数单测（开发计划 §4.2 DoD）。 覆盖 12 类： 1. **精度**：全部量化点必须 `ROUND_HALF_UP`（`Decimal` 默认是 `ROUND_HALF_EVEN`， 架构风险 #3 —— `.5` 边界处两种舍入结果不同，必须命中 HALF_UP 那一侧）； 2. **分档边界**：`6/7/29/30/179/180/364/365`（左闭右开，满 7 日归 7–30 档）； 3. **FIFO**：主序 `confirmed_at`、同 `confirmed_at` 以 `lot_id` 兜底（评审 S1）； 4. **跨批次计费**：逐批先舍入后求和（PRD §5.3 主示例 154.50 + 309.00）； 5. **补差费双口径**：B `252.40` / A `253.91`（同输入，差约 0.6%）；同费率归零； 6. **最低持有处置**：`force_transfer` / `force_redeem` 双动作、 **恰好等于阈值不触发**、**零剩余不触发**（实现级裁定）； 7. **全链自证**：PRD §5.3 主示例每个数字逐项复算（含 `rounding_diff = -0.0026`）； 8. **持有期**：不含申请日、T+1 起算少 1 天； 9. **净值**：无净值 → 503；过期 → 只标 `nav_stale` 不阻断； 10. **批次补建**：D18 确定性（`crc32` 而非内置 `hash()`）； 11. **错误码映射**：§8.3 全表； 12. **纯函数约束**：convert 包源码零仓储/引擎/配置依赖（DoD 第 3 条， 从"人工 grep"升级为断言）。 本文件**不依赖 DB、不依赖 conftest 夹具**——纯函数包的全部输入由测试自建。 """ from __future__ import annotations import re import zlib from datetime import date, datetime, time, timedelta from decimal import Decimal from pathlib import Path import pytest from app.service.convert import calc, fee, lot_bootstrap, nav from app.service.convert.errors import ( BelowMinQty, ConvertError, CrossEntityNotSupported, FeeRuleMissing, IdempotencyUnavailable, InsufficientShares, LotConflict, NavNotReady, ProductNotRedeemable, ProductNotSubscribable, SameProduct, TooManyLots, ) from app.service.convert.types import FeeRule, Lot # ── 公共夹具数据 ───────────────────────────────────────────────────── #: 与 PRD §5.3 示例同一交易日（也是 `06-seed-nav.sql` 的净值日期）。 TRADE_DATE = date(2026, 9, 4) #: PRD §4.3 / 07-seed-fee-rule.sql 的五档赎回费（22 号文 §10 下限）。 RULES = [ FeeRule(min_hold_days=0, max_hold_days=7, rate=Decimal("0.0150")), FeeRule(min_hold_days=7, max_hold_days=30, rate=Decimal("0.0100")), FeeRule(min_hold_days=30, max_hold_days=180, rate=Decimal("0.0050")), FeeRule(min_hold_days=180, max_hold_days=365, rate=Decimal("0.0025")), FeeRule(min_hold_days=365, max_hold_days=None, rate=Decimal("0.0000")), ] #: 主示例两端申购费率（09-seed-org.sql：110022 债基 0.0030 → 003095 主动偏股 0.0080）。 OUT_SUB_RATE = Decimal("0.0030") IN_SUB_RATE = Decimal("0.0080") def _lot(lot_id: str, hold_days: int, qty: str, nav: str = "1.0300") -> Lot: """构造 `hold_days` 天前确认的批次（交易日固定为 TRADE_DATE）。""" confirmed = datetime.combine(TRADE_DATE - timedelta(days=hold_days), time(10, 0, 0)) return Lot( lot_id=lot_id, confirmed_at=confirmed, remain_qty=Decimal(qty), nav=Decimal(nav), qty=Decimal(qty), ) def _fee_of(lot_allocation, hold_days: int) -> Decimal: """按 `plan_lots` 的分配结果算单批金额与赎回费。""" amount = calc.lot_amount(lot_allocation.qty, lot_allocation.nav) return calc.lot_fee(amount, fee.pick_fee_rate(RULES, hold_days)) # ── 1. 精度：HALF_UP 而非 HALF_EVEN ───────────────────────────────── class TestRounding: """所有量化点显式 `ROUND_HALF_UP`（架构风险 #3）。""" def test_round2_half_up_away_from_zero(self): # Decimal 默认 ROUND_HALF_EVEN 会得到 0.00（偶数侧）；HALF_UP 必须是 0.01 assert calc.round2(Decimal("0.005")) == Decimal("0.01") assert calc.round2(Decimal("0.015")) == Decimal("0.02") def test_default_context_differs_from_half_up(self): """反向自证：若实现漏传 rounding，本用例会红。""" from decimal import ROUND_HALF_EVEN banker = Decimal("0.005").quantize(Decimal("0.01"), rounding=ROUND_HALF_EVEN) assert banker == Decimal("0.00") assert calc.round2(Decimal("0.005")) != banker def test_lot_amount_2_places_half_up(self): # 0.5 份 × 1.0001 = 0.50005 → 0.50；1.5 份 × 0.0001 之类极小值也要能舍 assert calc.lot_amount(Decimal("1.5"), Decimal("0.0034")) == Decimal("0.01") assert calc.lot_amount(Decimal("30000.0000"), Decimal("1.0300")) == Decimal("30900.00") def test_lot_fee_2_places_half_up(self): assert calc.lot_fee(Decimal("30900.00"), Decimal("0.0050")) == Decimal("154.50") assert calc.lot_fee(Decimal("20600.00"), Decimal("0.0150")) == Decimal("309.00") def test_in_qty_2_places_half_up_not_floor(self): # v1.0 勘误：份额是 2 位四舍五入，不是 4 位向下取整 # 50784.10 / 0.95 = 53456.9473... → 53456.95（向下取整会得 53456.94） assert calc.in_qty(Decimal("50784.10"), Decimal("0.9500")) == Decimal("53456.95") def test_in_qty_rejects_non_positive_nav(self): with pytest.raises(ValueError): calc.in_qty(Decimal("100"), Decimal("0")) def test_rounding_diff_negative_means_client_gains(self): # 理论 53456.9474 − 实得 53456.95 = -0.0026（负 = 客户多得） diff = calc.rounding_diff(Decimal("50784.10"), Decimal("0.9500"), Decimal("53456.95")) assert diff == Decimal("-0.0026") # ── 2. 费率分档边界 ───────────────────────────────────────────────── class TestFeeBands: """左闭右开 `[min, max)`；满 7 日归 7–30 档（PRD §12 I-2）。""" @pytest.mark.parametrize( ("hold_days", "expected"), [ (0, "0.0150"), (6, "0.0150"), # < 7 日 (7, "0.0100"), # 满 7 日 → 进 7–30 档 (29, "0.0100"), (30, "0.0050"), # 满 30 日 → 进 30–180 档 (179, "0.0050"), (180, "0.0025"), (364, "0.0025"), (365, "0.0000"), (1000, "0.0000"), # max_hold_days = NULL 表示无上限 ], ) def test_band_boundaries(self, hold_days, expected): assert fee.pick_fee_rate(RULES, hold_days) == Decimal(expected) def test_missing_rule_raises_instead_of_zero(self): # 无命中绝不能静默按 0 收费（少收赎回费且不留痕） with pytest.raises(FeeRuleMissing) as exc: fee.pick_fee_rate([FeeRule(365, None, Decimal("0.0000"))], 3) assert exc.value.error_code == "FEE_RULE_MISSING" assert exc.value.status_code == 500 def test_subscribe_rules_are_ignored(self): """`fee_type='subscribe'` 本期不启用，混入规则表也不得被选中。""" rules = [ FeeRule(0, 7, Decimal("0.8000"), fee_type="subscribe"), FeeRule(0, 7, Decimal("0.0150"), fee_type="redeem"), ] assert fee.pick_fee_rate(rules, 3) == Decimal("0.0150") def test_overlapping_bands_take_most_specific(self): """区间误配重叠时取 `min_hold_days` 最大者，避免"取到哪档看运气"。""" rules = [ FeeRule(0, 365, Decimal("0.0050")), FeeRule(30, 180, Decimal("0.0025")), ] # 排序后被 max(min_hold_days) 选中，与传入顺序无关 assert fee.pick_fee_rate(rules, 100) == Decimal("0.0025") assert fee.pick_fee_rate(list(reversed(rules)), 100) == Decimal("0.0025") # ── 3. FIFO 排序与分配 ─────────────────────────────────────────────── class TestFifoPlan: def test_allocates_oldest_lot_first(self): lots = [_lot("LOT-B", 3, "20000.0000"), _lot("LOT-A", 100, "30000.0000")] plan = calc.plan_lots(lots, Decimal("40000")) assert [a.lot_id for a in plan.allocations] == ["LOT-A", "LOT-B"] assert plan.allocations[0].qty == Decimal("30000.0000") assert plan.allocations[1].qty == Decimal("10000.0000") def test_same_confirmed_at_tiebreak_by_lot_id(self): """同一注册日多批次顺序不可由存储引擎决定（评审 S1）。""" same = datetime.combine(TRADE_DATE - timedelta(days=10), time(10, 0, 0)) lots = [ Lot("LOT-Z", same, Decimal("100.0000"), Decimal("1.0000")), Lot("LOT-A", same, Decimal("100.0000"), Decimal("1.0000")), ] plan = calc.plan_lots(lots, Decimal("150")) assert [a.lot_id for a in plan.allocations] == ["LOT-A", "LOT-Z"] def test_ignores_zero_remain_qty_lots(self): lots = [ _lot("LOT-USED", 100, "1000.0000"), _lot("LOT-FRESH", 3, "500.0000"), ] lots[0] = Lot( lots[0].lot_id, lots[0].confirmed_at, Decimal("0.0000"), lots[0].nav ) plan = calc.plan_lots(lots, Decimal("500")) assert [a.lot_id for a in plan.allocations] == ["LOT-FRESH"] def test_insufficient_shares(self): lots = [_lot("LOT-A", 100, "1000.0000")] with pytest.raises(InsufficientShares) as exc: calc.plan_lots(lots, Decimal("1000.0001")) assert exc.value.error_code == "INSUFFICIENT_SHARES" def test_requested_must_be_positive(self): with pytest.raises(ValueError): calc.plan_lots([_lot("LOT-A", 100, "1000")], Decimal("0")) def test_batch_count_equals_allocations_len(self): lots = [_lot(f"LOT-{i}", 100 - i, "1000.0000") for i in range(3)] plan = calc.plan_lots(lots, Decimal("2500")) assert plan.batch_count == len(plan.allocations) == 3 def test_available_qty_uses_remain_qty(self): """份额足够判据以 `Σ remain_qty` 为准，不用 `core_holding.qty`（PRD §12）。""" lots = [_lot("LOT-A", 100, "1000.0000")] lots[0] = Lot( lots[0].lot_id, lots[0].confirmed_at, Decimal("300.0000"), lots[0].nav, qty=Decimal("1000.0000"), # 原始份额更大，但已扣减 ) plan = calc.plan_lots(lots, Decimal("300")) assert plan.available_qty == Decimal("300.0000") # ── 4. 最低持有余额处置 ────────────────────────────────────────────── class TestMinHoldAction: """触发条件严格为「余额 < 阈值」（非 ≤）；PRD §12 I-4。""" def test_below_threshold_forces_full_transfer(self): lots = [_lot("LOT-A", 100, "6000.0000")] plan = calc.plan_lots( lots, Decimal("5500"), Decimal("1000"), "force_transfer" ) assert plan.forced_full_transfer is True assert plan.action == "force_transfer" assert plan.actual_qty == Decimal("6000.0000") # 全转，客户指令被改变 assert plan.requested_qty == Decimal("5500") def test_below_threshold_force_redeem_action(self): lots = [_lot("LOT-A", 100, "6000.0000")] plan = calc.plan_lots(lots, Decimal("5500"), Decimal("1000"), "force_redeem") assert plan.action == "force_redeem" assert plan.actual_qty == Decimal("6000.0000") def test_leftover_equal_threshold_not_triggered(self): """PRD 审查例：持 6000、申请 5000、阈值 1000 → 余额恰好 1000 → 不触发。""" lots = [_lot("LOT-A", 100, "6000.0000")] plan = calc.plan_lots(lots, Decimal("5000"), Decimal("1000"), "force_transfer") assert plan.forced_full_transfer is False assert plan.action == "transfer" assert plan.actual_qty == Decimal("5000") def test_zero_leftover_not_forced(self): """实现级裁定：客户本就是清仓（余额 = 0）不得被标记为"强制"。 否则响应里的 `forced_full_transfer=True` 会被前端当成 "你的指令被系统改了"，而实际转出份额与申请完全一致。 """ lots = [_lot("LOT-A", 100, "6000.0000")] plan = calc.plan_lots(lots, Decimal("6000"), Decimal("1000"), "force_transfer") assert plan.forced_full_transfer is False assert plan.actual_qty == plan.requested_qty == Decimal("6000.0000") def test_zero_threshold_disables_check(self): lots = [_lot("LOT-A", 100, "6000.0000")] plan = calc.plan_lots(lots, Decimal("5000"), Decimal("0"), "force_transfer") assert plan.forced_full_transfer is False def test_unknown_action_rejected(self): lots = [_lot("LOT-A", 100, "6000.0000")] with pytest.raises(ValueError): calc.plan_lots(lots, Decimal("5500"), Decimal("1000"), "force_dance") # ── 5. 补差费双口径 ───────────────────────────────────────────────── class TestDiffFee: CONV = Decimal("51036.50") def test_amount_diff_is_default_and_matches_prd(self): # 405.05（转入端 0.8% 价外费）− 152.65（转出端 0.3% 价外费） assert calc.diff_fee(self.CONV, OUT_SUB_RATE, IN_SUB_RATE) == Decimal("252.40") assert calc.diff_fee( self.CONV, OUT_SUB_RATE, IN_SUB_RATE, "amount_diff" ) == Decimal("252.40") def test_rate_diff_matches_prd(self): assert calc.diff_fee( self.CONV, OUT_SUB_RATE, IN_SUB_RATE, "rate_diff" ) == Decimal("253.91") def test_two_modes_actually_differ(self): """两口径差约 0.6% —— 必须显式选口径，不可混用（PRD §2.1.1 / Q9）。""" b = calc.diff_fee(self.CONV, OUT_SUB_RATE, IN_SUB_RATE, "amount_diff") a = calc.diff_fee(self.CONV, OUT_SUB_RATE, IN_SUB_RATE, "rate_diff") assert a != b @pytest.mark.parametrize("mode", ["amount_diff", "rate_diff"]) def test_zero_when_in_rate_not_higher(self, mode): assert calc.diff_fee(self.CONV, IN_SUB_RATE, OUT_SUB_RATE, mode) == Decimal("0.00") assert calc.diff_fee(self.CONV, OUT_SUB_RATE, OUT_SUB_RATE, mode) == Decimal("0.00") def test_zero_rate_inbound_gives_zero(self): """转入端为货基（申购费 0）→ 不补差。""" assert calc.diff_fee(self.CONV, OUT_SUB_RATE, Decimal("0"), "amount_diff") == Decimal("0.00") def test_invalid_mode_rejected(self): with pytest.raises(ValueError): calc.diff_fee(self.CONV, OUT_SUB_RATE, IN_SUB_RATE, "whatever") # ── 6. 全链自证（PRD §5.3 逐行） ───────────────────────────────────── class TestPrdMainExample: """主示例每个派生值都由本块数据算出（v0.9.1 · 脚本实算回填）。""" @pytest.fixture def chain(self): lots = [_lot("LOT-A", 100, "30000.0000"), _lot("LOT-B", 3, "20000.0000")] plan = calc.plan_lots(lots, Decimal("50000")) out_amount = sum( (calc.lot_amount(a.qty, a.nav) for a in plan.allocations), Decimal("0") ) redeem_fee = sum( ( _fee_of(a, calc.hold_days(TRADE_DATE, a.confirmed_at)) for a in plan.allocations ), Decimal("0"), ) return plan, out_amount, redeem_fee def test_per_lot_breakdown(self, chain): plan, _, _ = chain assert [a.qty for a in plan.allocations] == [ Decimal("30000.0000"), Decimal("20000.0000"), ] assert calc.hold_days(TRADE_DATE, plan.allocations[0].confirmed_at) == 100 assert calc.hold_days(TRADE_DATE, plan.allocations[1].confirmed_at) == 3 assert _fee_of(plan.allocations[0], 100) == Decimal("154.50") assert _fee_of(plan.allocations[1], 3) == Decimal("309.00") def test_amounts_chain(self, chain): plan, out_amount, redeem_fee = chain assert out_amount == Decimal("51500.00") assert redeem_fee == Decimal("463.50") # 逐批舍入后求和 conv = calc.convert_amount(out_amount, redeem_fee) assert conv == Decimal("51036.50") diff_b = calc.diff_fee(conv, OUT_SUB_RATE, IN_SUB_RATE, "amount_diff") assert diff_b == Decimal("252.40") in_amount = calc.convert_amount(conv, diff_b) assert in_amount == Decimal("50784.10") assert calc.in_qty(in_amount, Decimal("0.9500")) == Decimal("53456.95") assert ( calc.rounding_diff(in_amount, Decimal("0.9500"), Decimal("53456.95")) == Decimal("-0.0026") ) def test_three_amount_fields_are_distinct(self, chain): """`out_amount` / `convert_amount` / `in_amount` 三者不可混用（PRD M-4）。""" _, out_amount, redeem_fee = chain conv = calc.convert_amount(out_amount, redeem_fee) in_amount = calc.convert_amount(conv, Decimal("252.40")) assert out_amount != conv != in_amount assert out_amount == Decimal("51500.00") assert conv == Decimal("51036.50") assert in_amount == Decimal("50784.10") def test_same_rate_control_group(self, chain): """同费率对照（110022 0.0030 → 510300 0.0030）：补差 0，份额 53722.63。""" _, out_amount, redeem_fee = chain conv = calc.convert_amount(out_amount, redeem_fee) diff = calc.diff_fee(conv, OUT_SUB_RATE, OUT_SUB_RATE, "amount_diff") assert diff == Decimal("0.00") in_amount = calc.convert_amount(conv, diff) assert in_amount == conv == Decimal("51036.50") assert calc.in_qty(in_amount, Decimal("0.9500")) == Decimal("53722.63") # ── 7. 持有期 ─────────────────────────────────────────────────────── class TestHoldDays: def test_excludes_application_day(self): """(交易日 − 确认日).days，不含申请日：昨日确认今天申请 = 1 天。""" assert calc.hold_days(date(2026, 9, 4), datetime(2026, 9, 3, 10)) == 1 def test_t_plus_one_confirmation_is_one_day_less(self): """转换转入批次以 T+1 为确认日 → 同一交易日下持有期比"自 T 起算"少 1 天。 这是更贴近真实的做法（真实持有期自确认日起算），费率档因此更严（PRD B-5）。 """ trade = date(2026, 9, 4) from_t = calc.hold_days(trade, datetime(2026, 8, 25, 10)) from_t1 = calc.hold_days(trade, datetime(2026, 8, 26, 10)) assert from_t == 10 assert from_t1 == 9 def test_full_seven_days_lands_in_second_band(self): assert calc.hold_days(date(2026, 9, 10), datetime(2026, 9, 3, 10)) == 7 assert fee.pick_fee_rate(RULES, 7) == Decimal("0.0100") def test_rejects_none(self): with pytest.raises(ValueError): calc.hold_days(None, datetime(2026, 9, 3, 10)) with pytest.raises(ValueError): calc.hold_days(date(2026, 9, 4), None) # ── 8. 净值口径 ───────────────────────────────────────────────────── class TestNav: def test_missing_nav_raises_503(self): with pytest.raises(NavNotReady) as exc: nav.ensure_nav_ready(None, product_id="PROD-X") assert exc.value.status_code == 503 assert exc.value.error_code == "NAV_NOT_READY" def test_present_nav_passes(self): nav.ensure_nav_ready(date(2026, 9, 4), product_id="PROD-110022") @pytest.mark.parametrize( ("nav_date", "expected"), [("2026-09-04", False), ("2026-09-01", False), ("2026-08-31", True)], ) def test_stale_boundary_is_strictly_greater(self, nav_date, expected): # 阈值 3 天：距交易日恰好 3 天不算过期（PRD §2.2「超过此值」） assert nav.is_stale(date.fromisoformat(nav_date), TRADE_DATE, 3) is expected def test_is_stale_rejects_missing_nav(self): """"无净值"必须走 503，不得混进 stale 分支（两者处置完全不同）。""" with pytest.raises(ValueError): nav.is_stale(None, TRADE_DATE, 3) def test_evaluate_nav_returns_stale_flag(self): assert nav.evaluate_nav(date(2026, 9, 4), TRADE_DATE, 3) is False assert nav.evaluate_nav(date(2026, 8, 1), TRADE_DATE, 3) is True def test_evaluate_nav_raises_when_missing(self): with pytest.raises(NavNotReady): nav.evaluate_nav(None, TRADE_DATE, 3, product_id="PROD-X") # ── 9. 批次兜底补建（D8 / D18） ────────────────────────────────────── class TestBootstrapLots: HOLDING = { "customer_id": "CUST-9527", "product_id": "PROD-110022", "qty": Decimal("80000.0000"), "cost_amount": Decimal("82400.00"), "market_value": Decimal("82400.00"), "pnl_pct": Decimal("0.0000"), "as_of": date(2026, 9, 4), } def test_single_lot_equals_holding_qty(self): lots = lot_bootstrap.bootstrap_lots(self.HOLDING) assert len(lots) == 1 assert lots[0].remain_qty == lots[0].qty == Decimal("80000.0000") # Σ remain_qty == core_holding.qty（08 种子的同一不变量） assert sum(lot.remain_qty for lot in lots) == self.HOLDING["qty"] def test_deterministic_across_calls(self): """同一持仓重复补建结果完全一致 —— D18 同源断言的立足点。""" first = lot_bootstrap.bootstrap_lots(self.HOLDING) for _ in range(50): assert lot_bootstrap.bootstrap_lots(self.HOLDING) == first def test_offset_uses_crc32_not_builtin_hash(self): """必须用 `zlib.crc32`：内置 `hash()` 受 PYTHONHASHSEED 随机化， 会让 gateway 与 rebuild_lots.py（两个进程）算出不同 confirmed_at。""" key = "CUST-9527|PROD-110022" expected = lot_bootstrap.BOOTSTRAP_OFFSET_DAYS[ zlib.crc32(key.encode("utf-8")) % len(lot_bootstrap.BOOTSTRAP_OFFSET_DAYS) ] assert lot_bootstrap.offset_for("CUST-9527", "PROD-110022") == expected def test_lot_id_is_deterministic_and_within_column_width(self): lot_id = lot_bootstrap.bootstrap_lot_id("CUST-9527", "PROD-110022") assert lot_id == lot_bootstrap.bootstrap_lot_id("CUST-9527", "PROD-110022") assert len(lot_id) <= 64 # 超长组合不抛异常（D8：兜底补建不跳过、不阻断） long_id = lot_bootstrap.bootstrap_lot_id("C" * 64, "P" * 64) assert len(long_id) <= 64 def test_nav_approximates_cost_per_share(self): lots = lot_bootstrap.bootstrap_lots(self.HOLDING) # 82400.00 / 80000 = 1.0300（4 位 HALF_UP） assert lots[0].nav == Decimal("1.0300") def test_offset_covers_all_five_bands_across_customers(self): """错开规则必须真能覆盖 5 档，否则分档分支在兜底路径上无数据。""" offsets = { lot_bootstrap.offset_for(f"CUST-{i:04d}", "PROD-110022") for i in range(1, 60) } assert offsets == set(lot_bootstrap.BOOTSTRAP_OFFSET_DAYS) def test_forced_offset_pins_band(self): lots = lot_bootstrap.bootstrap_lots(self.HOLDING, offset_days=3) assert calc.hold_days(TRADE_DATE, lots[0].confirmed_at) == 3 assert fee.pick_fee_rate(RULES, 3) == Decimal("0.0150") def test_zero_qty_holding_returns_no_lot(self): holding = dict(self.HOLDING, qty=Decimal("0.0000")) assert lot_bootstrap.bootstrap_lots(holding) == [] def test_missing_as_of_rejected(self): holding = dict(self.HOLDING, as_of=None) with pytest.raises(ValueError): lot_bootstrap.bootstrap_lots(holding) # ── 10. 错误码映射（架构 §8.3） ────────────────────────────────────── ERROR_CASES = [ (ProductNotRedeemable, 400, "PRODUCT_NOT_REDEEMABLE"), (ProductNotSubscribable, 400, "PRODUCT_NOT_SUBSCRIBABLE"), (InsufficientShares, 400, "INSUFFICIENT_SHARES"), (BelowMinQty, 400, "BELOW_MIN_QTY"), (SameProduct, 400, "SAME_PRODUCT"), (CrossEntityNotSupported, 400, "CROSS_ENTITY_NOT_SUPPORTED"), (LotConflict, 409, "LOT_CONFLICT"), (NavNotReady, 503, "NAV_NOT_READY"), (IdempotencyUnavailable, 503, "IDEMPOTENCY_UNAVAILABLE"), (FeeRuleMissing, 500, "FEE_RULE_MISSING"), ] class TestErrorMapping: @pytest.mark.parametrize(("cls", "status", "code"), ERROR_CASES) def test_status_and_code(self, cls, status, code): err = cls() assert isinstance(err, ConvertError) assert err.status_code == status assert err.error_code == code assert err.message # 必须有可读 message，不能是空串 def test_too_many_lots_carries_batch_context(self): err = TooManyLots(batch_count=260, max_lots=200) assert err.status_code == 400 assert err.error_code == "TOO_MANY_LOTS" assert err.extra == {"batch_count": 260, "max_lots": 200} assert "260" in err.message and "200" in err.message def test_ensure_batch_limit_raises_over_limit(self): lots = [_lot(f"LOT-{i}", 100 - i, "1000.0000") for i in range(5)] plan = calc.plan_lots(lots, Decimal("5000")) calc.ensure_batch_limit(plan, 5) # 恰好等于上限 → 放行 with pytest.raises(TooManyLots) as exc: calc.ensure_batch_limit(plan, 4) assert exc.value.batch_count == 5 def test_batch_limit_checked_after_planning(self): """先规划再判上限：异常里带的是**实际所需**批次数，而非上限本身（§8.3）。""" lots = [_lot(f"LOT-{i}", 100 - i, "1000.0000") for i in range(7)] plan = calc.plan_lots(lots, Decimal("7000")) with pytest.raises(TooManyLots) as exc: calc.ensure_batch_limit(plan, 3) assert exc.value.batch_count == 7 # ── 11. 纯函数约束（DoD 第 3 条） ──────────────────────────────────── CONVERT_DIR = Path(__file__).resolve().parents[1] / "app" / "service" / "convert" #: import 行中一旦出现即说明纯函数包被 IO 污染（"不查库、不碰 SQL"是 D1/D18 的前提）。 #: 只比对 **import 语句**：docstring 里提到 `core_ro.get_nav_as_of(...)` 是在说明 #: "输入由调用方取好再传入"，属合法引用，不应误判。 FORBIDDEN_IMPORT_PREFIXES = ( "sqlalchemy", "app.repository", "app.config", "app.utils.db", "app.gateway", ) #: 全文禁用符号（连 import 都不允许，出现在任何位置都是漏了依赖边界）。 FORBIDDEN_SYMBOLS = ("get_engine", "create_engine") _IMPORT_RE = re.compile(r"^\s*(?:from|import)\s+([A-Za-z_][\w.]*)", re.M) class TestPurity: @pytest.mark.parametrize("path", sorted(CONVERT_DIR.glob("*.py")), ids=lambda p: p.name) def test_no_repository_or_engine_dependency(self, path: Path): source = path.read_text(encoding="utf-8") imported = _IMPORT_RE.findall(source) bad_imports = [ name for name in imported if name.startswith(FORBIDDEN_IMPORT_PREFIXES) ] bad_symbols = [sym for sym in FORBIDDEN_SYMBOLS if sym in source] assert not bad_imports, f"{path.name} 引入了 IO 依赖：{bad_imports}" assert not bad_symbols, f"{path.name} 出现引擎符号：{bad_symbols}" @pytest.mark.parametrize("path", sorted(CONVERT_DIR.glob("*.py")), ids=lambda p: p.name) def test_imports_only_from_stdlib_or_own_package(self, path: Path): source = path.read_text(encoding="utf-8") allowed_roots = ( "app.service.convert", "__future__", # 唯一允许的包外依赖：复用 `register_error_handlers` 的统一错误体出口 # （架构 §8.3「不新增异常出口、不改中间件」）。它是纯类型基类，不触 IO。 "app.utils.exceptions", ) stdlib = {"dataclasses", "datetime", "decimal", "typing", "zlib", "re"} external = [ name for name in _IMPORT_RE.findall(source) if not name.startswith(allowed_roots) and name.split(".")[0] not in stdlib ] assert not external, f"{path.name} 引入了包外依赖：{external}" def test_lot_bootstrap_avoids_builtin_hash(self): """内置 `hash` 受 PYTHONHASHSEED 随机化，会破坏 D18 的跨进程同源。""" source = (CONVERT_DIR / "lot_bootstrap.py").read_text(encoding="utf-8") assert "hash(" not in source