tradingSystem/scripts/test_batch7_units.py

# -*- coding: utf-8 -*-
"""
第七批模块单测 (零外部依赖, 不连库不触网)
==========================================
运行: 在 tradingSystem 仓库根目录执行  python scripts/test_batch7_units.py
覆盖: 上游选股计划接口 (plan_feed) 的解析、新鲜度校验、候选筛选三段纯逻辑。

fixture 用的是 2026-07-30 上游给的真实应答样例 (截取前若干条, 结构一字未改),
坏数据用例另造 —— 「上游改了字段/漏了字段」这类事故必须由单测先炸, 而不是盘中静默少票。
"""
import os
import sys
import traceback

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from app.services import plan_feed as pf                                       # noqa: E402

RESULTS = []


def case(name):
    def deco(fn):
        RESULTS.append((name, fn))
        return fn
    return deco


def raises(exc, fn, *a, **kw):
    try:
        fn(*a, **kw)
    except exc:
        return True
    except Exception as e:
        raise AssertionError(f"期望 {exc.__name__}, 实际 {type(e).__name__}: {e}")
    raise AssertionError(f"期望抛 {exc.__name__}, 但没抛")


# ================================================================ fixture
def _m(rank, code, name, score, theme, heat, upside, tier="强传导", n_sources=7, moved=0.0):
    return {"rank": rank, "code": code, "name": name, "score": score,
            "evidence": {"theme": theme, "n_sources": n_sources, "moved_ratio": moved},
            "heat": heat, "upside": upside, "tier": tier}


def _o(rank, code, name, score, theme, heat, n_sources=9, moved=0.075):
    return {"rank": rank, "code": code, "name": name, "score": score,
            "evidence": {"theme": theme, "n_sources": n_sources, "moved_ratio": moved},
            "heat": heat, "upside": None}


SAMPLE = {
    "date": "2026-07-29",
    "counts": {"main": 961, "observe": 107, "gate_covered": 2386},
    "market_snapshot_days": ["2026-07-28"],
    "heat_date": "2026-07-27",
    "theme_cap": 5,
    "main": [
        _m(1, "SH600418", "江淮汽车", 242.24, "整车", 0.2249, 2.1203202525935945),
        _m(2, "SH688717", "艾罗能源", 242.05, "储能", 0.4165, 1.5817543859649121),
        _m(3, "SZ300952", "恒辉安防", 241.91, "传感器", 0.3391, 1.4061712846347607),
        _m(4, "SH605598", "上海港湾", 241.72, "储能", 0.214, 1.292594091460947),
        _m(6, "SH600875", "东方电气", 241.34, "整机", 0.4074, 1.2340787236824915,
           n_sources=9, moved=0.075),
        _m(10, "SH600104", "上汽集团", 241.18, "整车", 0.1239, 0.8065614360879017),
    ],
    "observe": [
        _o(1, "SH600877", "电科芯片", 101.18, "集成电路设计", 0.4743, moved=0.0),
        _o(2, "SH603611", "诺力股份", 101.12, "整机", 0.5868),
        _o(3, "SZ000768", "中航西飞", 101.06, "整机", 0.121),
    ],
    "changes": None,
    "encoding": "主榜分=200+传导档位×20+组内分（还没热、还便宜）；观察档分=100+0.6z(传导)+0.4z(−热度)",
}


# ================================================================ 解析
@case("解析·真实样例: 计数/主题上限/热度日全部落位, 返回条数与 counts 分开记")
def _():
    p = pf.parse_plan(SAMPLE)
    assert p["date"] == "2026-07-29" and p["heat_date"] == "2026-07-27"
    assert p["theme_cap"] == 5
    assert p["counts"] == {"main": 961, "observe": 107, "gate_covered": 2386}
    # counts 是上游全量, returned 是这次应答里真的给了几条 —— 混用会看不出上游截断了
    assert p["returned"] == {"main": 6, "observe": 3}
    assert p["market_snapshot_days"] == ["2026-07-28"]
    assert p["encoding"].startswith("主榜分=200")


@case("解析·代码从前缀式归一为点式 (沪/深/科创/创业板四种都要对)")
def _():
    p = pf.parse_plan(SAMPLE)
    codes = [r["ts_code"] for r in p["main"]]
    assert codes[:4] == ["600418.SH", "688717.SH", "300952.SZ", "605598.SH"], codes
    assert [r["ts_code"] for r in p["observe"]][2] == "000768.SZ"


