# -*- coding: utf-8 -*- """ 第十批单测: 静默失败专项 (零外部依赖, 不连库) ================================================= 运行: python scripts/test_batch10_units.py 2026-07-31 专项排查的产物。这一批守的不是某一条业务规则, 而是一类**故障形态**: 失败长得像成功。 本系统里大量写入函数「失败不抛异常, 只回 {"ok": False, ...} 或 0 行」。返回值一丢, 写入没发生, 而调用方照常往下走、照常回 ok=True、页面照常显示"已完成"。八条实例: 1 取不到现价时安全垫按 0 记 → 凭空触发保垫减仓 (卖出方向不设确认门槛, 直接出手) 2 不追高闸的当日涨幅恒为 None → 自主买入这道闸从来没真正跑过 3 清仓命令悄悄漏掉无价的票, 嘴上还说"全部 N 只" 4 参数表读不到时 HALT 开关按 False 放行 (fail-open) 5 连续天数的"上次推进日"键漏在白名单外 → 按日推进形同虚设 6 行业减仓算错分母 → 卖完仍超限却报 DONE 7 命令撤在途指令只改本端状态, 下游子单原封不动继续成交 8 信号去重键在落库**之前**就烧掉 → 落库失败后这条信号当天再也不会重来 [A] 组是这批里唯一一条**静态**用例: 它不测行为, 它扫源码, 守住"关键路径不许丢返回值" 这条纪律本身 —— 新加的调用点一旦又把返回值丢了, 这里立刻红。 """ import ast import os import sys sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) CASES = [] def case(name): def deco(fn): CASES.append((name, fn)) return fn return deco ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) # ================================================================ # [A] 关键路径禁止丢弃返回值 (静态扫描) # ================================================================ # 这些函数**写失败时不抛异常**, 只把失败写在返回值里。调用点必须接住。 # key = 函数名, value = 允许的调用者模块名 (None = 不限, 用于 _db(f, ...) 这种间接调用) SOFT_FAIL = { "set_param": {"param_store"}, # → {"ok": False, "error": ...} "save_neg_streak": {"portfolio"}, # → {"ok": False, "error": ...} "bump_once_guards": {"proposal_service"}, # → {"ok": False, "error": ...} "cancel_instruction": {"executor"}, # → {"ok": False, "message"|"error": ...} "dispatch": {"dispatcher"}, # → {"ok": False, "error": ...} "cancel": {"dispatcher"}, # → {"ok": False, "error": ...} "update_order": {"qmt_repo"}, # → 影响行数, 0 = 出口表里没这行 } # 确实可以丢的调用点写在这里, **必须带理由**。空着比乱加强。 ALLOWED = { # "app/xxx.py:123": "理由", } def _discarded_calls(path: str) -> list: """找出「整条语句就是一次调用、返回值没被任何人接住」的软失败调用。""" with open(path, encoding="utf-8") as f: tree = ast.parse(f.read(), path) out = [] for node in ast.walk(tree): if not isinstance(node, ast.Expr): # 表达式语句 = 返回值直接扔掉 continue call = node.value if isinstance(call, ast.Await): call = call.value if not isinstance(call, ast.Call): continue fn = call.func cand = [] if isinstance(fn, ast.Attribute): cand.append((getattr(fn.value, "id", None), fn.attr)) elif isinstance(fn, ast.Name) and fn.id == "_db" and call.args: # await _db(qmt_repo.update_order, ...) —— 线程池里跑的同一件事 a = call.args[0] if isinstance(a, ast.Attribute): cand.append((getattr(a.value, "id", None), a.attr)) for mod, attr in cand: if attr in SOFT_FAIL and (mod is None or mod in SOFT_FAIL[attr]): out.append((node.lineno, f"{mod}.{attr}")) return out @case("[A1] 关键路径禁止丢弃返回值: app/ 全库无「调用了却不看结果」的软失败写入") def _(): bad = [] for d, _dirs, files in os.walk(os.path.join(ROOT, "app")): for f in sorted(files): if not f.endswith(".py"): continue p = os.path.join(d, f) rel = os.path.relpath(p, ROOT) for lineno, what in _discarded_calls(p): if ALLOWED.get(f"{rel}:{lineno}"): continue bad.append(f"{rel}:{lineno} 丢弃了 {what}() 的返回值") assert not bad, ("以下调用点把「写失败」的返回值扔了 —— 写不进去时调用方一无所知, " "会照常报成功:\n " + "\n ".join(bad)) @case("[A2] 扫描器本身有效: 造一个丢返回值的调用, 必须扫得出来") def _(): # 守住守卫。扫描器写错 (比如 AST 节点类型判断反了) 会让 A1 永远绿, 那比没有更糟。 import tempfile src = ("def f(param_store, dispatcher, x):\n" " param_store.set_param('K', 1, 'test')\n" # ← 该被抓 " r = param_store.set_param('K', 2, 'test')\n" # ← 接住了, 放行 " if dispatcher.dispatch(x)['ok']:\n" # ← 用上了, 放行 " return r\n") with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False, encoding="utf-8") as fh: fh.write(src) tmp = fh.name try: hits = _discarded_calls(tmp) assert [h[1] for h in hits] == ["param_store.set_param"], hits assert hits[0][0] == 2, hits finally: os.unlink(tmp) @case("[A3] 软失败清单没漏: 名单里的函数确实是「不抛异常只回 ok=False」") def _(): from app.services import param_store, portfolio, proposal_service # set_param: 不可修改的键 → 回 ok=False 而不是抛 r = param_store.set_param("PROXY_DB_URL", "x") assert isinstance(r, dict) and r.get("ok") is False and r.get("error"), r # 另外两个的失败分支在 [E]/[F] 组用桩验证, 这里只钉住"返回的是 dict 不是 None" assert portfolio.save_neg_streak.__doc__, "save_neg_streak 必须写明返回成败" assert proposal_service.bump_once_guards.__doc__ # ================================================================ # [B] 开关命令: 参数没写进去就不许报"已完成" # ================================================================ @case("[B1] HALT_BUY 参数写入失败 → 命令置 CANCELLED 并回 ok=False (不能报已完成)") def _(): from test_wiring import install_fakes from app.services import command_service, param_store from app.core import command_spec as cs install_fakes(prices={}) orig = param_store.set_param try: param_store.set_param = lambda k, v, by="user": {"ok": False, "error": "库挂了"} r = command_service.issue("HALT_BUY", {}, issued_by="test") # issue 内部即时排方案 # 刹车没踩上, 就不许说"已完成" assert r["ok"] is False, r assert r["status"] == cs.ST_CANCELLED, r assert "库挂了" in (r["errors"][0] if r["errors"] else ""), r assert any("开关未生效" in str(n) for n in r["plan"]["notes"]), r["plan"]["notes"] finally: param_store.set_param = orig @case("[B2] 参数写得进去时 HALT_BUY 照常 DONE (修复不能把正常路堵死)") def _(): from test_wiring import install_fakes from app.services import command_service, param_store from app.core import command_spec as cs fake = install_fakes(prices={}) r = command_service.issue("HALT_BUY", {}, issued_by="test") assert r["ok"] and r["status"] == cs.ST_DONE, r assert param_store.get("PMS_GLOBAL_BUY_HALT") is True, fake.params @case("[B3] instant 命令自身失败 (撤销一个不存在的命令) → CANCELLED 而不是 DONE") def _(): from test_wiring import install_fakes from app.services import command_service from app.core import command_spec as cs install_fakes(prices={}) r = command_service.issue("CANCEL_COMMAND", {"target_command_id": "CMD_NOPE_0001"}, issued_by="test") assert r["ok"] is False and r["status"] == cs.ST_CANCELLED, r assert "不存在" in str(r["errors"]), r # ================================================================ # [C] 撤在途指令: 下游拒了必须露出来 # ================================================================ @case("[C1] 撤单被下游拒 → 不计入 cancelled, 顶到 notes 第一条") def _(): from test_wiring import install_fakes from app.services import command_service, executor install_fakes(prices={}) calls = [] def fake_cancel(iid, reason=None): calls.append(iid) return {"ok": False, "message": f"下游拒绝撤单: {iid}"} orig = executor.cancel_instruction try: executor.cancel_instruction = fake_cancel items = [{"ts_code": "600000.SH", "action": "HALT", "cancel_instruction_id": "INS_A"}, {"ts_code": "000001.SZ", "action": "HALT", "cancel_instruction_id": "INS_B"}] r = command_service._cancel_marked_instructions(items) assert calls == ["INS_A", "INS_B"], calls # 真的调了下游, 不是只改本端 assert r["cancelled"] == [], r assert [x["instruction_id"] for x in r["failed"]] == ["INS_A", "INS_B"], r assert "下游拒绝撤单" in r["failed"][0]["why"], r finally: executor.cancel_instruction = orig @case("[C2] 部分成功: cancelled 数的是真撤掉的那些, 不是点名的条数") def _(): from test_wiring import install_fakes from app.services import command_service, executor install_fakes(prices={}) orig = executor.cancel_instruction try: executor.cancel_instruction = ( lambda iid, reason=None: {"ok": iid == "INS_OK", "message": "no"}) r = command_service._cancel_marked_instructions( [{"cancel_instruction_id": "INS_OK"}, {"cancel_instruction_id": "INS_BAD"}]) assert r["cancelled"] == ["INS_OK"] and len(r["failed"]) == 1, r finally: executor.cancel_instruction = orig @case("[C3] 方案生成器的 notes 措辞是「点名撤销」而不是「已撤销」") def _(): from app.core import planner as pl r = pl.plan_halt_buy(pending_buys=[{"ts_code": "600000.SH", "qty": 100, "instruction_id": "INS_1"}]) assert r["items"][0]["cancel_instruction_id"] == "INS_1", r # planner 还不知道撤没撤成, 不许承诺结果 assert "点名" in r["notes"][0], r["notes"] # ================================================================ # [D] 组合刹车: 该踩没踩上必须报错 # ================================================================ @case("[D1] PMS_BRAKE_UNTIL 写入失败 → 盘前准备报 errors, 不许 ok=True") def _(): from test_wiring import install_fakes from app.services import ledger_service as ls, param_store, portfolio install_fakes(prices={}) orig_set, orig_view = param_store.set_param, portfolio.positions_view try: # 高水位 100 万, 现值 80 万 → 回撤 20%, 远超默认 5%, 必须踩刹车 portfolio.positions_view = lambda **kw: { "held": [], "positions": [], "params": {}, "totals": {"portfolio_mv": 800000.0}} param_store.set_param = lambda k, v, by="user": ( {"ok": True} if k == "PMS_HIGH_WATER" else {"ok": False, "error": "库挂了"}) orig_get = param_store.get_float param_store.get_float = lambda k, d=0.0: (1000000.0 if k == "PMS_HIGH_WATER" else orig_get(k, d)) try: b = ls._settle_brake() finally: param_store.get_float = orig_get assert b["engaged"] is False, b # 没踩上就不能说踩上了 assert any("刹车未生效" in w for w in b.get("warnings", [])), b finally: param_store.set_param, portfolio.positions_view = orig_set, orig_view @case("[D2] 高水位写入失败 → 回撤按旧高点算并留 warning, 不静默") def _(): from test_wiring import install_fakes from app.services import ledger_service as ls, param_store, portfolio install_fakes(prices={}) orig_set, orig_view = param_store.set_param, portfolio.positions_view try: portfolio.positions_view = lambda **kw: { "held": [], "positions": [], "params": {}, "totals": {"portfolio_mv": 500000.0}} param_store.set_param = lambda k, v, by="user": {"ok": False, "error": "库挂了"} b = ls._settle_brake() assert b["high_water"] == 0.0, b # 没写进去就不许当成写进去了 assert any("高水位" in w for w in b.get("warnings", [])), b finally: param_store.set_param, portfolio.positions_view = orig_set, orig_view # ================================================================ # [E] 安全垫连负天数: 写不上要让日终结算整体报失败 # ================================================================ @case("[E1] save_neg_streak 失败 → 返回 ok=False (不再吞成 warning 回 None)") def _(): from test_wiring import install_fakes from app.services import portfolio from app.repo import pms_repo install_fakes(prices={}) orig = pms_repo.set_param try: def boom(*a, **kw): raise RuntimeError("库挂了") pms_repo.set_param = boom r = portfolio.save_neg_streak({"600000.SH": 3}) assert r["ok"] is False and "库挂了" in r["error"], r finally: pms_repo.set_param = orig @case("[E2] 连负天数写不上 → daily_settle 的 ok 必须是 False") def _(): from test_wiring import install_fakes from app.services import ledger_service as ls, portfolio install_fakes(prices={}) orig = portfolio.save_neg_streak try: portfolio.save_neg_streak = lambda m: {"ok": False, "error": "库挂了"} out = ls.daily_settle() assert out["ok"] is False, out assert any("安全垫" in e for e in out["errors"]), out["errors"] finally: portfolio.save_neg_streak = orig # ================================================================ # [F] 一次性守卫计数器: 没写上要留痕 # ================================================================ @case("[F1] update_position 影响 0 行 → bump_once_guards 回 ok=False") def _(): from test_wiring import install_fakes from app.services import proposal_service from app.repo import pms_repo install_fakes(prices={}) orig = pms_repo.update_position try: pms_repo.update_position = lambda code, **kw: 0 # 持仓行不存在 r = proposal_service.bump_once_guards("600000.SH", "FILL") assert r["ok"] is False and "0 行" in r["error"], r # 不涉及计数器的动作不该被误判成失败 assert proposal_service.bump_once_guards("600000.SH", "TRIM")["ok"] is True finally: pms_repo.update_position = orig @case("[F2] 计数器没写上 → 评审账本留一条 WARN 痕 (指令仍落表, 但纪律失效要有人知道)") def _(): from test_wiring import install_fakes from app.services import proposal_service from datetime import datetime fake = install_fakes(prices={"600000.SH": 10.0}) orig = proposal_service.bump_once_guards try: proposal_service.bump_once_guards = ( lambda code, act, hn=None, now=None: {"ok": False, "fields": {"fill_count": 1}, "error": "库挂了"}) iid = proposal_service._make_instruction( {"ts_code": "600000.SH", "action": "FILL", "side": "buy", "qty": 100, "reason": "测试", "hard_numbers": {}}, 10.0, datetime.now()) assert iid, iid warn = [x for x in fake.ledger if x.get("verdict") == "WARN" and "一次性守卫" in str(x.get("reason"))] assert len(warn) == 1, fake.ledger assert warn[0]["ref_id"] == iid, warn finally: proposal_service.bump_once_guards = orig # ================================================================ # [G] 信号去重键: 落库成功之后才算用掉 # ================================================================ @case("[G1] 落指令抛异常 → 当天的去重键不许被烧掉 (下一跳还能重来)") def _(): from test_wiring import install_fakes from app.services import signal_service from app.core import signal_rules as sr install_fakes(prices={"600000.SH": 10.0}) seen, out = set(), {"ignored": 0, "recorded": 0, "exits": [], "proposals": [], "errors": []} sig = {"msg_id": "M1", "ts_code": "600000.SH", "action": "SELL", "confidence": 0.95, "source": "test", "reason": "风控"} view = {"positions": [{"ts_code": "600000.SH", "total_qty": 1000, "avail_qty": 1000, "price": 10.0}]} prm = {} d = sr.digest(sig, view["positions"][0], prm) if d["action"] != sr.ACT_EXIT: return # 规则口径变了就跳过, 不假装测到了 key = sr.dedup_key(sig, 20260731) orig = signal_service._make_exit try: def boom(*a, **kw): raise RuntimeError("库挂了") signal_service._make_exit = boom try: signal_service._handle(sig, view, prm, seen, 20260731, False, out) except RuntimeError: pass assert key not in seen, ("落库失败却把去重键用掉了 —— 这条风控卖出信号今天" "再也不会被消化, 而页面只多一行 error") finally: signal_service._make_exit = orig @case("[G2] 落成功后去重键照常生效 (修复不能把去重关掉)") def _(): from test_wiring import install_fakes from app.services import signal_service from app.core import signal_rules as sr install_fakes(prices={"600000.SH": 10.0}) seen, out = set(), {"ignored": 0, "recorded": 0, "exits": [], "proposals": [], "errors": []} sig = {"msg_id": "M1", "ts_code": "600000.SH", "action": "SELL", "confidence": 0.95, "source": "test", "reason": "风控"} view = {"positions": [{"ts_code": "600000.SH", "total_qty": 1000, "avail_qty": 1000, "price": 10.0}]} d = sr.digest(sig, view["positions"][0], {}) if d["action"] != sr.ACT_EXIT: return orig = signal_service._has_inflight try: signal_service._has_inflight = lambda c: False signal_service._handle(sig, view, {}, seen, 20260731, False, out) assert len(out["exits"]) == 1, out assert sr.dedup_key(sig, 20260731) in seen, seen signal_service._handle(sig, view, {}, seen, 20260731, False, out) assert len(out["exits"]) == 1 and out["ignored"] == 1, out # 第二次被去重挡掉 finally: signal_service._has_inflight = orig # ================================================================ # [H] 日终结算: 对账拒绝时不许报 ok=True # ================================================================ @case("[H1] reconcile 回 ok=False → daily_settle 必须 ok=False 并说明原因") def _(): from test_wiring import install_fakes from app.services import ledger_service as ls install_fakes(prices={}) orig = ls.reconcile try: ls.reconcile = lambda **kw: {"ok": False, "errors": ["两个源都无应答"], "diffs": [], "fixes": []} out = ls.daily_settle() assert out["ok"] is False, out assert any("对账未完成" in e and "无应答" in e for e in out["errors"]), out["errors"] finally: ls.reconcile = orig @case("[H2] 对账连续不一致升到 ERROR → daily_settle 同样不许报成功") def _(): from test_wiring import install_fakes from app.services import ledger_service as ls install_fakes(prices={}) orig = ls.reconcile try: ls.reconcile = lambda **kw: {"ok": True, "severity": "ERROR", "streak": 3, "diffs": [{"ts_code": "600000.SH"}], "fixes": []} out = ls.daily_settle() assert out["ok"] is False, out assert any("连续 3 日" in e for e in out["errors"]), out["errors"] finally: ls.reconcile = orig # ================================================================ # [I] 取不到现价的票: 不许拿成本价冒充, 更不许凭空算出安全垫 # ================================================================ @case("[I1] 无价的票 cushion_pct 必须是 None, 不能是 0 (0 会被读成「不赚不亏」)") def _(): from test_wiring import install_fakes from app.services import portfolio fake = install_fakes(prices={"600000.SH": 12.0}) # 000001.SZ 故意没价 for code, cost in (("600000.SH", 10.0), ("000001.SZ", 10.0)): fake.insert_lot(ts_code=code, lot_type="BASE", qty=1000, open_price=cost, open_date="2026-07-01") fake.update_position(code, total_qty=1000, avail_qty=1000, avg_cost=cost) v = portfolio.positions_view() by = {x["ts_code"]: x for x in v["held"]} assert by["600000.SH"]["price_ok"] is True assert abs(by["600000.SH"]["cushion_pct"] - 0.2) < 1e-6, by["600000.SH"] # 取不到价的那只: price 用成本顶着好让市值不塌, 但垫子必须是"不知道" assert by["000001.SZ"]["price_ok"] is False, by["000001.SZ"] assert by["000001.SZ"]["cushion_pct"] is None, by["000001.SZ"] assert "000001.SZ" in v["price_missing"], v.get("price_missing") @case("[I2] 动作引擎跳过无价的票, 且**跳过这件事本身是可见的**") def _(): from app.core import action_engine as ae r = ae.scan(positions=[{"ts_code": "000001.SZ", "total_qty": 1000, "avail_qty": 1000, "avg_cost": 10.0, "price": 10.0, "price_ok": False, "cushion_pct": None, "cushion_peak": 0.0}], params={}, market={}) assert not r["candidates"], r assert any(s["ts_code"] == "000001.SZ" and "取不到现价" in s["why"] for s in r.get("skipped", [])), r @case("[I3] 清仓命令不许悄悄漏掉无价的票, 只数要对得上持仓只数") def _(): from app.core import planner as pl positions = [{"ts_code": "600000.SH", "total_qty": 1000, "avail_qty": 1000, "price": 10.0, "price_ok": True}, {"ts_code": "000001.SZ", "total_qty": 500, "avail_qty": 500, "price": 10.0, "price_ok": False}] r = pl.plan_liquidate_all(positions=positions, pending_buys=[]) codes = {i["ts_code"] for i in r["items"] if i["action"] in ("EXIT", "SELL")} assert codes == {"600000.SH", "000001.SZ"}, r["items"] need = [i for i in r["items"] if i.get("need_price")] assert [i["ts_code"] for i in need] == ["000001.SZ"], need assert r["ok"] is True, r # ================================================================ # [J] 不追高闸: 拿不到数就说拿不到, 不许当成"通过" # ================================================================ @case("[J1] 当日涨幅超上限 → NO_CHASE_DAYUP 拦住 (这道闸得真能拦)") def _(): from app.core import rule_gate as rg r = rg.check(side="buy", action="FILL", qty=100, price=11.0, ctx={"position": {"total_qty": 0, "avail_qty": 0}, "caps": None, "params": {"buy_halt_dayup": 0.05}, "day": {"price": 11.0, "day_chg_from_open": 0.09, "ma5": 11.0}, "flags": {}}) assert any("NO_CHASE_DAYUP" in f for f in r["failed"]), r @case("[J2] 取不到当日涨幅 → 留 DAYUP_MISSING 警示, 绝不当成校验通过") def _(): from app.core import rule_gate as rg r = rg.check(side="buy", action="FILL", qty=100, price=11.0, ctx={"position": {"total_qty": 0, "avail_qty": 0}, "caps": None, "params": {"buy_halt_dayup": 0.05}, "day": {"price": 11.0, "day_chg_from_open": None, "ma5": 11.0}, "flags": {}}) assert not any("NO_CHASE_DAYUP" in f for f in r["failed"]), r assert any("DAYUP_MISSING" in w for w in r["warnings"]), r @case("[J3] 自主提议给规则闸的 day 必须是真行情, 不是拿 price 拼出来的空壳") def _(): import inspect from app.services import proposal_service # 曾经是 {"vwap": price, "day_chg_from_open": None} —— 当日涨幅恒 None, 于是 # "不追高(涨幅)"这一项对所有自主买入从来没有真正跑过。这里直接验行为: 造一只 # 当日大涨的票, 走 _route_one, 规则闸必须拿到真涨幅并拦下来。 from datetime import datetime from test_wiring import install_fakes from app.services import market, proposal_service install_fakes(prices={"600000.SH": 11.0}) orig = market.day_snapshot try: market.day_snapshot = lambda c: {"price": 11.0, "vwap": 10.8, "open": 10.0, "day_chg_from_open": 0.10, "bars": 60} mkt = proposal_service._market_ctx([{"ts_code": "600000.SH"}], datetime.now()) assert mkt["600000.SH"]["day"]["day_chg_from_open"] == 0.10, mkt # 取快照抛异常也不能让整轮扫描崩, 但要留空让规则闸记 DAYUP_MISSING def boom(c): raise RuntimeError("行情库不可用") market.day_snapshot = boom mkt = proposal_service._market_ctx([{"ts_code": "600000.SH"}], datetime.now()) assert mkt["600000.SH"]["day"] == {}, mkt finally: market.day_snapshot = orig # ================================================================ # [K] 参数表读不到时, 安全开关按"拦"而不是按默认值放行 # ================================================================ @case("[K1] 参数表读失败 → HALT 开关 fail-closed 取 True (宁可多拦一轮)") def _(): from test_wiring import install_fakes from app.services import param_store from app.repo import pms_repo install_fakes(prices={}) orig, snap = pms_repo.all_params, dict(param_store._cache) try: def boom(): raise RuntimeError("DB 挂了") pms_repo.all_params = boom param_store.refresh(force=True) assert param_store._cache["error"], param_store._cache assert param_store.get("PMS_GLOBAL_BUY_HALT") is True assert param_store.get("PMS_GLOBAL_EXEC_HALT") is True # 非安全开关不受影响, 照常回初值 —— fail-closed 只用在"拦得住"的地方 assert param_store.get("PMS_AUTONOMY") in ("full", "propose_only", "off") finally: pms_repo.all_params = orig param_store._cache.clear() param_store._cache.update(snap) @case("[K2] 表读得到、只是没设过这个键 → 照常走默认值 (不能把'没设'当成'读不到')") def _(): from test_wiring import install_fakes from app.services import param_store install_fakes(prices={}) # 空参数表, 但**读得到** param_store.refresh(force=True) assert not param_store._cache["error"], param_store._cache assert param_store.get("PMS_GLOBAL_BUY_HALT") is not True, "误伤: 没设过被当成读不到" # ================================================================ def main(): ok = fail = 0 for name, fn in CASES: try: fn() print(f" ok {name}") ok += 1 except Exception as e: print(f" FAIL {name}\n {type(e).__name__}: {e}") fail += 1 print("-" * 62) print(f"通过 {ok} 例, 失败 {fail} 例") if fail: print("BATCH10 FAIL") sys.exit(1) print("BATCH10 PASS") if __name__ == "__main__": main()