# -*- coding: utf-8 -*- """ 第十九批: 2026-08-28 全库审查修复的纯逻辑回归 —— 不联网、不连库 ===================================================================== 这一批钉住的都是当次审查改掉的真伤, 每条用例头上写清"原来错在哪": 1. 科创板最小申报统一口径 (sizer.lot_of / lot_qty 浮点容差 / split_batches 可行性线); 2. planner 取整与部分卖修正 (ceil_lot 小数截断 / _min_sell 200 股线); 3. 规则闸科创板买卖申报校验 (买 <200 拒单 / 部分卖 <200 拒、全清放行); 4. 动作引擎同轮买卖互斥 (TRIM 触发时买入侧让路, 不再自动对倒); 5. 信号口径 (_norm_conf_pct 百分制契约: 1 = 1% 不是 100%; digest 科创板卖量修正); 6. 除权调整已核销批次同步缩放 (recon.apply_ex_right); 7. 交易日历按年探测降级 (chinesecalendar 装了但没有当年数据); 8. 网格只买中枢下方 / 跟踪止盈部分卖一次性闩锁 (strategy_runner); 9. 策略买入暂停按来源分记 (strategy_service, accum 与 signal 互不误伤); 10. 宏观失败路径保留当日动作留痕 (macro_service._upsert_unavailable); 11. 双实例表名前缀 (SQL 执行唯一入口统一映射 pms_* 表名 + 旁路扫描)。 运行: python scripts/test_batch19_units.py """ import os import sys import traceback from datetime import date sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) RESULTS = [] def case(name): def deco(fn): RESULTS.append((name, fn)) return fn return deco # ================================================================ # [A] sizer: 最小申报数量的唯一出处 # ================================================================ @case("[A1] lot_of: 688/689 开头 200 股, 其余 100; 空值与 '含688不在开头' 不误判") def _(): from app.core.sizer import lot_of assert lot_of("688802.SH") == 200 and lot_of("689009.SH") == 200 assert lot_of("600000.SH") == 100 and lot_of("000001.SZ") == 100 assert lot_of("300688.SZ") == 100 # 688 在中间不算科创板 assert lot_of(None) == 100 and lot_of("") == 100 @case("[A2] lot_qty 浮点容差: 407 元买 4.07 元的票**恰好一手**, 不许被浮点误差算成零手") def _(): from app.core.sizer import lot_qty # 407/4.07 在浮点里是 99.999…, 原实现直接 int() 会把"恰好买得起一手"算成 0 手 assert lot_qty(407, 4.07) == 100, lot_qty(407, 4.07) assert lot_qty(814, 4.07) == 200 assert lot_qty(1500, 10.0) == 100 and lot_qty(999, 10.0) == 0 assert lot_qty(0, 10.0) == 0 and lot_qty(1000, 0) == 0 and lot_qty(None, 5) == 0 assert lot_qty(4000, 10.0, 200) == 400 # lot=200 时按 200 的整数倍 @case("[A3] split_batches min_lot=200: 可行性线抬到科创板 200 股, 降档与失败话术都说清") def _(): from app.core.sizer import split_batches # 50/25/25 里 25% 批只有 100 股 (<200) → 自动降档到 60/40 (300/200 股, 都合法) r = split_batches(6000, 10.0, min_lot=200) assert r["ok"] and r["scheme"] == (0.6, 0.4), r assert [b["qty"] for b in r["batches"]] == [300, 200], r # 全部阶梯都买不足 200 股 → 失败, 原因里点名科创板 r2 = split_batches(1500, 10.0, min_lot=200) assert not r2["ok"] and "科创板最少 200 股" in r2["reason"], r2 # 主板行为一字不变 r3 = split_batches(6000, 10.0) assert r3["ok"] and r3["scheme"] == (0.5, 0.25, 0.25), r3 # ================================================================ # [B] planner: 取整与科创板部分卖 # ================================================================ @case("[B1] ceil_lot 小数向上取整: 100.5 → 200 (原实现先截断再进位, 缺口永远盖不掉)") def _(): from app.core.planner import ceil_lot, floor_lot assert ceil_lot(100.5) == 200, ceil_lot(100.5) assert ceil_lot(101) == 200 and ceil_lot(100) == 100 and ceil_lot(0) == 0 assert ceil_lot(100.0000001) == 100 # 1e-9 容差防浮点噪声顶成 200 assert floor_lot(199) == 100 and floor_lot(-5) == 0 @case("[B2] _min_sell: 科创板部分卖 <200 时, 可卖够就抬到 200, 不够就本轮不切; 主板原样") def _(): from app.core.planner import _min_sell assert _min_sell(100, 1000, "688802.SH") == 200 # 抬到 200 (偏保守多卖一点) assert _min_sell(100, 150, "688802.SH") == 0 # 可卖不足 200, 这票本轮不切部分卖 assert _min_sell(200, 1000, "688802.SH") == 200 # 已合法, 原样 assert _min_sell(100, 1000, "600000.SH") == 100 # 主板不动 assert _min_sell(0, 1000, "688802.SH") == 0 @case("[B3] plan_sector_exit 停牌票不许静默消失: 无价也下整票卖单 (金额 0, 留痕说明)") def _(): from app.core.planner import plan_sector_exit r = plan_sector_exit(sector="半导体", positions=[ {"ts_code": "600000.SH", "total_qty": 6000, "price": 10.0, "sector": "半导体"}, # 停牌票: 行情缺失, price_ok=False (positions_view 拿摊薄成本顶的价) {"ts_code": "688111.SH", "total_qty": 3000, "price": 8.0, "price_ok": False, "sector": "半导体"}, ]) assert r["ok"], r ex = {i["ts_code"]: i for i in r["items"] if i["action"] == "EXIT"} assert set(ex) == {"600000.SH", "688111.SH"}, ex # 停牌票也在 assert ex["688111.SH"]["amount"] == 0.0 and ex["688111.SH"].get("need_price"), ex assert ex["688111.SH"]["qty"] == 3000 assert any("取不到现价" in n for n in r["notes"]), r["notes"] # ================================================================ # [C] 规则闸: 科创板申报数量 # ================================================================ @case("[C1] 规则闸买入: 科创板 <200 股必拒 (交易所会拒单); 主板整百照旧") def _(): from app.core import rule_gate as rg ctx = {"ts_code": "688802.SH", "position": {"total_qty": 0, "avail_qty": 0}, "caps": None, "params": {}, "flags": {}, "day": {"price": 10.0, "day_chg_from_open": 0.0, "ma5": 10.0}} r = rg.check(side="buy", action="OPEN", qty=100, price=10.0, ctx=ctx) assert any("科创板买入申报最少 200" in f for f in r["failed"]), r["failed"] r2 = rg.check(side="buy", action="OPEN", qty=200, price=10.0, ctx=ctx) assert not any("科创板" in f for f in r2["failed"]), r2["failed"] ctx3 = dict(ctx, ts_code="600000.SH") r3 = rg.check(side="buy", action="OPEN", qty=100, price=10.0, ctx=ctx3) assert not any("LOT_INVALID" in f for f in r3["failed"]), r3["failed"] @case("[C2] 规则闸卖出: 科创板部分卖 <200 拒; 余额不足 200 一次性清出放行 (零股全清)") def _(): from app.core import rule_gate as rg def sell(code, qty, total): return rg.check(side="sell", action="TRIM", qty=qty, price=10.0, ctx={"ts_code": code, "caps": None, "params": {}, "flags": {}, "position": {"total_qty": total, "avail_qty": total}, "day": {"price": 10.0}}) r = sell("688802.SH", 100, 1000) assert any("科创板部分减持最少 200" in f for f in r["failed"]), r["failed"] assert sell("688802.SH", 200, 1000)["passed"], "200 股部分卖合法" assert sell("688802.SH", 150, 150)["passed"], "余额 150 一次性清出是交易所允许的例外" assert sell("600000.SH", 100, 1000)["passed"], "主板 100 股照旧" # ================================================================ # [D] 动作引擎: 同轮买卖互斥 # ================================================================ @case("[D1] 同一只票同轮 TRIM+ADD 同时成立 → 买入侧让路, 不再自动对倒空耗手续费") def _(): from app.core import action_engine as ae params = {"scale": 2000000, "cushion_solid": 0.03, "trim_peak": 0.06, "trim_giveback": 0.5, "stock_target_default": 0.06} # 峰值 8% 回吐到 3% (过半) → TRIM 成立; 垫 3% 且创 5 日新高 → ADD 也成立 # (峰值是全时段只增不减, 加仓看近 5 日窗口 —— 两套时间基准可以同时为真) pos = {"ts_code": "600000.SH", "total_qty": 6000, "avail_qty": 6000, "cushion_pct": 0.03, "cushion_peak": 0.08, "price": 10.3, "market_value": 61800.0, "target_pct": 0.06} mkt = {"600000.SH": {"high5": 10.3, "ma5": 10.3, "tdays_since_open": None, "tdays_since_last_add": None}} r = ae.scan(positions=[pos], params=params, market=mkt) acts = [c["action"] for c in r["candidates"]] assert acts == ["TRIM"], r["candidates"] # 只留减仓, 买入侧让路 assert any(s["action"] == "ADD" and "互斥" in s["why"] for s in r["skipped"]), r["skipped"] # 对照: 峰值不足、TRIM 不触发时, ADD 照常产出 (互斥只在同轮同票双触发时生效) pos2 = dict(pos, cushion_peak=0.04) r2 = ae.scan(positions=[pos2], params=params, market=mkt) assert [c["action"] for c in r2["candidates"]] == ["ADD"], r2["candidates"] # ================================================================ # [E] 信号口径 # ================================================================ @case("[E1] _norm_conf_pct 百分制契约: 1 = 1% (原启发式把 1 当 100% 直接触发自动清仓)") def _(): from app.core.signal_rules import _norm_conf_pct assert abs(_norm_conf_pct(1) - 0.01) < 1e-12, _norm_conf_pct(1) assert abs(_norm_conf_pct(0.9) - 0.009) < 1e-12 # 0~1% 噪声级, 不再漏缩放 assert abs(_norm_conf_pct(92) - 0.92) < 1e-12 assert _norm_conf_pct(150) == 1.0 and _norm_conf_pct(-5) == 0.0 assert _norm_conf_pct("abc") == 0.0 # 解析不了按 0, 落"低于门槛"档 @case("[E2] digest 科创板中置信减持: 量抬到 200 / 持仓不足 200 退化全卖; 主板不变") def _(): from app.core import signal_rules as sr prm = {"sell_conf_min": 0.75, "auto_exit_conf": 0.85, "trim_ratio": 1 / 3} def d(code, held, conf=0.8): sig = {"source": "risk_sell", "ts_code": code, "action": "SELL", "confidence": conf} return sr.digest(sig, {"total_qty": held, "avail_qty": held}, prm) r = d("688111.SH", 300) assert r["action"] == sr.ACT_PROPOSE and r["qty"] == 200, r # 100 → 抬到 200 r2 = d("688111.SH", 150) assert r2["qty"] == 150, r2 # 不足 200: 一次性全清是合法例外 r3 = d("600000.SH", 3000) assert r3["qty"] == 1000, r3 # 主板 1/3 照旧 r4 = d("688111.SH", 3000) assert r4["qty"] == 1000, r4 # 量本来就 ≥200, 不动 # ================================================================ # [F] 除权 / 日历 # ================================================================ @case("[F1] apply_ex_right: 已部分核销的批次, closed_qty 与核销均价同比例调 (单位不混算)") def _(): from app.core.recon import apply_ex_right lots = [{"id": 1, "qty": 500, "open_price": 20.0, "closed_qty": 500, "close_avg_price": 22.0}, {"id": 2, "qty": 1000, "open_price": 18.0, "closed_qty": 0, "close_avg_price": None}] out = apply_ex_right(lots, 2.0) # 10 送 10 a, b = out[0], out[1] assert a["qty"] == 1000 and abs(a["open_price"] - 10.0) < 1e-9, a # 原来只调剩余数量: 剩余是新股数单位、已核销还是旧单位, 摊薄成本照样错 assert a["closed_qty"] == 1000 and abs(a["close_avg_price"] - 11.0) < 1e-9, a assert b["qty"] == 2000 and b["closed_qty"] == 0 and b["close_avg_price"] is None, b assert "除权调整" in a["note"] @case("[F2] 交易日历按年探测: 库装了但没当年数据 → degraded=True, 工作日放行不静默跳") def _(): import app.core.tradedays as td0 orig_has, orig_fn = td0._HAS_CAL, td0._is_workday orig_cache = dict(td0._YEAR_OK) try: td0._HAS_CAL = True def fake_workday(d): if d.year >= 2027: # 模拟: 库只有 2026 及以前的数据 raise NotImplementedError("no data for 2027") return True td0._is_workday = fake_workday td0._YEAR_OK.clear() # 原来 degraded 只看"装没装": 年初库没升级时, 全年法定节假日都被当交易日, # 页面却显示一切正常 —— 这正是最常见的降级场景。 assert td0.calendar_degraded(date(2026, 8, 28)) is False assert td0.calendar_degraded(date(2027, 1, 15)) is True assert td0.is_trade_day(date(2027, 1, 15)) is True # 周五: 降级按工作日放行 assert td0.is_trade_day(date(2027, 1, 16)) is False # 周六照样拦 assert td0._YEAR_OK.get(2027) is False and td0._YEAR_OK.get(2026) is True finally: td0._HAS_CAL, td0._is_workday = orig_has, orig_fn td0._YEAR_OK.clear() td0._YEAR_OK.update(orig_cache) # ================================================================ # [G] 策略运行侧: 网格中枢 / 止盈闩锁 # ================================================================ @case("[G1] 网格只买中枢下方: 上半区回落一档不接盘, 档位照常推进 (不再高买低不买)") def _(): from app.services import strategy_runner as srun prm = {"lower": 9.0, "upper": 11.0, "center": 10.0, "step_pct": 0.02, "per_lot": 100, "max_capital": 50000} lv = srun._grid_levels(prm) k = srun._band(lv, 10.6) # 中枢上方的一档 assert lv[k] >= 10.0, (k, lv) st = {"last_band": k + 1, "filled_levels": {}} d = srun._eval_grid({"ts_code": "600000.SH", "params": prm}, {"avail_qty": 0, "add_qty": 0}, {"price": 10.6}, None, {"state": st, "notes": [], "buy_paused": False}) assert d is None and st["last_band"] == k, (d, st) # 不买, 但档位随价下移 # 中枢下方照常接 (与 batch17 科创板用例同一条路, 这里钉主板+显式中枢) b0 = srun._band(lv, 9.5) st2 = {"last_band": b0 + 1, "filled_levels": {}} d2 = srun._eval_grid({"ts_code": "600000.SH", "params": prm}, {"avail_qty": 0, "add_qty": 0}, {"price": 9.5}, None, {"state": st2, "notes": [], "buy_paused": False}) assert d2 and d2["side"] == "buy", d2 # 没显式配 center 时用 (下界+上界)/2 兜底, 行为一致 prm2 = {"lower": 9.0, "upper": 11.0, "step_pct": 0.02, "per_lot": 100} st3 = {"last_band": k + 1, "filled_levels": {}} d3 = srun._eval_grid({"ts_code": "600000.SH", "params": prm2}, {"avail_qty": 0, "add_qty": 0}, {"price": 10.6}, None, {"state": st3, "notes": [], "buy_paused": False}) assert d3 is None and st3["last_band"] == k, (d3, st3) @case("[G2] 跟踪止盈部分卖一次性闩锁: 同一高水位只卖一次, 创新高后才许再卖; 全清不上锁") def _(): from app.services import strategy_runner as srun def trail(avail, price, state, ratio=0.5): # start_line 显式给, 别让缺省值走 param_store (这批单测不连库) st = {"ts_code": "600000.SH", "params": {"giveback": 0.05, "sell_ratio": ratio, "start_line": 0.03}} pos = {"avg_cost": 8.0, "avail_qty": avail, "total_qty": avail * 2, "cushion_pct": price / 8.0 - 1} return srun._eval_trail(st, pos, {"price": price}, None, {"state": state, "notes": []}) state = {"armed": True, "high_water": 12.0} d1 = trail(1000, 10.0, state) assert d1 and d1["qty"] == 500, d1 # 第一次回落: 卖一半 assert state.get("trail_fired_hw") == 12.0, state # 闩锁记下高水位 # 原 bug: 卖完条件仍成立, 每隔一单再卖剩余一半, 几何级联直到卖光 assert trail(500, 10.0, state) is None, "同一高水位不许再卖" trail(500, 13.0, state) # 创新高 → 高水位抬到 13 assert state["high_water"] == 13.0, state d3 = trail(500, 12.3, state) # 新一轮回落 ≥5% → 允许再卖 assert d3 and d3["qty"] == 200, d3 # 全清路径不上锁: 清仓意图失败了就该重试 state4 = {"armed": True, "high_water": 12.0, "trail_fired_hw": 12.0} d4 = trail(1000, 10.0, state4, ratio=1.0) assert d4 and d4["action"] == srun.A_EXIT, d4 # ================================================================ # [H] 策略买入暂停按来源分记 # ================================================================ @case("[H1] buypause 多来源并存: accum 解除只摘自己的, 不放开风控停的; 旧格式条目自动迁移") def _(): from app.repo import pms_repo from app.services import strategy_service as svc store = {} orig_get, orig_set = pms_repo.get_param, pms_repo.set_param orig_ls = pms_repo.list_strategies try: pms_repo.get_param = lambda k: store.get(k) pms_repo.set_param = lambda k, v, by="user": store.__setitem__(k, str(v)) or 1 pms_repo.list_strategies = ( lambda *, ts_code=None, statuses=None, limit=500, include_archived=False: [{"strategy_id": "S1", "ts_code": ts_code}]) # 旧格式条目 (没有 sources 子表) 先躺在表里 → pause_buy 迁移成 sources import json store[svc.BUYPAUSE_KEY] = json.dumps( {"600000.SH": {"reason": "旧风控", "source": "signal", "at": "2026-08-27"}}) assert svc.pause_buy("600000.SH", reason="定性失效", source="accum") == ["S1"] m = svc.buypause_map() assert set(m["600000.SH"]["sources"]) == {"signal", "accum"}, m # 原 bug: 先写先赢、后来的来源被吞 —— advisor 按 accum 解除时把风控停的也放开了 r = svc.clear_buypause("600000.SH", only_source="accum") assert r["ok"] and r["cleared"] and r["still_paused_by"] == ["signal"], r m2 = svc.buypause_map() assert "600000.SH" in m2 and set(m2["600000.SH"]["sources"]) == {"signal"}, m2 # 来源不匹配: 不动, 不算错 r2 = svc.clear_buypause("600000.SH", only_source="accum") assert r2["ok"] and r2["cleared"] is False, r2 # 无 only_source: 整条解除 r3 = svc.clear_buypause("600000.SH") assert r3["cleared"] and "600000.SH" not in svc.buypause_map(), r3 finally: pms_repo.get_param, pms_repo.set_param = orig_get, orig_set pms_repo.list_strategies = orig_ls # ================================================================ # [I] 宏观失败路径不冲留痕 # ================================================================ @case("[I1] _upsert_unavailable: 重扫失败保留当日 CMD_ISSUED/建议/周期, 不再整行重写") def _(): from app.repo import macro_repo from app.core import macro_rules as mr from app.services import macro_service as ms got = {} orig_get, orig_up = macro_repo.get_signal, macro_repo.upsert_signal try: macro_repo.get_signal = lambda key, d: { "action": "CMD_ISSUED", "ref_id": "CMD_MACRO_1", "note": "已下减仓命令", "detail": {"cycle": {"done_shift": 1}, "exit_acted": True, "advice": "减"}} macro_repo.upsert_signal = lambda **kw: got.update(kw) or 1 ms._upsert_unavailable("stock_fx_hedge", 20260828, "数据源超时") # 原来失败路径按默认值整行覆盖: 上午的 CMD_ISSUED 被冲掉 → "当日不重复下命令" # 判据失效, 次日取昨日周期也拿不到 assert got["action"] == "CMD_ISSUED" and got["ref_id"] == "CMD_MACRO_1", got assert got["zone"] == mr.Z_UNAVAILABLE and got["value"] is None, got assert got["detail"]["cycle"] == {"done_shift": 1}, got["detail"] assert got["detail"]["exit_acted"] is True and got["detail"]["advice"] == "减", got assert "重扫失败" in got["note"] and "已下减仓命令" in got["note"], got["note"] # 当日无留痕 (action=NONE) 时: 正常落 UNAVAILABLE, note 就是失败原因本身 got.clear() macro_repo.get_signal = lambda key, d: None ms._upsert_unavailable("stock_fx_hedge", 20260828, "数据源超时") assert got["action"] == "NONE" and got["note"] == "数据源超时", got finally: macro_repo.get_signal, macro_repo.upsert_signal = orig_get, orig_up # ================================================================ # [K] 双实例表名前缀: 一处映射, 不许有旁路 # ================================================================ @case("[K1] 表名映射: 唯一执行入口统一改写 pms_*; 前缀空原样; 已带前缀不二次加; 非法前缀拒绝") def _(): from app.db import session as se orig = se.TABLE_PREFIX try: se.TABLE_PREFIX = "real_" f = se.map_tables assert f("SELECT * FROM pms_command WHERE status = :s") == \ "SELECT * FROM real_pms_command WHERE status = :s" assert f("INSERT INTO pms_lot (c) VALUES (:c) ON DUPLICATE KEY UPDATE q = :q") == \ "INSERT INTO real_pms_lot (c) VALUES (:c) ON DUPLICATE KEY UPDATE q = :q" assert f("UPDATE pms_plan_snapshot SET x=1") == "UPDATE real_pms_plan_snapshot SET x=1" # DDL 与反引号写法 (init_db 建表走的就是这条路) assert f("CREATE TABLE IF NOT EXISTS `pms_qmt_order` (id INT)") == \ "CREATE TABLE IF NOT EXISTS `real_pms_qmt_order` (id INT)" # 长表名贪婪匹配整个标识符, 不会只改前半截 assert f("DELETE FROM pms_plan_snapshot") == "DELETE FROM real_pms_plan_snapshot" # 下游表不带 pms_ 前缀, 一个字不动; 大写参数键名不受影响 assert f("SELECT * FROM trading_order") == "SELECT * FROM trading_order" s = "SELECT v FROM pms_runtime_param WHERE param_key = 'PMS_TOTAL_SCALE'" assert f(s) == s.replace("pms_runtime_param", "real_pms_runtime_param"), f(s) # 幂等: 已带前缀的不会被二次加前缀 (real_pms_ 里的 pms_ 前面没有词边界) assert f("SELECT * FROM real_pms_command") == "SELECT * FROM real_pms_command" assert f(f("SELECT * FROM pms_command")) == "SELECT * FROM real_pms_command" # 映射后的 SQL 仍过单表守卫 se.assert_single_table(f("SELECT * FROM pms_command WHERE id = :i")) se.TABLE_PREFIX = "" assert f("SELECT * FROM pms_command") == "SELECT * FROM pms_command" finally: se.TABLE_PREFIX = orig # 前缀白名单: 注入形态直接拒绝 (点号/分号/引号都不许) assert se._PREFIX_RE.match("real_") and se._PREFIX_RE.match("r2_") for bad in ("a.b", "x;drop", "p'q", "1abc", "a b"): assert not se._PREFIX_RE.match(bad), bad @case("[K2] 映射不许有旁路: 自建引擎的文件不得写 pms_ 表; 直执 DDL 的脚本必须显式过映射") def _(): import re as _re root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) offenders = [] for base in ("app", "scripts"): for dp, _dn, fns in os.walk(os.path.join(root, base)): for fn in fns: if not fn.endswith(".py"): continue p = os.path.join(dp, fn) rel = p[len(root) + 1:].replace(os.sep, "/") if rel == "app/db/session.py" or rel.startswith("scripts/test_"): continue src = open(p, encoding="utf-8").read() if "create_engine" not in src: continue # 自建引擎绕开了执行入口的表名映射 —— 这样的文件绝不能碰 pms_ 表 if _re.search(r"\bpms_[a-z0-9_]+", src): offenders.append(rel) assert not offenders, ("这些文件绕开 db.session 自建引擎、又直接写 pms_ 表 SQL, " "双实例表名前缀对它们不生效: %s" % offenders) # init_db / migrate_archived_at 拿 get_engine 直执行 DDL (有意绕过单表守卫), # 它们必须显式调 map_tables —— 否则正式实例 make deploy 会建出**无前缀**的表, # 与模拟实例的表撞在一起 for f in ("scripts/init_db.py", "scripts/migrate_archived_at.py"): src = open(os.path.join(root, f), encoding="utf-8").read() assert "map_tables" in src, f + " 直执 DDL 必须过 map_tables" def main(): passed, failed = 0, [] for name, fn in RESULTS: try: fn() passed += 1 print(f" ✓ {name}") except Exception as e: failed.append((name, e)) print(f" ✗ {name}: {type(e).__name__}: {e}") traceback.print_exc() print() if failed: print(f"FAILED {len(failed)}/{len(RESULTS)}") sys.exit(1) print(f"ALL PASS ({passed} cases)") if __name__ == "__main__": main()