@case("解析·字段口径: score/heat/upside 转 float, evidence 拆平, bucket 打标")
def _():
    p = pf.parse_plan(SAMPLE)
    r = p["main"][0]
    assert r["name"] == "江淮汽车" and r["rank"] == 1 and r["tier"] == "强传导"
    assert abs(r["score"] - 242.24) < 1e-9 and abs(r["heat"] - 0.2249) < 1e-9
    assert abs(r["upside"] - 2.1203202525935945) < 1e-12
    assert r["theme"] == "整车" and r["n_sources"] == 7 and r["moved_ratio"] == 0.0
    assert r["bucket"] == pf.BUCKET_MAIN and p["observe"][0]["bucket"] == pf.BUCKET_OBSERVE
    # 观察档没有 tier、upside 是 null —— 都得是 None 而不是 0
    assert p["observe"][0]["tier"] is None and p["observe"][0]["upside"] is None


@case("解析·themes 映射: 主榜优先于观察档, 同码不被观察档覆盖")
def _():
    d = dict(SAMPLE)
    d["observe"] = list(SAMPLE["observe"]) + [_o(9, "SH600418", "江淮汽车", 100.1, "汽车零部件", 0.2)]
    p = pf.parse_plan(d)
    assert p["themes"]["600418.SH"] == "整车", p["themes"]["600418.SH"]
    assert p["themes"]["600877.SH"] == "集成电路设计"


@case("解析·变形应答一律抛 PlanFeedError (不是 JSON 对象 / 缺 date / 两档全空)")
def _():
    raises(pf.PlanFeedError, pf.parse_plan, ["不是对象"])
    raises(pf.PlanFeedError, pf.parse_plan, None)
    raises(pf.PlanFeedError, pf.parse_plan, {"main": SAMPLE["main"]})          # 缺 date
    raises(pf.PlanFeedError, pf.parse_plan, {"date": "2026-07-29", "main": [], "observe": []})
    raises(pf.PlanFeedError, pf.parse_plan, {"date": " ", "main": SAMPLE["main"]})


@case("解析·脏行跳过而不整体失效: 非 dict / 缺 code / 同码重复 / evidence 非 dict")
def _():
    d = {"date": "2026-07-29", "main": [
        "我是脏行", None, 42,
        {"rank": 1, "code": "SH600418", "score": 1.0, "evidence": "不是对象"},
        {"rank": 2, "code": "", "score": 2.0},
        {"rank": 3, "code": "600418", "score": 3.0},          # 与第一条同码 (归一后相同)
        {"code": "SZ300952", "score": 4.0},                    # 缺 rank
    ]}
    p = pf.parse_plan(d)
    codes = [r["ts_code"] for r in p["main"]]
    assert codes == ["600418.SH", "300952.SZ"], codes
    assert p["main"][0]["theme"] is None and p["main"][0]["n_sources"] is None
    # rank 缺失用出现序号补 (第 7 个元素 → 7), 免得排序键出现 None
    assert p["main"][1]["rank"] == 7, p["main"][1]["rank"]
    assert p["counts"] == {"main": None, "observe": None, "gate_covered": None}


@case("解析·数值容错: 字符串数字可用, 空串/None/非数字一律 None 而不是 0")
def _():
    d = {"date": "2026-07-29", "main": [
        {"rank": "5", "code": "SH600418", "score": "242.24", "heat": "", "upside": None,
         "evidence": {"theme": "  整车  ", "n_sources": "7", "moved_ratio": "abc"}}]}
    r = pf.parse_plan(d)["main"][0]
    assert r["rank"] == 5 and abs(r["score"] - 242.24) < 1e-9
    assert r["heat"] is None and r["upside"] is None and r["moved_ratio"] is None
    assert r["n_sources"] == 7 and r["theme"] == "整车"          # theme 两头空白要去掉


@case("解析·漏斗: counts 是打分池, returned 是过筛后 —— 两者相减没有意义, 只做展示")
def _():
    p = pf.parse_plan(SAMPLE)
    assert p["funnel"] == {"scored_main": 961, "returned_main": 6,
                           "scored_observe": 107, "returned_observe": 3}, p["funnel"]


