393 lines
17 KiB
Python
393 lines
17 KiB
Python
|
# -*- coding: utf-8 -*-
|
|||
|
|
"""
|
|||
|
|
第十二批模块单测 (实机运行, 零外部依赖, 不连库)
|
|||
|
|
==================================================
|
|||
|
|
运行: 在 tradingSystem 仓库根目录执行 python scripts/test_batch12_units.py
|
|||
|
|
覆盖: 自主新建仓 (动作引擎 OPEN) 的选票、数量、名额与金额的滚动扣减、行业集中度滚动;
|
|||
|
|
研判请求的硬数字裁剪 (仓位数字不送决策系统);
|
|||
|
|
参考位漂移的当日首答锁定与偏离即停;
|
|||
|
|
研判的单轮时间预算。
|
|||
|
|
约定同前: 全过输出 "ALL PASS (n cases)" 退出码 0。
|
|||
|
|
"""
|
|||
|
|
import os
|
|||
|
|
import sys
|
|||
|
|
import traceback
|
|||
|
|
|
|||
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|||
|
|
|
|||
|
|
from app.core import action_engine as ae # noqa: E402
|
|||
|
|
from app.services import exec_advisor as ea # noqa: E402
|
|||
|
|
from app.services import judge as jd # noqa: E402
|
|||
|
|
from app.services import param_store as ps # noqa: E402
|
|||
|
|
|
|||
|
|
RESULTS = []
|
|||
|
|
|
|||
|
|
|
|||
|
|
def case(name):
|
|||
|
|
def deco(fn):
|
|||
|
|
RESULTS.append((name, fn))
|
|||
|
|
return fn
|
|||
|
|
return deco
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ================================================================ 夹具
|
|||
|
|
def params(**kw):
|
|||
|
|
"""action_engine 那侧要的参数快照 (口径同 param_store.sizing_params)。"""
|
|||
|
|
p = {"scale": 2_000_000.0, "stock_target_default": 0.06,
|
|||
|
|
"batch_split": (0.5, 0.25, 0.25), "min_lot_merge": True}
|
|||
|
|
p.update(kw)
|
|||
|
|
return p
|
|||
|
|
|
|||
|
|
|
|||
|
|
def caps(**kw):
|
|||
|
|
"""portfolio.caps_ctx 的产出形状 (组合快照)。"""
|
|||
|
|
c = {"scale": 2_000_000.0, "portfolio_cap": 0.70, "stock_cap": 0.08,
|
|||
|
|
"max_names": 20, "portfolio_mv": 0.0, "names_count": 0,
|
|||
|
|
"stock_mv": 0.0, "is_new_name": False, "sector": None,
|
|||
|
|
"sector_names": 0, "sector_mv": 0.0,
|
|||
|
|
"sector_names_map": {}, "sector_mv_map": {},
|
|||
|
|
"sector_max_names": 4, "sector_max_ratio": 0.40,
|
|||
|
|
"cash_reserve": 0.0, "sector_source_ready": True,
|
|||
|
|
"cash_avail": None, "cash_source": "estimate"}
|
|||
|
|
c.update(kw)
|
|||
|
|
return c
|
|||
|
|
|
|||
|
|
|
|||
|
|
def cand(code, price=10.0, score=100.0, sector=None, rank=1, **kw):
|
|||
|
|
d = {"ts_code": code, "price": price, "score": score, "sector": sector,
|
|||
|
|
"theme": kw.pop("theme", None), "tier": kw.pop("tier", "强传导"),
|
|||
|
|
"upside": kw.pop("upside", 0.3), "heat": kw.pop("heat", 0.5),
|
|||
|
|
"rank": rank, "bucket": "main", "src": "plan_api"}
|
|||
|
|
d.update(kw)
|
|||
|
|
return d
|
|||
|
|
|
|||
|
|
|
|||
|
|
class _PatchParams:
|
|||
|
|
"""把 param_store 的取值函数临时换成一份字典 —— 单测不连库。"""
|
|||
|
|
|
|||
|
|
def __init__(self, values):
|
|||
|
|
self.values = values
|
|||
|
|
self.saved = {}
|
|||
|
|
|
|||
|
|
def __enter__(self):
|
|||
|
|
for name in ("get", "get_int", "get_float", "get_bool"):
|
|||
|
|
self.saved[name] = getattr(ps, name)
|
|||
|
|
v = self.values
|
|||
|
|
ps.get = lambda k, d=None: v.get(k, d)
|
|||
|
|
ps.get_int = lambda k, d=0: int(v.get(k, d))
|
|||
|
|
ps.get_float = lambda k, d=0.0: float(v.get(k, d))
|
|||
|
|
ps.get_bool = lambda k, d=False: bool(v.get(k, d))
|
|||
|
|
return self
|
|||
|
|
|
|||
|
|
def __exit__(self, *a):
|
|||
|
|
for name, fn in self.saved.items():
|
|||
|
|
setattr(ps, name, fn)
|
|||
|
|
return False
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ================================================================ 动作引擎: 新建仓
|
|||
|
|
@case("新建仓·正常产出: 只提底仓批, 数量按 50% 取整到一手")
|
|||
|
|
def _():
|
|||
|
|
# 目标 6% × 200 万 = 12 万; 底仓 50% = 6 万; 现价 10 元 → 6000 股
|
|||
|
|
r = ae.scan_open(candidates=[cand("600000.SH")], params=params(), caps=caps(),
|
|||
|
|
room_amt=1_400_000, slots=20)
|
|||
|
|
assert len(r["candidates"]) == 1, r
|
|||
|
|
c = r["candidates"][0]
|
|||
|
|
assert c["action"] == "OPEN" and c["side"] == "buy", c
|
|||
|
|
assert c["qty"] == 6000, c
|
|||
|
|
assert c["judge_required"] is True, "OPEN 必须有资格送研判闸"
|
|||
|
|
assert c["needs_user_confirm"] is False, "新建仓不强制人工确认 (档位说了算)"
|
|||
|
|
# 现价与行业必须挂在候选上 —— 新票在账本里没有行, 服务层取不到这两样
|
|||
|
|
assert c["price"] == 10.0 and "sector" in c, c
|
|||
|
|
hn = c["hard_numbers"]
|
|||
|
|
assert hn["score"] == 100.0 and hn["price"] == 10.0, hn
|
|||
|
|
assert hn["target_amount"] == 120000.0, hn # 三批取整后正好 12 万
|
|||
|
|
assert hn["base_amount"] == 60000.0, hn
|
|||
|
|
|
|||
|
|
|
|||
|
|
@case("新建仓·没有名额时一条不产, 且说得出为什么")
|
|||
|
|
def _():
|
|||
|
|
r = ae.scan_open(candidates=[cand("600000.SH")], params=params(),
|
|||
|
|
caps=caps(names_count=20), room_amt=1_400_000, slots=0)
|
|||
|
|
assert r["candidates"] == [], r
|
|||
|
|
assert "名额" in r["skipped"][0]["why"], r["skipped"]
|
|||
|
|
|
|||
|
|
|
|||
|
|
@case("新建仓·没有可投金额时一条不产, 且说得出为什么")
|
|||
|
|
def _():
|
|||
|
|
r = ae.scan_open(candidates=[cand("600000.SH")], params=params(), caps=caps(),
|
|||
|
|
room_amt=0, slots=20)
|
|||
|
|
assert r["candidates"] == [], r
|
|||
|
|
assert "可投金额" in r["skipped"][0]["why"], r["skipped"]
|
|||
|
|
|
|||
|
|
|
|||
|
|
@case("新建仓·取不到实时价整只跳过并留痕 (昨收不能拿来下单)")
|
|||
|
|
def _():
|
|||
|
|
r = ae.scan_open(candidates=[cand("600000.SH", price=None)], params=params(),
|
|||
|
|
caps=caps(), room_amt=1_400_000, slots=20)
|
|||
|
|
assert r["candidates"] == [], r
|
|||
|
|
assert "实时价" in r["skipped"][0]["why"], r["skipped"]
|
|||
|
|
|
|||
|
|
|
|||
|
|
@case("新建仓·底仓批买不足一手 → 不提, 原因是一手不是别的")
|
|||
|
|
def _():
|
|||
|
|
# 目标 6% × 10 万 = 6000 元; 现价 500 元一手就要 5 万 → 拆不出
|
|||
|
|
r = ae.scan_open(candidates=[cand("600000.SH", price=500.0)],
|
|||
|
|
params=params(scale=100_000.0), caps=caps(scale=100_000.0),
|
|||
|
|
room_amt=70_000, slots=20)
|
|||
|
|
assert r["candidates"] == [], r
|
|||
|
|
assert "一手" in r["skipped"][0]["why"], r["skipped"]
|
|||
|
|
|
|||
|
|
|
|||
|
|
@case("新建仓·名额边走边扣: 只剩一个名额就只产一条, 其余留到下一跳")
|
|||
|
|
def _():
|
|||
|
|
cs = [cand("600000.SH", score=300, rank=1), cand("600001.SH", score=200, rank=2),
|
|||
|
|
cand("600002.SH", score=100, rank=3)]
|
|||
|
|
r = ae.scan_open(candidates=cs, params=params(), caps=caps(names_count=19),
|
|||
|
|
room_amt=1_400_000, slots=1)
|
|||
|
|
assert [c["ts_code"] for c in r["candidates"]] == ["600000.SH"], r["candidates"]
|
|||
|
|
whys = [s["why"] for s in r["skipped"]]
|
|||
|
|
assert any("名额已用完" in w for w in whys), whys
|
|||
|
|
|
|||
|
|
|
|||
|
|
@case("新建仓·金额边走边扣: 三条候选只够两条, 第三条被金额挡下")
|
|||
|
|
def _():
|
|||
|
|
# 每条整只目标 12 万; 可投 25 万 → 两条 (24 万) 之后只剩 1 万, 不够一整只
|
|||
|
|
cs = [cand("600000.SH", score=300, rank=1), cand("600001.SH", score=200, rank=2),
|
|||
|
|
cand("600002.SH", score=100, rank=3)]
|
|||
|
|
r = ae.scan_open(candidates=cs, params=params(), caps=caps(),
|
|||
|
|
room_amt=250_000, slots=20)
|
|||
|
|
assert [c["ts_code"] for c in r["candidates"]] == ["600000.SH", "600001.SH"], r["candidates"]
|
|||
|
|
whys = " ".join(s["why"] for s in r["skipped"])
|
|||
|
|
assert "不开半截新仓" in whys, whys
|
|||
|
|
|
|||
|
|
|
|||
|
|
@case("新建仓·钱不够一整只就不开 (不拿名额去换零头仓位)")
|
|||
|
|
def _():
|
|||
|
|
# 目标 12 万, 只剩 5 万 —— 拆得出一手批次, 但那是个补不到目标的零头仓位
|
|||
|
|
r = ae.scan_open(candidates=[cand("600000.SH")], params=params(), caps=caps(),
|
|||
|
|
room_amt=50_000, slots=20)
|
|||
|
|
assert r["candidates"] == [], r
|
|||
|
|
assert "不开半截新仓" in r["skipped"][0]["why"], r["skipped"]
|
|||
|
|
|
|||
|
|
|
|||
|
|
@case("新建仓·滚动扣减真的生效: 组合上限只容得下两条时不会三条都放出来")
|
|||
|
|
def _():
|
|||
|
|
# 总仓上限 70% × 200 万 = 140 万, 已有 116 万 → 只剩 24 万 = 两只
|
|||
|
|
# 如果不滚动 (每条都拿同一份旧快照算), 三条会各自都通过 —— 这一例就是为它写的
|
|||
|
|
cs = [cand(f"60000{i}.SH", score=300 - i, rank=i + 1) for i in range(3)]
|
|||
|
|
r = ae.scan_open(candidates=cs, params=params(),
|
|||
|
|
caps=caps(portfolio_mv=1_160_000, names_count=5),
|
|||
|
|
room_amt=240_000, slots=20)
|
|||
|
|
assert len(r["candidates"]) == 2, [c["ts_code"] for c in r["candidates"]]
|
|||
|
|
|
|||
|
|
|
|||
|
|
@case("新建仓·行业集中度也滚动: 同行业名额只剩一个时第二只被拦")
|
|||
|
|
def _():
|
|||
|
|
cs = [cand("600000.SH", score=300, rank=1, sector="储能"),
|
|||
|
|
cand("600001.SH", score=200, rank=2, sector="储能")]
|
|||
|
|
r = ae.scan_open(candidates=cs, params=params(),
|
|||
|
|
caps=caps(sector_max_names=1), room_amt=1_400_000, slots=20)
|
|||
|
|
assert len(r["candidates"]) == 1, [c["ts_code"] for c in r["candidates"]]
|
|||
|
|
whys = " ".join(s["why"] for s in r["skipped"])
|
|||
|
|
assert "SECTOR_NAMES" in whys, whys
|
|||
|
|
|
|||
|
|
|
|||
|
|
@case("新建仓·行业数据源没配时不误拦 (sector 一律按空处理)")
|
|||
|
|
def _():
|
|||
|
|
cs = [cand("600000.SH", score=300, rank=1, sector="储能"),
|
|||
|
|
cand("600001.SH", score=200, rank=2, sector="储能")]
|
|||
|
|
r = ae.scan_open(candidates=cs, params=params(),
|
|||
|
|
caps=caps(sector_max_names=1, sector_source_ready=False),
|
|||
|
|
room_amt=1_400_000, slots=20)
|
|||
|
|
assert len(r["candidates"]) == 2, [c["ts_code"] for c in r["candidates"]]
|
|||
|
|
|
|||
|
|
|
|||
|
|
@case("新建仓·已有在途或今日被拒过的票不重复提")
|
|||
|
|
def _():
|
|||
|
|
r = ae.scan_open(candidates=[cand("600000.SH")], params=params(), caps=caps(),
|
|||
|
|
room_amt=1_400_000, slots=20, skip={("600000.SH", "OPEN")})
|
|||
|
|
assert r["candidates"] == [], r
|
|||
|
|
assert "在途" in r["skipped"][0]["why"], r["skipped"]
|
|||
|
|
|
|||
|
|
|
|||
|
|
@case("新建仓·单票评估异常不拖垮整轮, 后面的票照常评")
|
|||
|
|
def _():
|
|||
|
|
orig = ae.eval_open
|
|||
|
|
|
|||
|
|
def boom(c, *a, **kw):
|
|||
|
|
if c["ts_code"] == "600000.SH":
|
|||
|
|
raise RuntimeError("求值炸了")
|
|||
|
|
return orig(c, *a, **kw)
|
|||
|
|
|
|||
|
|
ae.eval_open = boom
|
|||
|
|
try:
|
|||
|
|
cs = [cand("600000.SH", score=300, rank=1), cand("600001.SH", score=200, rank=2)]
|
|||
|
|
r = ae.scan_open(candidates=cs, params=params(), caps=caps(),
|
|||
|
|
room_amt=1_400_000, slots=20)
|
|||
|
|
finally:
|
|||
|
|
ae.eval_open = orig
|
|||
|
|
assert [c["ts_code"] for c in r["candidates"]] == ["600001.SH"], r["candidates"]
|
|||
|
|
whys = " ".join(s["why"] for s in r["skipped"])
|
|||
|
|
assert "评估异常" in whys and "RuntimeError" in whys, whys
|
|||
|
|
|
|||
|
|
|
|||
|
|
@case("回归·既有四类动作的判据与研判范围没被动过")
|
|||
|
|
def _():
|
|||
|
|
assert ae.JUDGE_ACTIONS == {"FILL", "ADD", "DCA", "OPEN"}, ae.JUDGE_ACTIONS
|
|||
|
|
assert "TRIM" not in ae.JUDGE_ACTIONS, "减持永远不送研判"
|
|||
|
|
assert [a for a, _ in ae.EVALUATORS] == ["TRIM", "ADD", "FILL", "DCA"]
|
|||
|
|
# 持仓那条扫描的输入与产出一个字没改: 空持仓进去, 空候选出来, 不会去碰候选池
|
|||
|
|
r = ae.scan(positions=[], params=params(), market={})
|
|||
|
|
assert r == {"candidates": [], "skipped": []}, r
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ================================================================ 研判请求的硬数字裁剪
|
|||
|
|
@case("研判请求·新建仓只送定性材料, 仓位数字一个不送")
|
|||
|
|
def _():
|
|||
|
|
hard = {"price": 10.0, "score": 300.0, "theme": "储能", "tier": "强传导",
|
|||
|
|
"upside": 0.3, "heat": 0.5, "plan_rank": 1, "plan_bucket": "main",
|
|||
|
|
"plan_src": "plan_api", "sector": "电力设备",
|
|||
|
|
# 下面这些是仓位口径, 决策系统本来就不管仓位
|
|||
|
|
"target_pct": 0.06, "target_amount": 120000.0, "base_amount": 60000.0,
|
|||
|
|
"batch_scheme": "0.5,0.25,0.25", "names_before": 3, "max_names": 20,
|
|||
|
|
"room_amt_before": 1_400_000.0}
|
|||
|
|
got = jd._judge_hard_numbers("OPEN", hard)
|
|||
|
|
assert set(got) == set(jd.OPEN_JUDGE_KEYS), got
|
|||
|
|
for k in ("target_amount", "names_before", "max_names", "room_amt_before",
|
|||
|
|
"batch_scheme", "base_amount", "target_pct"):
|
|||
|
|
assert k not in got, f"{k} 不该送到决策系统"
|
|||
|
|
assert got["score"] == 300.0 and got["price"] == 10.0, got
|
|||
|
|
|
|||
|
|
|
|||
|
|
@case("研判请求·加仓类原样送 (它们的硬数字本身就是判据)")
|
|||
|
|
def _():
|
|||
|
|
hard = {"cushion_pct": -0.085, "stage": 1, "base_qty": 1000}
|
|||
|
|
for act in ("FILL", "ADD", "DCA", "SWITCH"):
|
|||
|
|
assert jd._judge_hard_numbers(act, hard) == hard, act
|
|||
|
|
assert jd._judge_hard_numbers("DCA", None) == {}
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ================================================================ 参考位漂移
|
|||
|
|
@case("参考位漂移·只管新建仓, 别的动作一律不比对")
|
|||
|
|
def _():
|
|||
|
|
with _PatchParams({"PMS_OPEN_REF_DRIFT_MAX": 0.03}):
|
|||
|
|
prog = {"ref_lock": {"ymd": 20260806, "support": 31.27, "pressure": 40.0}}
|
|||
|
|
adv = {"support": 29.00, "pressure": 40.0}
|
|||
|
|
assert ea._check_ref_drift(action="FILL", prog=prog, adv=adv, today=20260806) == ""
|
|||
|
|
assert ea._check_ref_drift(action="ADD", prog=prog, adv=adv, today=20260806) == ""
|
|||
|
|
assert ea._check_ref_drift(action=None, prog=prog, adv=adv, today=20260806) == ""
|
|||
|
|
|
|||
|
|
|
|||
|
|
@case("参考位漂移·当日首答只锁定不比对")
|
|||
|
|
def _():
|
|||
|
|
with _PatchParams({"PMS_OPEN_REF_DRIFT_MAX": 0.03}):
|
|||
|
|
prog = {}
|
|||
|
|
adv = {"support": 31.27, "pressure": 40.0, "buy_band": [30.96, 33.89],
|
|||
|
|
"consulted_at": "09:35"}
|
|||
|
|
assert ea._check_ref_drift(action="OPEN", prog=prog, adv=adv, today=20260806) == ""
|
|||
|
|
assert prog["ref_lock"]["ymd"] == 20260806
|
|||
|
|
assert prog["ref_lock"]["support"] == 31.27
|
|||
|
|
assert prog["ref_lock"]["at"] == "09:35"
|
|||
|
|
|
|||
|
|
|
|||
|
|
@case("参考位漂移·没漂过阈值就放行")
|
|||
|
|
def _():
|
|||
|
|
with _PatchParams({"PMS_OPEN_REF_DRIFT_MAX": 0.03}):
|
|||
|
|
prog = {"ref_lock": {"ymd": 20260806, "support": 31.27, "pressure": 40.0}}
|
|||
|
|
adv = {"support": 31.50, "pressure": 40.0} # 差 0.7%
|
|||
|
|
assert ea._check_ref_drift(action="OPEN", prog=prog, adv=adv, today=20260806) == ""
|
|||
|
|
|
|||
|
|
|
|||
|
|
@case("参考位漂移·漂过阈值就报出来 (002335 实证: 支撑 31.27→29.00)")
|
|||
|
|
def _():
|
|||
|
|
with _PatchParams({"PMS_OPEN_REF_DRIFT_MAX": 0.03}):
|
|||
|
|
prog = {"ref_lock": {"ymd": 20260806, "support": 31.27, "pressure": 40.0,
|
|||
|
|
"at": "09:35"}}
|
|||
|
|
adv = {"support": 29.00, "pressure": 40.0}
|
|||
|
|
why = ea._check_ref_drift(action="OPEN", prog=prog, adv=adv, today=20260806)
|
|||
|
|
assert why, "漂了 7.3% 必须报出来"
|
|||
|
|
assert "31.27" in why and "29.0" in why, why
|
|||
|
|
assert "当日暂停" in why, why
|
|||
|
|
|
|||
|
|
|
|||
|
|
@case("参考位漂移·压力位单独漂也算 (000035 实证: 压力 5.2→5.15 未过阈值)")
|
|||
|
|
def _():
|
|||
|
|
with _PatchParams({"PMS_OPEN_REF_DRIFT_MAX": 0.03}):
|
|||
|
|
prog = {"ref_lock": {"ymd": 20260806, "support": 5.0, "pressure": 5.2}}
|
|||
|
|
assert ea._check_ref_drift(action="OPEN", prog=prog,
|
|||
|
|
adv={"support": 5.0, "pressure": 5.15},
|
|||
|
|
today=20260806) == "" # 差 0.96%, 不到阈值
|
|||
|
|
why = ea._check_ref_drift(action="OPEN", prog=prog,
|
|||
|
|
adv={"support": 5.0, "pressure": 4.5},
|
|||
|
|
today=20260806) # 差 13.5%
|
|||
|
|
assert "压力" in why, why
|
|||
|
|
|
|||
|
|
|
|||
|
|
@case("参考位漂移·锁按日重置, 次日的新结论不算漂移")
|
|||
|
|
def _():
|
|||
|
|
with _PatchParams({"PMS_OPEN_REF_DRIFT_MAX": 0.03}):
|
|||
|
|
prog = {"ref_lock": {"ymd": 20260806, "support": 31.27, "pressure": 40.0}}
|
|||
|
|
adv = {"support": 29.00, "pressure": 40.0, "consulted_at": "09:31"}
|
|||
|
|
assert ea._check_ref_drift(action="OPEN", prog=prog, adv=adv, today=20260807) == ""
|
|||
|
|
assert prog["ref_lock"]["ymd"] == 20260807, "次日首答应该重新锁定"
|
|||
|
|
assert prog["ref_lock"]["support"] == 29.00
|
|||
|
|
|
|||
|
|
|
|||
|
|
@case("参考位漂移·阈值设成 0 等于关掉这道防护")
|
|||
|
|
def _():
|
|||
|
|
with _PatchParams({"PMS_OPEN_REF_DRIFT_MAX": 0.0}):
|
|||
|
|
prog = {"ref_lock": {"ymd": 20260806, "support": 31.27, "pressure": 40.0}}
|
|||
|
|
adv = {"support": 1.0, "pressure": 40.0}
|
|||
|
|
assert ea._check_ref_drift(action="OPEN", prog=prog, adv=adv, today=20260806) == ""
|
|||
|
|
|
|||
|
|
|
|||
|
|
@case("参考位漂移·拿不到支撑压力时不锁也不误报")
|
|||
|
|
def _():
|
|||
|
|
with _PatchParams({"PMS_OPEN_REF_DRIFT_MAX": 0.03}):
|
|||
|
|
prog = {}
|
|||
|
|
assert ea._check_ref_drift(action="OPEN", prog=prog,
|
|||
|
|
adv={"support": None, "pressure": None},
|
|||
|
|
today=20260806) == ""
|
|||
|
|
assert "ref_lock" not in prog, "没有位就不该锁一份空的进去"
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ================================================================ 研判的单轮时间预算
|
|||
|
|
@case("研判预算·没设预算就一律放行")
|
|||
|
|
def _():
|
|||
|
|
from app.services import proposal_service as prs
|
|||
|
|
assert prs._judge_budget_left(None) is True
|
|||
|
|
|
|||
|
|
|
|||
|
|
@case("研判预算·够一次就放行, 不够就留到下一跳")
|
|||
|
|
def _():
|
|||
|
|
import time
|
|||
|
|
from app.services import proposal_service as prs
|
|||
|
|
with _PatchParams({"PMS_JUDGE_TIMEOUT": 90}):
|
|||
|
|
assert prs._judge_budget_left(time.monotonic() + 150) is True # 150 > 90
|
|||
|
|
assert prs._judge_budget_left(time.monotonic() + 30) is False # 30 < 90
|
|||
|
|
assert prs._judge_budget_left(time.monotonic() - 1) is False # 已经超了
|
|||
|
|
|
|||
|
|
|
|||
|
|
# ================================================================ 跑
|
|||
|
|
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()
|