diff --git a/test_plan_cache.py b/test_plan_cache.py index 66c1856..133a575 100644 --- a/test_plan_cache.py +++ b/test_plan_cache.py @@ -31,18 +31,23 @@ def t(name, cond, extra=""): class _Spy: - """替身:记录 collect 被调了几次、每次算多久。""" + """替身:记录「算一份计划」被调了几次、每次算多久。 + + 挂在 api._plan_build 上,不挂在 plan.collect 上:缓存与单飞管的是「同一个键算几次」, + 底下走快照还是现场装配与它无关。2026-09-10 接口改成优先读快照之后,挂在 collect 上的 + 替身就再也不会被调到,测试会假通过。 + """ def __init__(self, delay=0.0): self.calls = [] self.delay = delay self._lock = threading.Lock() - def collect(self, date, top, obs_top, theme_cap): + def build(self, date, top, obs_top, theme_cap): with self._lock: self.calls.append((date, top, obs_top, theme_cap)) time.sleep(self.delay) - return {"date": date or "2026-09-09", "main": [], "observe": [], "_full": "该被丢掉"} + return {"date": date or "2026-09-09", "main": [], "observe": [], "plan_source": "替身"} def _reset(): @@ -54,15 +59,15 @@ def test_hit_and_miss(): print("[命中与未命中]") spy = _Spy() _reset() - old_collect, old_latest = plan.collect, plan._latest_date - plan.collect = spy.collect + old_build, old_latest = api._plan_build, plan._latest_date + api._plan_build = spy.build plan._latest_date = lambda tbl: "2026-09-09" try: d1, hit1 = api._plan_cached(None, 300, 100, 2, False) d2, hit2 = api._plan_cached(None, 300, 100, 2, False) t("第一次未命中、第二次命中", hit1 is False and hit2 is True) t("只算了一次", len(spy.calls) == 1, spy.calls) - t("内部大字段被丢掉,不进缓存也不出接口", "_full" not in d1 and "_full" not in d2) + t("拿到的就是统一入口给的那一份,缓存不改内容", d1 is d2 and d1["date"] == "2026-09-09") # 参数不同 = 不同的键。PMS 用 300/100,人工联调常用 30/20,两份互不干扰 api._plan_cached(None, 30, 20, 2, False) @@ -80,15 +85,15 @@ def test_hit_and_miss(): _d, hit5 = api._plan_cached(None, 300, 100, 2, False) t("强刷之后缓存是新的", hit5 is True and len(spy.calls) == n + 1) finally: - plan.collect, plan._latest_date = old_collect, old_latest + api._plan_build, plan._latest_date = old_build, old_latest def test_single_flight(): print("[同键只算一次 —— 09-10 超时的那一半]") spy = _Spy(delay=0.4) # 装配很慢,真实环境是三十多秒 _reset() - old_collect, old_latest = plan.collect, plan._latest_date - plan.collect = spy.collect + old_build, old_latest = api._plan_build, plan._latest_date + api._plan_build = spy.build plan._latest_date = lambda tbl: "2026-09-09" try: results, errs = [], [] @@ -114,15 +119,15 @@ def test_single_flight(): t("只有一路是未命中,其余都吃了缓存", sum(1 for _d, h in results if not h) == 1, [h for _d, h in results]) finally: - plan.collect, plan._latest_date = old_collect, old_latest + api._plan_build, plan._latest_date = old_build, old_latest def test_different_keys_not_blocked(): print("[不同参数互不阻塞]") spy = _Spy(delay=0.4) _reset() - old_collect, old_latest = plan.collect, plan._latest_date - plan.collect = spy.collect + old_build, old_latest = api._plan_build, plan._latest_date + api._plan_build = spy.build plan._latest_date = lambda tbl: "2026-09-09" try: def one(top): @@ -137,7 +142,7 @@ def test_different_keys_not_blocked(): t("三种参数各算一次", len(spy.calls) == 3, spy.calls) t("是并行不是排队", el < 0.4 * 2.5, "%.2f 秒" % el) finally: - plan.collect, plan._latest_date = old_collect, old_latest + api._plan_build, plan._latest_date = old_build, old_latest def test_ttl_covers_trading_day():