@case("解析·吃满才叫截断: 拿 counts 判会永远报警 (961 vs 55 是两把尺子)")
def _():
    # 上游签名默认 top=20 / obs_top=10。样例只有 6+3 条 —— 没吃满, 就是上游能给的全部,
    # 尽管 counts 写着 961。**这正是不能拿 counts 判截断的原因。**
    p = pf.parse_plan(SAMPLE)
    assert p["requested"] == {"top": 20, "obs_top": 10, "theme_cap": None}, p["requested"]
    assert p["truncated"] == {"main": False, "observe": False}, p["truncated"]
    # 要了 6 条正好给 6 条 → 吃满, 可能还有更多
    q = pf.parse_plan(SAMPLE, requested={"top": 6, "obs_top": 3, "theme_cap": 999})
    assert q["truncated"] == {"main": True, "observe": True}, q["truncated"]
    assert q["requested"]["theme_cap"] == 999
    # 要 100 条只给 6 条 → 没吃满
    r = pf.parse_plan(SAMPLE, requested={"top": 100, "obs_top": 100})
    assert r["truncated"] == {"main": False, "observe": False}
    # top=0 (不传) 时无从判断, 不许瞎报
    z = pf.parse_plan(SAMPLE, requested={"top": 0, "obs_top": 0})
    assert z["truncated"] == {"main": False, "observe": False}


@case("取数·附加查询串: 解析 / 合并进请求 / date 不可被覆盖 / 坏串忽略不炸")
def _():
    assert pf.parse_query_extra("limit=1000") == {"limit": "1000"}
    assert pf.parse_query_extra("?limit=1000&offset=20") == {"limit": "1000", "offset": "20"}
    assert pf.parse_query_extra("") == {} and pf.parse_query_extra(None) == {}
    assert pf.parse_query_extra("   ") == {}
    assert pf.parse_query_extra("这不是查询串") == {}      # 解析不出 k=v → 空, 不抛
    calls, restore = _with_fake_requests(lambda *a: _FakeResp(SAMPLE))
    try:
        pf.fetch(base="http://x:8300", path="/plan", timeout=3,
                 extra_params={"limit": "1000", "date": "被覆盖的"}, date="2026-07-29")
    finally:
        restore()
    # date 是 PMS 自己的语义, extra 里同名键不许赢
    assert calls[0]["params"] == {"limit": "1000", "date": "2026-07-29"}, calls[0]["params"]


# ================================================================ 新鲜度
@case("新鲜度·日龄: 当天 0 / 上一交易日 1 / 未来日期按 0 (为下一交易日出的计划)")
def _():
    assert pf.plan_age_tdays("2026-07-30", today="2026-07-30") == 0
    assert pf.plan_age_tdays("2026-07-29", today="2026-07-30") == 1
    assert pf.plan_age_tdays("2026-07-27", today="2026-07-30") == 3
    assert pf.plan_age_tdays("2026-07-31", today="2026-07-30") == 0


@case("新鲜度·跨周末只算交易日: 周五的计划到周一仍是 1 个交易日龄")
def _():
    # 2026-07-24 周五, 2026-07-27 周一 —— 中间两天不是交易日, 不该被算进日龄
    assert pf.plan_age_tdays("2026-07-24", today="2026-07-27") == 1
    assert pf.plan_age_tdays("2026-07-24", today="2026-07-28") == 2


@case("新鲜度·超期抛错并说清日龄 (上游停更时拿旧榜当今天比没候选更危险)")
def _():
    p = pf.parse_plan(SAMPLE)          # date = 2026-07-29
    assert pf.assert_fresh(p, max_stale_tdays=1, today="2026-07-30") == 1
    assert pf.assert_fresh(p, max_stale_tdays=0, today="2026-07-29") == 0
    raises(pf.PlanFeedError, pf.assert_fresh, p, max_stale_tdays=0, today="2026-07-30")
    try:
        pf.assert_fresh(p, max_stale_tdays=1, today="2026-08-05")
    except pf.PlanFeedError as e:
        assert "2026-07-29" in str(e) and "交易日" in str(e), str(e)


