399 lines
21 KiB
Python
399 lines
21 KiB
Python
# -*- 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)。
|
|
运行: 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
|
|
|
|
|
|
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()
|