zlt
/
akg-factor-bridge
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

缓存自检的替身挂到「算一份计划」的统一入口，不再挂在现算那条路上 

接口 2026-09-10 改成优先读快照后，挂在 plan.collect 上的替身再也不会被调到，
测试会假通过。缓存与单飞管的是「同一个键算几次」，与底下走哪条路无关。

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
zlt 2026-09-10 09:25:50 +08:00
parent 15a02d520b
commit 8770fbb683
1 changed files with 18 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
test_plan_cache.py
View File

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