# ================================================================ 筛选
@case("筛选·按 score 降序取前 N, 同分按上游 rank 稳定次序")
def _():
    p = pf.parse_plan(SAMPLE)
    r = pf.select_candidates(p, top_n=3)
    assert [x["ts_code"] for x in r["items"]] == ["600418.SH", "688717.SH", "300952.SZ"]
    assert r["considered"] == 6 and r["eligible"] == 6 and r["dropped"]["capped"] == 3
    # 同分: rank 小的在前
    d = {"date": "2026-07-29", "main": [_m(9, "SH600001", "甲", 100.0, "整车", 0.1, 1.0),
                                        _m(2, "SH600002", "乙", 100.0, "整车", 0.1, 1.0)]}
    rr = pf.select_candidates(pf.parse_plan(d), top_n=2)
    assert [x["ts_code"] for x in rr["items"]] == ["600002.SH", "600001.SH"]


@case("筛选·top_n=0 视为不截断 (别把'不限'写成'一只都不要')")
def _():
    p = pf.parse_plan(SAMPLE)
    assert len(pf.select_candidates(p, top_n=0)["items"]) == 6
    assert len(pf.select_candidates(p, top_n=999)["items"]) == 6


@case("筛选·已持有与黑名单剔除, 且两者分开计数 (输入接受前缀式)")
def _():
    p = pf.parse_plan(SAMPLE)
    r = pf.select_candidates(p, held=["SH600418", "688717.SH"], black=["SZ300952"], top_n=10)
    codes = [x["ts_code"] for x in r["items"]]
    assert "600418.SH" not in codes and "688717.SH" not in codes and "300952.SZ" not in codes
    assert r["dropped"]["held"] == 2 and r["dropped"]["black"] == 1
    assert r["eligible"] == 3


@case("筛选·tier 白名单只对带 tier 的行生效; 观察档的闸门是 include_observe")
def _():
    d = dict(SAMPLE)
    d["main"] = list(SAMPLE["main"]) + [
        _m(500, "SH600519", "弱票", 220.0, "白酒", 0.1, 0.5, tier="弱传导")]
    p = pf.parse_plan(d)
    r = pf.select_candidates(p, tiers=["强传导"], top_n=50)
    assert "600519.SH" not in [x["ts_code"] for x in r["items"]]
    assert r["dropped"]["tier"] == 1
    # 观察档没有 tier —— tiers 非空时也不该被 tier 规则误杀, 但默认根本不进池
    assert all(x["bucket"] == "main" for x in r["items"])
    r2 = pf.select_candidates(p, tiers=["强传导"], include_observe=True, top_n=50)
    obs = [x["ts_code"] for x in r2["items"] if x["bucket"] == "observe"]
    assert obs == ["600877.SH", "603611.SH", "000768.SZ"], obs
    assert r2["dropped"]["tier"] == 1


@case("筛选·跨档同码只留一次, 且留主榜那份 (主榜分高, 排序天然优先)")
def _():
    d = dict(SAMPLE)
    d["observe"] = list(SAMPLE["observe"]) + [_o(9, "SH600418", "江淮汽车", 100.1, "汽车零部件", 0.2)]
    r = pf.select_candidates(pf.parse_plan(d), include_observe=True, top_n=50)
    hit = [x for x in r["items"] if x["ts_code"] == "600418.SH"]
    assert len(hit) == 1 and hit[0]["bucket"] == "main" and hit[0]["theme"] == "整车"
    assert r["dropped"]["dup"] == 1


@case("筛选·score 下限与 n_sources 下限各自独立计数")
def _():
    p = pf.parse_plan(SAMPLE)
    r = pf.select_candidates(p, min_score=241.7, top_n=50)
    assert [x["ts_code"] for x in r["items"]] == ["600418.SH", "688717.SH", "300952.SZ",
                                                 "605598.SH"]
    assert r["dropped"]["score"] == 2
    r2 = pf.select_candidates(p, min_sources=9, top_n=50)
    assert [x["ts_code"] for x in r2["items"]] == ["600875.SH"], r2["items"]
    assert r2["dropped"]["sources"] == 5


