92 lines
4.2 KiB
Python
92 lines
4.2 KiB
Python
"""regime.py 离线单测（无需网络、无需库）：未配置 -> UNKNOWN；两种快照形状都能归一；
|
||
弱势日按旋钮判；追加进临时快照并能读回。
|
||
|
||
跑法：python3 test_regime.py 或 pytest test_regime.py
|
||
"""
|
||
import io
|
||
import json
|
||
import os
|
||
import tempfile
|
||
import urllib.request
|
||
|
||
import config
|
||
import regime
|
||
|
||
|
||
def t(name, cond):
|
||
assert cond, name
|
||
print(" ok", name)
|
||
|
||
|
||
class _Resp(io.BytesIO):
|
||
def __enter__(self):
|
||
return self
|
||
|
||
def __exit__(self, *a):
|
||
return False
|
||
|
||
|
||
def main():
|
||
config.REGIME_API_URL = ""
|
||
r = regime.fetch("2026-09-01")
|
||
t("未配置来源 -> UNKNOWN 且不拦票（weak_day 为空）",
|
||
r["status"] == "UNKNOWN" and r["weak_day"] is None and "unconfigured" in r["source"])
|
||
|
||
config.REGIME_API_URL = "http://127.0.0.1:1/api/v1/market/regime"
|
||
config.REGIME_WEAK_COUNT = 3
|
||
# 决策系统缓存快照的原始形状：indices 是字典、广度叫 breadth_weak
|
||
raw = {"status": "OK", "data_date": "2026-09-01", "computed_at": "x",
|
||
"breadth_weak": 3, "breadth_total": 8,
|
||
"indices": {"000300.SH": {"name": "沪深300", "ok": True, "weak": True, "weak_via": "ma20", "qrs_stale": False},
|
||
"399006.SZ": {"name": "创业板指", "ok": True, "weak": False, "qrs_stale": True},
|
||
"000688.SH": {"name": "科创50", "ok": True, "weak": True},
|
||
"000001.SH": {"name": "上证指数", "ok": True, "weak": True}}}
|
||
orig = urllib.request.urlopen
|
||
urllib.request.urlopen = lambda url, timeout=0: _Resp(json.dumps(raw).encode("utf-8"))
|
||
try:
|
||
r = regime.fetch("2026-09-01")
|
||
t("原始形状归一：OK、弱势数 3、指数转列表、降级按 qrs_stale 判",
|
||
r["status"] == "OK" and r["weak_count"] == 3 and len(r["indices"]) == 4
|
||
and r["degraded"] is True and r["weak_total"] == 8)
|
||
t("弱势日：弱势数 3 达到旋钮 3", r["weak_day"] is True)
|
||
config.REGIME_WEAK_COUNT = 4
|
||
t("旋钮改 4 -> 非弱势日", regime.fetch("2026-09-01")["weak_day"] is False)
|
||
config.REGIME_WEAK_COUNT = 3
|
||
|
||
# 契约形状：indices 为列表、weak_count 直给
|
||
contract = {"status": "OK", "data_date": "2026-09-01", "weak_count": 1,
|
||
"indices": [{"code": "000300.SH", "name": "沪深300", "weak": True}]}
|
||
urllib.request.urlopen = lambda url, timeout=0: _Resp(json.dumps(contract).encode("utf-8"))
|
||
r = regime.fetch("2026-09-01")
|
||
t("契约形状：弱势数 1、非弱势日", r["weak_count"] == 1 and r["weak_day"] is False)
|
||
|
||
urllib.request.urlopen = lambda url, timeout=0: _Resp(json.dumps({"status": "DISABLED"}).encode())
|
||
r = regime.fetch("2026-09-01")
|
||
t("DISABLED 归 UNKNOWN 并注明", r["status"] == "UNKNOWN" and "总开关" in r["source"])
|
||
|
||
def boom(url, timeout=0):
|
||
raise OSError("refused")
|
||
urllib.request.urlopen = boom
|
||
r = regime.fetch("2026-09-01")
|
||
t("来源不可达 -> UNKNOWN，不抛错", r["status"] == "UNKNOWN" and "不可达" in r["source"])
|
||
|
||
# 追加进快照
|
||
urllib.request.urlopen = lambda url, timeout=0: _Resp(json.dumps(raw).encode("utf-8"))
|
||
with tempfile.TemporaryDirectory() as d:
|
||
config.PLAN_SNAPSHOT_DIR = d
|
||
t("快照不存在时不创建空快照", regime.append_to_snapshot("2026-09-01")["status"] == "OK"
|
||
and not os.path.exists(regime.snapshot_path("2026-09-01")))
|
||
with open(regime.snapshot_path("2026-09-01"), "w", encoding="utf-8") as f:
|
||
json.dump({"date": "2026-09-01", "main": []}, f)
|
||
regime.append_to_snapshot("2026-09-01")
|
||
back = regime.read_from_snapshot("2026-09-01")
|
||
t("追加后能从快照读回 regime 段", back and back["status"] == "OK" and back["weak_count"] == 3)
|
||
t("快照其余内容不丢", json.load(open(regime.snapshot_path("2026-09-01"), encoding="utf-8"))["main"] == [])
|
||
finally:
|
||
urllib.request.urlopen = orig
|
||
print("ALL OK — 环境标签：未配置 / 两种形状 / 弱势日旋钮 / 关闭 / 不可达 / 快照追加 全部通过")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|