293 lines
12 KiB
Python
293 lines
12 KiB
Python
# tests/factor/test_data_gap_check.py
|
|
"""数据缺口比对 TDD——决议 J/§11.3-②: 需求(注册表 params.source) vs 供给(manifest)."""
|
|
from sanguo_factor.data_gap_check import (check_gaps, collect_requirements,
|
|
load_manifest, main)
|
|
from sanguo_factor.version_registry import add_version, save_registry, \
|
|
upsert_factor
|
|
|
|
|
|
def _reg(tmp_path):
|
|
reg = {"factors": {}}
|
|
upsert_factor(reg, "pledge_net_chg", hypothesis="H-1")
|
|
add_version(reg, "pledge_net_chg", commit="c1",
|
|
params={"source": "pledge_detail"}, effective_from="2026-09-22")
|
|
upsert_factor(reg, "future_alpha", hypothesis="H-2")
|
|
add_version(reg, "future_alpha", commit="c2",
|
|
params={"source": "lhb_seat"}, # manifest 没有的域
|
|
effective_from="2026-09-22")
|
|
p = str(tmp_path / "reg.yaml")
|
|
save_registry(p, reg)
|
|
return reg, p
|
|
|
|
|
|
def test_collect_requirements_latest_version_source(tmp_path):
|
|
reg, _ = _reg(tmp_path)
|
|
reqs = collect_requirements(reg)
|
|
assert reqs["pledge_net_chg"] == ["pledge_detail"]
|
|
assert reqs["future_alpha"] == ["lhb_seat"]
|
|
|
|
|
|
def test_check_gaps_flags_missing_domain(tmp_path):
|
|
reg, _ = _reg(tmp_path)
|
|
manifest = {"domains": {"pledge_detail": {"desc": "质押明细"},
|
|
"bars_daily": {"desc": "日线"}}}
|
|
gaps = check_gaps(collect_requirements(reg), manifest)
|
|
assert gaps == {"future_alpha": ["lhb_seat"]}
|
|
|
|
|
|
def test_main_exit_codes(tmp_path):
|
|
_, reg_p = _reg(tmp_path)
|
|
mf = tmp_path / "manifest.yaml"
|
|
mf.write_text("domains:\n pledge_detail: {desc: 质押明细}\n", encoding="utf-8")
|
|
out = str(tmp_path / "out")
|
|
rc = main(["--registry", reg_p, "--manifest", str(mf), "--out-dir", out])
|
|
assert rc == 2 # future_alpha 缺 lhb_seat
|
|
import json
|
|
import os
|
|
with open(os.path.join(out, "data_gaps.json")) as f:
|
|
assert json.load(f) == {"future_alpha": ["lhb_seat"]}
|
|
mf.write_text("domains:\n pledge_detail: {desc: x}\n lhb_seat: {desc: y}\n",
|
|
encoding="utf-8")
|
|
assert main(["--registry", reg_p, "--manifest", str(mf), "--out-dir", out]) == 0
|
|
|
|
|
|
# —— 决议 M 半环: 指纹去重自动开单(网络全 mock) ——
|
|
def test_gap_fingerprint_stable_and_sensitive():
|
|
from sanguo_factor.data_gap_check import gap_fingerprint
|
|
a = {"f2": ["s_x"], "f1": ["s_a", "s_b"]}
|
|
assert gap_fingerprint(a) == gap_fingerprint(
|
|
{"f1": ["s_b", "s_a"], "f2": ["s_x"]}) # 顺序无关
|
|
assert gap_fingerprint(a) != gap_fingerprint({"f1": ["s_a"]})
|
|
assert len(gap_fingerprint(a)) == 8
|
|
|
|
|
|
def test_render_gap_issue_shape():
|
|
from sanguo_factor.data_gap_check import render_gap_issue
|
|
title, body = render_gap_issue({"fa_x": ["s_new"]}, "config/data_manifest.yaml")
|
|
assert title.startswith("[数据缺口] 1 因子待接源 [") and title.endswith("]")
|
|
assert "fa_x" in body and "s_new" in body
|
|
assert "data session" in body
|
|
|
|
|
|
def test_maybe_open_dedup_skips_existing(monkeypatch, tmp_path):
|
|
import sanguo_factor.data_gap_check as dgc
|
|
monkeypatch.setenv("SANGUO_GITEA_TOKEN", "tok")
|
|
monkeypatch.setattr(dgc, "find_open_issue", lambda *a, **k: 42)
|
|
opened = []
|
|
monkeypatch.setattr(dgc, "open_gitea_issue",
|
|
lambda *a, **k: opened.append(a) or 99)
|
|
assert dgc.maybe_open_gap_issue(
|
|
{"fa_x": ["s_new"]}, "m.yaml",
|
|
ledger_path=str(tmp_path / "ledger.jsonl")) is None
|
|
assert opened == [] # 同指纹在册→跳过
|
|
|
|
|
|
def test_maybe_open_creates_when_absent(monkeypatch, tmp_path):
|
|
import sanguo_factor.data_gap_check as dgc
|
|
monkeypatch.setenv("SANGUO_GITEA_TOKEN", "tok")
|
|
monkeypatch.setattr(dgc, "find_open_issue", lambda *a, **k: None)
|
|
opened = []
|
|
monkeypatch.setattr(dgc, "open_gitea_issue",
|
|
lambda title, body, token, **k: opened.append(title) or 7)
|
|
assert dgc.maybe_open_gap_issue(
|
|
{"fa_x": ["s_new"]}, "m.yaml",
|
|
ledger_path=str(tmp_path / "ledger.jsonl")) == 7
|
|
assert opened and opened[0].startswith("[数据缺口]")
|
|
|
|
|
|
def test_maybe_open_no_token_returns_none(monkeypatch):
|
|
import sanguo_factor.data_gap_check as dgc
|
|
monkeypatch.delenv("SANGUO_GITEA_TOKEN", raising=False)
|
|
assert dgc.maybe_open_gap_issue({"fa_x": ["s_new"]}, "m.yaml") is None
|
|
|
|
|
|
def test_find_open_issue_matches_exact_title(monkeypatch):
|
|
import sanguo_factor.data_gap_check as dgc
|
|
|
|
class FakeResp:
|
|
def __enter__(self):
|
|
return self
|
|
def __exit__(self, *a):
|
|
return False
|
|
def read(self):
|
|
import json
|
|
return json.dumps(
|
|
[{"number": 5, "title": "别的单"},
|
|
{"number": 11, "title": "[数据缺口] 1 因子待接源 [abcd1234]"}
|
|
]).encode()
|
|
|
|
import urllib.request
|
|
monkeypatch.setattr(urllib.request, "urlopen", lambda req, timeout=None: FakeResp())
|
|
assert dgc.find_open_issue("[数据缺口] 1 因子待接源 [abcd1234]", "tok") == 11
|
|
assert dgc.find_open_issue("不在册标题", "tok") is None
|
|
|
|
|
|
# —— P2-13 三重加固: search 兜底/异常≠不存在/本地台账 + P3-20 地址单源 ——
|
|
import json as _json
|
|
|
|
|
|
class _JsonResp:
|
|
"""按给定 items 回 JSON 的 urlopen 假响应."""
|
|
|
|
def __init__(self, items):
|
|
self._items = items
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *a):
|
|
return False
|
|
|
|
def read(self):
|
|
return _json.dumps(self._items).encode()
|
|
|
|
|
|
def test_gitea_base_url_single_source():
|
|
"""P3-20: data_gap_check 的内网地址必须 import 自 version_registry(同一对象)."""
|
|
import sanguo_factor.data_gap_check as dgc
|
|
import sanguo_factor.version_registry as vr
|
|
assert dgc.GITEA_BASE_URL is vr.GITEA_BASE_URL
|
|
assert vr.GITEA_BASE_URL == "http://192.168.2.154:3000"
|
|
|
|
|
|
def test_find_open_issue_list_hit_no_search(monkeypatch):
|
|
import sanguo_factor.data_gap_check as dgc
|
|
urls = []
|
|
|
|
def fake(req, timeout=None):
|
|
urls.append(req.full_url)
|
|
return _JsonResp([{"number": 3,
|
|
"title": "[数据缺口] 1 因子待接源 [abcd1234]"}])
|
|
|
|
monkeypatch.setattr("urllib.request.urlopen", fake)
|
|
assert dgc.find_open_issue(
|
|
"[数据缺口] 1 因子待接源 [abcd1234]", "tok") == 3
|
|
assert len(urls) == 1 and "limit=50" in urls[0] # 首页命中不触发 search
|
|
|
|
|
|
def test_find_open_issue_search_fallback_beyond_first_page(monkeypatch):
|
|
"""首页 50 条塞满他人单,目标单在首页外——search API 按指纹兜底命中."""
|
|
import sanguo_factor.data_gap_check as dgc
|
|
title = "[数据缺口] 1 因子待接源 [abcd1234]"
|
|
urls = []
|
|
|
|
def fake(req, timeout=None):
|
|
urls.append(req.full_url)
|
|
if "limit=50" in req.full_url:
|
|
return _JsonResp([{"number": i, "title": f"占位单{i}"}
|
|
for i in range(50)])
|
|
return _JsonResp([{"number": 88, "title": title}])
|
|
|
|
monkeypatch.setattr("urllib.request.urlopen", fake)
|
|
assert dgc.find_open_issue(title, "tok") == 88
|
|
assert len(urls) == 2
|
|
assert "q=abcd1234" in urls[1] # search 按指纹兜底
|
|
|
|
|
|
def test_find_open_issue_both_miss_confirmed_absent(monkeypatch):
|
|
import sanguo_factor.data_gap_check as dgc
|
|
|
|
def fake(req, timeout=None):
|
|
return _JsonResp([])
|
|
|
|
monkeypatch.setattr("urllib.request.urlopen", fake)
|
|
assert dgc.find_open_issue("不在册标题", "tok") is None # 双 miss=确认不存在
|
|
|
|
|
|
def test_find_open_issue_list_error_raises_not_none(monkeypatch):
|
|
"""LIST 网络异常≠不存在——抛 GiteaLookupError 供调用方跳过开单."""
|
|
import pytest
|
|
|
|
import sanguo_factor.data_gap_check as dgc
|
|
|
|
def boom(req, timeout=None):
|
|
raise OSError("connection refused")
|
|
|
|
monkeypatch.setattr("urllib.request.urlopen", boom)
|
|
with pytest.raises(dgc.GiteaLookupError):
|
|
dgc.find_open_issue("[数据缺口] 1 因子待接源 [abcd1234]", "tok")
|
|
|
|
|
|
def test_find_open_issue_search_error_raises(monkeypatch):
|
|
"""search 异常同义:异常≠空,归 GiteaLookupError(调用方跳过开单本月)."""
|
|
import pytest
|
|
|
|
import sanguo_factor.data_gap_check as dgc
|
|
|
|
def fake(req, timeout=None):
|
|
if "limit=50" in req.full_url:
|
|
return _JsonResp([])
|
|
raise OSError("search timeout")
|
|
|
|
monkeypatch.setattr("urllib.request.urlopen", fake)
|
|
with pytest.raises(dgc.GiteaLookupError):
|
|
dgc.find_open_issue("[数据缺口] 1 因子待接源 [abcd1234]", "tok")
|
|
|
|
|
|
def test_maybe_open_skips_on_lookup_error(monkeypatch, tmp_path, capsys):
|
|
"""LIST 异常时跳过开单本月(stderr 留痕,下月再试),不再 fail-open."""
|
|
import sanguo_factor.data_gap_check as dgc
|
|
|
|
def boom(*a, **k):
|
|
raise dgc.GiteaLookupError("conn refused")
|
|
|
|
monkeypatch.setenv("SANGUO_GITEA_TOKEN", "tok")
|
|
monkeypatch.setattr(dgc, "find_open_issue", boom)
|
|
opened = []
|
|
monkeypatch.setattr(dgc, "open_gitea_issue",
|
|
lambda *a, **k: opened.append(a) or 99)
|
|
ledger = str(tmp_path / "ledger.jsonl")
|
|
assert dgc.maybe_open_gap_issue({"fa_x": ["s_new"]}, "m.yaml",
|
|
ledger_path=ledger) is None
|
|
assert opened == [] # 异常不开单
|
|
assert "跳过开单" in capsys.readouterr().err # stderr 留痕
|
|
|
|
|
|
def test_maybe_open_ledger_hit_skips_before_network(monkeypatch, tmp_path,
|
|
capsys):
|
|
"""台账第二道去重:指纹命中即跳过,不再打网络."""
|
|
import sanguo_factor.data_gap_check as dgc
|
|
fp = dgc.gap_fingerprint({"fa_x": ["s_new"]})
|
|
ledger = tmp_path / "ledger.jsonl"
|
|
ledger.write_text(_json.dumps(
|
|
{"fingerprint": fp, "issue": 42, "opened_at": "2026-10-01"}) + "\n",
|
|
encoding="utf-8")
|
|
|
|
def no_network(*a, **k):
|
|
raise AssertionError("台账命中后不应再触网")
|
|
|
|
monkeypatch.setenv("SANGUO_GITEA_TOKEN", "tok")
|
|
monkeypatch.setattr(dgc, "find_open_issue", no_network)
|
|
monkeypatch.setattr(dgc, "open_gitea_issue", no_network)
|
|
assert dgc.maybe_open_gap_issue({"fa_x": ["s_new"]}, "m.yaml",
|
|
ledger_path=str(ledger)) is None
|
|
assert "台账命中" in capsys.readouterr().err
|
|
|
|
|
|
def test_maybe_open_appends_ledger_after_opening(monkeypatch, tmp_path):
|
|
"""开单成功→台账 append 一行(指纹/issue号/日期三键)."""
|
|
import sanguo_factor.data_gap_check as dgc
|
|
monkeypatch.setenv("SANGUO_GITEA_TOKEN", "tok")
|
|
monkeypatch.setattr(dgc, "find_open_issue", lambda *a, **k: None)
|
|
monkeypatch.setattr(dgc, "open_gitea_issue", lambda *a, **k: 77)
|
|
ledger = tmp_path / "sub" / "ledger.jsonl" # 父目录不存在也建
|
|
assert dgc.maybe_open_gap_issue({"fa_x": ["s_new"]}, "m.yaml",
|
|
ledger_path=str(ledger)) == 77
|
|
rec = _json.loads(ledger.read_text(encoding="utf-8").strip())
|
|
assert rec["fingerprint"] == dgc.gap_fingerprint({"fa_x": ["s_new"]})
|
|
assert rec["issue"] == 77 and rec["opened_at"]
|
|
|
|
|
|
def test_append_ledger_failure_degrades_not_blocks(monkeypatch, tmp_path,
|
|
capsys):
|
|
"""台账写失败(此处:路径是目录)不阻断开单主流程,stderr 降级留痕."""
|
|
import sanguo_factor.data_gap_check as dgc
|
|
monkeypatch.setenv("SANGUO_GITEA_TOKEN", "tok")
|
|
monkeypatch.setattr(dgc, "find_open_issue", lambda *a, **k: None)
|
|
monkeypatch.setattr(dgc, "open_gitea_issue", lambda *a, **k: 77)
|
|
bad_ledger = tmp_path / "its_a_dir" # makedirs 会成功,open 会失败
|
|
bad_ledger.mkdir()
|
|
assert dgc.maybe_open_gap_issue({"fa_x": ["s_new"]}, "m.yaml",
|
|
ledger_path=str(bad_ledger)) == 77
|
|
assert "台账写入失败" in capsys.readouterr().err
|