@case("筛选·upside 下限 (2.12=+212%): 对所有行生效, 观察档因 upside=null 被一并挡掉")
def _():
    d = dict(SAMPLE)
    d["observe"] = list(SAMPLE["observe"])
    p = pf.parse_plan(d)
    r = pf.select_candidates(p, min_upside=1.3, top_n=50)
    assert [x["ts_code"] for x in r["items"]] == ["600418.SH", "688717.SH", "300952.SZ"], r["items"]
    assert r["dropped"]["upside"] == 3          # 1.2926 / 1.2341 / 0.8066 三只在门槛下
    # 观察档 upside 恒为 null → 按 0 算, 设了下限就全挡掉 (方向保守: 宁可少票)
    r2 = pf.select_candidates(p, min_upside=1.3, include_observe=True, top_n=50)
    assert all(x["bucket"] == "main" for x in r2["items"]), r2["items"]
    assert r2["dropped"]["upside"] == 6
    # 不设下限时观察档照旧能进 (闸门只有 include_observe)
    r3 = pf.select_candidates(p, include_observe=True, top_n=50)
    assert any(x["bucket"] == "observe" for x in r3["items"])
    assert r3["dropped"]["upside"] == 0


@case("筛选·upside 永不参与排序 (券商目标价噪音大, 排序只认 score)")
def _():
    d = {"date": "2026-07-29", "main": [
        _m(1, "SH600001", "高分低空间", 242.0, "整车", 0.1, 0.30),
        _m(2, "SH600002", "低分高空间", 200.0, "整车", 0.1, 5.00)]}
    r = pf.select_candidates(pf.parse_plan(d), top_n=2)
    assert [x["ts_code"] for x in r["items"]] == ["600001.SH", "600002.SH"], r["items"]


@case("筛选·ST 剔除 (上游明确不剔, 这道闸归 PMS): ST/*ST/退市 都要挡, 名字缺失不误杀")
def _():
    assert pf.is_st("ST喜临门") and pf.is_st("*ST辉丰") and pf.is_st("S*ST前锋")
    assert pf.is_st("SST华塑") and pf.is_st("*st 辉丰")            # 大小写与空格都要吃掉
    assert pf.is_st("退市海润") and pf.is_st("海润退")              # 退市整理期两种写法
    assert not pf.is_st("江淮汽车") and not pf.is_st("") and not pf.is_st(None)
    # 「圣农发展」「深赛格」这类 S 开头的中文名不许误杀 (A 股名称都是中文, 拉丁 ST 前缀安全)
    assert not pf.is_st("圣农发展") and not pf.is_st("深赛格") and not pf.is_st("三安光电")
    d = {"date": "2026-07-29", "main": [
        _m(1, "SH600001", "ST喜临门", 240.0, "整机", 0.1, 1.0),
        _m(2, "SH600002", "*ST辉丰", 239.0, "整机", 0.1, 1.0),
        _m(3, "SH600003", "江淮汽车", 238.0, "整车", 0.1, 1.0),
        {"rank": 4, "code": "SH600004", "score": 237.0, "tier": "强传导",
         "evidence": {"theme": "整车"}}]}                      # 没有 name
    p = pf.parse_plan(d)
    r = pf.select_candidates(p, top_n=10, exclude_st=True)
    assert [x["ts_code"] for x in r["items"]] == ["600003.SH", "600004.SH"], r["items"]
    assert r["dropped"]["st"] == 2
    assert r["st_unknown"] == ["600004.SH"], r["st_unknown"]   # 判不了要报出来, 不静默
    assert len(pf.select_candidates(p, top_n=10, exclude_st=False)["items"]) == 4


@case("筛选·白名单档位取全判据: 看见白名单之外的档位才算齐 (吃满 top 不等于漏了)")
def _():
    strong = [_m(i, f"SH60{i:04d}", f"强{i}", 240 - i, "整车", 0.1, 1.0) for i in range(1, 4)]
    weak = [_m(9, "SH609999", "弱", 220.0, "整车", 0.1, 1.0, tier="弱传导")]
    # 只有强传导 → 判不出后面还有没有, tier_complete=False (被 top 卡住了)
    only_strong = pf.select_candidates(pf.parse_plan({"date": "2026-07-29", "main": strong}),
                                       tiers=["强传导"], top_n=10)
    assert only_strong["tier_complete"] is False
    # 出现弱传导 → 强传导已经全在前面了
    both = pf.select_candidates(pf.parse_plan({"date": "2026-07-29", "main": strong + weak}),
                                tiers=["强传导"], top_n=10)
    assert both["tier_complete"] is True
    # 没设白名单时无从判断
    assert pf.select_candidates(pf.parse_plan({"date": "2026-07-29", "main": strong}),
                                top_n=10)["tier_complete"] is None


@case("筛选·候选级主题限额在 top_n 截断之前生效 (否则前 N 名被单一主题垄断)")
def _():
    # 造一个"宽池子": 储能 6 只分最高, 传感器 3 只, 整车 2 只
    rows = ([_m(i, f"SH60{i:04d}", f"储{i}", 300 - i, "储能", 0.2, 1.0) for i in range(1, 7)]
            + [_m(10 + i, f"SH61{i:04d}", f"传{i}", 200 - i, "传感器", 0.2, 1.0) for i in range(1, 4)]
            + [_m(20 + i, f"SH62{i:04d}", f"整{i}", 100 - i, "整车", 0.2, 1.0) for i in range(1, 3)])
    p = pf.parse_plan({"date": "2026-07-29", "main": rows})
    # 不限主题: 前 5 名全是储能 —— 规则闸按 PMS_SECTOR_MAX_NAMES 一拦就废掉大半
    plain = pf.select_candidates(p, top_n=5)
    assert {x["theme"] for x in plain["items"]} == {"储能"}, plain["items"]
    # 限 2 只/主题: 先摊开再截断, 5 个名额分到三个主题
    capped = pf.select_candidates(p, top_n=5, theme_cap=2)
    got = [(x["theme"], x["ts_code"]) for x in capped["items"]]
    assert [t for t, _ in got] == ["储能", "储能", "传感器", "传感器", "整车"], got
    assert capped["dropped"]["theme"] == 5      # 储能多 4 只 + 传感器多 1 只
    # 主题为空的行归到「(无主题)」一档, 不跟着别的主题挤
    p2 = pf.parse_plan({"date": "2026-07-29", "main": [
        {"rank": 1, "code": "SH600001", "score": 9}, {"rank": 2, "code": "SH600002", "score": 8}]})
    assert len(pf.select_candidates(p2, top_n=5, theme_cap=1)["items"]) == 1


@case("取数·build_query: 显式 top/obs_top/theme_cap 覆盖 QUERY_EXTRA 同名键; 0=不传")
def _():
    assert pf.build_query({"top": 300, "obs_top": 100, "theme_cap": 999, "query_extra": {}}) == \
        {"top": "300", "obs_top": "100", "theme_cap": "999"}
    # 0 = 不传该参数 (用上游默认)
    assert pf.build_query({"top": 0, "obs_top": 0, "theme_cap": 0, "query_extra": {}}) == {}
    # QUERY_EXTRA 里的同名键输给显式参数; 非同名键照旧透传
    q = pf.build_query({"top": 300, "obs_top": 0, "theme_cap": 0,
                        "query_extra": {"top": "20", "foo": "bar"}})
    assert q == {"top": "300", "foo": "bar"}, q


@case("筛选·输出字段: sector 用 theme 灌 (planner 吃这个), score 缺失兜 0.0 不留 None")
def _():
    d = {"date": "2026-07-29", "main": [{"rank": 1, "code": "SH600418",
                                         "evidence": {"theme": "整车"}}]}
    x = pf.select_candidates(pf.parse_plan(d), top_n=5)["items"][0]
    assert x["score"] == 0.0 and x["sector"] == "整车" and x["theme"] == "整车"
    assert x["src"] == "plan_api" and x["ts_code"] == "600418.SH"
    assert "price" not in x, "计划不带价格, 这里绝不能凭空造一个价出来"


@case("筛选·计数自洽: considered = eligible + 各项丢弃; items = eligible − capped")
def _():
    d = dict(SAMPLE)
    d["main"] = list(SAMPLE["main"]) + [
        _m(500, "SH600519", "弱票", 220.0, "白酒", 0.1, 0.5, tier="弱传导")]
    p = pf.parse_plan(d)
    r = pf.select_candidates(p, held=["SH600418"], black=["SZ300952"], tiers=["强传导"],
                             min_sources=7, top_n=2, include_observe=True, theme_cap=2)
    dr = r["dropped"]
    assert r["considered"] == 10, r["considered"]
    assert r["considered"] == r["eligible"] + dr["held"] + dr["black"] + dr["tier"] \
        + dr["score"] + dr["sources"] + dr["upside"] + dr["theme"] + dr["dup"], (r, dr)
    assert len(r["items"]) == r["eligible"] - dr["capped"] == 2


@case("筛选·空计划不炸 (main/observe 缺键) 且 date 透传")
def _():
    r = pf.select_candidates({"date": "2026-07-29"}, top_n=5)
    assert r["items"] == [] and r["considered"] == 0 and r["eligible"] == 0
    assert r["date"] == "2026-07-29"


# ================================================================ 取数守卫
class _FakeResp:
    def __init__(self, payload, status=200):
        self._p, self.status_code = payload, status

    def raise_for_status(self):
        if self.status_code >= 400:
            raise RuntimeError(f"HTTP {self.status_code}")

    def json(self):
        if isinstance(self._p, Exception):
            raise self._p
        return self._p


def _with_fake_requests(handler):
    """把 requests 换成假模块 (装在 sys.modules 上, plan_feed 是函数内 import)。"""
    import types
    calls = []
    fake = types.ModuleType("requests")

    def _get(url, params=None, timeout=None):
        calls.append({"url": url, "params": params, "timeout": timeout})
        return handler(url, params, timeout)
    fake.get = _get
    prev = sys.modules.get("requests")
    sys.modules["requests"] = fake

    def restore():
        if prev is None:
            sys.modules.pop("requests", None)
        else:
            sys.modules["requests"] = prev
    return calls, restore


@case("取数·base 为空立即抛 PlanFeedError, 一个 HTTP 请求都不发")
def _():
    calls, restore = _with_fake_requests(
        lambda *a: (_ for _ in ()).throw(AssertionError("不该发请求")))
    try:
        raises(pf.PlanFeedError, pf.fetch, base="", path="/plan", timeout=1)
        raises(pf.PlanFeedError, pf.fetch, base="   ", path="/plan", timeout=1)
    finally:
        restore()
    assert calls == [], calls


@case("取数·成功路径: URL 拼接/date 传参/超时透传, 并盖上 url 与 fetched_at")
def _():
    calls, restore = _with_fake_requests(lambda *a: _FakeResp(SAMPLE))
    try:
        p = pf.fetch(base="http://192.168.16.155:8300/", path="plan", timeout=7,
                     date="2026-07-29")
    finally:
        restore()
    assert calls[0]["url"] == "http://192.168.16.155:8300/plan", calls
    assert calls[0]["params"] == {"date": "2026-07-29"} and calls[0]["timeout"] == 7
    assert p["date"] == "2026-07-29" and p["requested_date"] == "2026-07-29"
    assert p["url"].endswith("/plan") and p["fetched_at"] > 0
    # 不传 date 时不能带一个 date=None 的空参上去
    calls2, restore2 = _with_fake_requests(lambda *a: _FakeResp(SAMPLE))
    try:
        pf.fetch(base="http://x:8300", path="/plan", timeout=3)
    finally:
        restore2()
    assert calls2[0]["params"] is None, calls2


@case("取数·HTTP 报错/JSON 坏掉一律包装成 PlanFeedError 并带上 URL")
def _():
    for handler in (lambda *a: _FakeResp(SAMPLE, status=500),
                    lambda *a: _FakeResp(ValueError("Expecting value")),
                    lambda *a: (_ for _ in ()).throw(OSError("Connection refused"))):
        calls, restore = _with_fake_requests(handler)
        try:
            try:
                pf.fetch(base="http://192.168.16.155:8300", path="/plan", timeout=1)
                raise AssertionError("应当抛 PlanFeedError")
            except pf.PlanFeedError as e:
                assert "192.168.16.155:8300/plan" in str(e), str(e)
        finally:
            restore()


@case("取数·来源常量与旧表来源都在: plan_api / buy_plan / both")
def _():
    assert (pf.SRC_PLAN_API, pf.SRC_BUY_PLAN, pf.SRC_BOTH) == ("plan_api", "buy_plan", "both")
    assert pf.FAIL_CACHE_SEC > 0, "失败也要缓存一会儿, 否则每分钟的调度位会把超时叠成雪崩"


def main():
    import logging
    logging.disable(logging.CRITICAL)
    passed, failed = 0, 0
    for name, fn in RESULTS:
        try:
            fn()
            print(f"  PASS  {name}")
            passed += 1
        except Exception:
            print(f"  FAIL  {name}")
            traceback.print_exc()
            failed += 1
    print("-" * 60)
    if failed:
        print(f"FAILED: {failed} / {passed + failed}")
        sys.exit(1)
    print(f"ALL PASS ({passed} cases)")


if __name__ == "__main__":
    main()