759 lines
36 KiB
Python
759 lines
36 KiB
Python
# tests/api/test_routes_pipeline_hypotheses.py
|
||
"""向导三端点:draft 不落库/保存再验/列表.全部 mock LLM,零真调用.
|
||
|
||
分解=job 契约(10-09):POST 202 立返+后台任务,卡片列表项 decomposeJob
|
||
轮询到终态——fixture 须 with TestClient 保活事件循环(裸用=每请求独立
|
||
portal,后台 asyncio 任务响应后即被弃);decompose_jobs 内存字典逐测清空.
|
||
"""
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
from sanguo_api import hypothesis_card as hc
|
||
from sanguo_api.app import create_app
|
||
from sanguo_api.auth import create_token, set_jwt_config
|
||
|
||
DRAFT = {
|
||
"title": "高管增持公告后 60 日超额收益为正",
|
||
"logic": "If 高管真金白银增持,则内部人信息优势预示基本面改善",
|
||
"expected_sign": "positive",
|
||
"falsifiable": "若增持公告后 60 日超额收益均值<=0 则证伪",
|
||
"data_needs": ["corpus_sentiment"],
|
||
}
|
||
|
||
|
||
@pytest.fixture()
|
||
def client(tmp_path, monkeypatch):
|
||
# fixture 逐字对齐 test_routes_pipeline_monthly.py 既有范式,
|
||
# 外包 with 保活后台任务(job 化前提,见模块 docstring)
|
||
monkeypatch.setenv("SANGUO_PIPELINE_DB", str(tmp_path / "pipeline.db"))
|
||
monkeypatch.setenv("SANGUO_LLM_API_KEY", "sk-test")
|
||
set_jwt_config(secret="test", expire_minutes=60)
|
||
from sanguo_api import decompose_jobs
|
||
decompose_jobs._JOBS.clear()
|
||
decompose_jobs._LATEST.clear()
|
||
app = create_app(db_path=str(tmp_path / "t.db"), file_dir=None)
|
||
with TestClient(app) as c:
|
||
c.headers.update({"Authorization": f"Bearer {create_token('admin')}"})
|
||
yield c
|
||
|
||
|
||
@pytest.fixture()
|
||
def fake_llm(client, monkeypatch):
|
||
"""把 LLMClient.chat_json 整个替成内存假实现(记录调用)."""
|
||
import asyncio
|
||
|
||
calls, queue = [], []
|
||
|
||
class Fake:
|
||
def __init__(self, *a, **k):
|
||
pass
|
||
|
||
async def chat_json(self, messages, **k):
|
||
calls.append(messages)
|
||
if queue:
|
||
out = queue.pop(0)
|
||
if isinstance(out, Exception):
|
||
raise out
|
||
return out
|
||
return dict(DRAFT)
|
||
|
||
import sanguo_api.routes_pipeline as rp
|
||
monkeypatch.setattr(rp, "LLMClient", Fake)
|
||
monkeypatch.setattr(hc, "load_domains",
|
||
lambda: ["corpus_sentiment", "bars_daily"])
|
||
monkeypatch.setattr(rp, "_wizard_domains", hc.load_domains)
|
||
return calls, queue
|
||
|
||
|
||
class TestDraft:
|
||
def test_draft_returns_five_fields_and_does_not_persist(self, client, fake_llm):
|
||
calls, _ = fake_llm
|
||
r = client.post("/api/v1/pipeline/hypotheses/draft",
|
||
json={"sentence": "高管增持之后股价会涨"})
|
||
assert r.status_code == 200
|
||
d = r.json()["draft"]
|
||
assert d["expectedSign"] == "positive"
|
||
assert d["dataNeeds"] == ["corpus_sentiment"]
|
||
# prompt 带白名单(同源纪律)
|
||
assert "corpus_sentiment" in calls[0][0]["content"]
|
||
# 不落库
|
||
assert client.get("/api/v1/pipeline/hypotheses").json()["items"] == []
|
||
|
||
def test_draft_sentence_too_long_400(self, client, fake_llm):
|
||
r = client.post("/api/v1/pipeline/hypotheses/draft",
|
||
json={"sentence": "长" * 501})
|
||
assert r.status_code == 400
|
||
|
||
def test_draft_llm_failure_502_no_retry_at_route(self, client, fake_llm):
|
||
_, queue = fake_llm
|
||
from sanguo_api.llm import LLMError
|
||
queue.append(LLMError("LLM 返回非 JSON(两轮): 上游体片段 sk-secret123"))
|
||
r = client.post("/api/v1/pipeline/hypotheses/draft",
|
||
json={"sentence": "x"})
|
||
assert r.status_code == 502
|
||
# P3-8: detail 固定文案,不透传上游响应体任何片段(端点/账号上下文)
|
||
assert r.json()["detail"] == "LLM 上游返回异常"
|
||
assert "sk-secret123" not in r.json()["detail"]
|
||
|
||
def test_draft_llm_bad_fields_502(self, client, fake_llm):
|
||
_, queue = fake_llm
|
||
queue.append({**DRAFT, "expected_sign": "横盘"})
|
||
r = client.post("/api/v1/pipeline/hypotheses/draft",
|
||
json={"sentence": "x"})
|
||
assert r.status_code == 502
|
||
|
||
def test_draft_unconfigured_503(self, client, monkeypatch):
|
||
monkeypatch.delenv("SANGUO_LLM_API_KEY")
|
||
r = client.post("/api/v1/pipeline/hypotheses/draft",
|
||
json={"sentence": "x"})
|
||
assert r.status_code == 503 and "SANGUO_LLM_API_KEY" in r.json()["detail"]
|
||
|
||
|
||
class TestSave:
|
||
def test_save_confirmed_card_201_and_lists(self, client, fake_llm):
|
||
body = {"title": DRAFT["title"], "logic": DRAFT["logic"],
|
||
"expectedSign": "positive",
|
||
"falsifiable": DRAFT["falsifiable"],
|
||
"dataNeeds": ["corpus_sentiment"],
|
||
"sentence": "高管增持之后股价会涨"}
|
||
r = client.post("/api/v1/pipeline/hypotheses", json=body)
|
||
assert r.status_code == 201
|
||
item = r.json()["item"]
|
||
assert item["state"] == "queued" and item["id"].startswith("hyp-")
|
||
|
||
items = client.get("/api/v1/pipeline/hypotheses").json()["items"]
|
||
assert len(items) == 1 and items[0]["falsifiable"] == DRAFT["falsifiable"]
|
||
|
||
def test_save_fabricated_domain_dropped(self, client, fake_llm):
|
||
body = {"title": "t", "logic": "If a,则 b", "expectedSign": "positive",
|
||
"falsifiable": "若 a<=0 则证伪",
|
||
"dataNeeds": ["corpus_sentiment", "编造域"]}
|
||
r = client.post("/api/v1/pipeline/hypotheses", json=body)
|
||
assert r.status_code == 201
|
||
assert r.json()["item"]["dataNeeds"] == ["corpus_sentiment"]
|
||
|
||
def test_save_invalid_sign_400(self, client, fake_llm):
|
||
body = {"title": "t", "logic": "l", "expectedSign": "横盘",
|
||
"falsifiable": "f", "dataNeeds": []}
|
||
assert client.post("/api/v1/pipeline/hypotheses",
|
||
json=body).status_code == 400
|
||
|
||
def test_save_sentence_too_long_400(self, client, fake_llm):
|
||
body = {"title": "t", "logic": "l", "expectedSign": "positive",
|
||
"falsifiable": "f", "dataNeeds": [],
|
||
"sentence": "长" * 501}
|
||
assert client.post("/api/v1/pipeline/hypotheses",
|
||
json=body).status_code == 400
|
||
|
||
|
||
class TestAuth:
|
||
def test_no_token_401(self, tmp_path):
|
||
set_jwt_config(secret="test", expire_minutes=60)
|
||
c = TestClient(create_app(db_path=str(tmp_path / "t.db"), file_dir=None))
|
||
assert c.get("/api/v1/pipeline/hypotheses").status_code == 401
|
||
|
||
|
||
DECOMP_GOOD = {"factors": [
|
||
{"name": "llm_mom20", "expression": "cs_rank(ts_mean(close, 20))",
|
||
"description": "20 日收盘均价动量,量度股价短期延续性",
|
||
"justification": "20 日动量承载假设"}]}
|
||
|
||
|
||
def _seed_card(client):
|
||
r = client.post("/api/v1/pipeline/hypotheses", json={
|
||
"title": "t", "logic": "If a,则 b", "expectedSign": "positive",
|
||
"falsifiable": "若 a<=0 则证伪", "dataNeeds": ["bars_daily"],
|
||
"sentence": "s"})
|
||
return r.json()["item"]["id"]
|
||
|
||
|
||
def _poll_job(client, hyp):
|
||
"""轮询卡片列表项 decomposeJob 到终态(job 契约=前端同款轮询源)."""
|
||
import time
|
||
for _ in range(200):
|
||
items = client.get("/api/v1/pipeline/hypotheses").json()["items"]
|
||
cur = next(i for i in items if i["id"] == hyp)["decomposeJob"]
|
||
if cur and cur["status"] != "running":
|
||
return cur
|
||
time.sleep(0.05)
|
||
raise AssertionError("decompose job 未到终态(10s)")
|
||
|
||
|
||
def _wait_job(client, hyp):
|
||
"""POST 分解→202 立返(running)→轮询到终态,返回终态 job."""
|
||
r = client.post(f"/api/v1/pipeline/hypotheses/{hyp}/decompose")
|
||
assert r.status_code == 202, r.text
|
||
assert r.json()["job"]["status"] == "running"
|
||
return _poll_job(client, hyp)
|
||
|
||
|
||
class TestDecompose:
|
||
def test_decompose_registers_and_moves_card(self, client, fake_llm,
|
||
tmp_path, monkeypatch):
|
||
import yaml as _yaml
|
||
reg = tmp_path / "factor_reg.yaml"
|
||
monkeypatch.setenv("SANGUO_FACTOR_REGISTRY", str(reg))
|
||
_, queue = fake_llm
|
||
queue.append(DECOMP_GOOD)
|
||
hyp = _seed_card(client)
|
||
job = _wait_job(client, hyp)
|
||
assert job["status"] == "completed" and job["rounds"] == 1
|
||
assert len(job["registered"]) == 1
|
||
assert job["registered"][0]["source"] == "bars_daily"
|
||
items = client.get("/api/v1/pipeline/hypotheses").json()["items"]
|
||
item = next(i for i in items if i["id"] == hyp)
|
||
assert item["state"] == "building"
|
||
assert item["factorId"] == "llm_mom20"
|
||
doc = _yaml.safe_load(reg.read_text())["factors"]["llm_mom20"]
|
||
assert doc["hypothesis"] == hyp and doc["status"] == "incubating"
|
||
assert doc["origin"] == "decomposer" # #91① 条目级身份戳建条即落
|
||
v = doc["versions"][0]
|
||
assert v["params"]["expression"].startswith("cs_rank")
|
||
assert v["params"]["origin"] == "decomposer"
|
||
assert v["params"]["source"] == "bars_daily"
|
||
|
||
def test_decompose_feedback_second_round(self, client, fake_llm,
|
||
tmp_path, monkeypatch):
|
||
monkeypatch.setenv("SANGUO_FACTOR_REGISTRY",
|
||
str(tmp_path / "fr.yaml"))
|
||
_, queue = fake_llm
|
||
queue.append({"factors": [
|
||
{"name": "llm_bad", "expression": "cs_rank(pledge_pct)",
|
||
"description": "质押比例因子", "justification": "j"}]})
|
||
queue.append(DECOMP_GOOD)
|
||
hyp = _seed_card(client)
|
||
job = _wait_job(client, hyp)
|
||
assert job["rounds"] == 2
|
||
assert job["registered"][0]["name"] == "llm_mom20"
|
||
|
||
def test_decompose_404_409_503(self, client, fake_llm, monkeypatch):
|
||
assert client.post("/api/v1/pipeline/hypotheses/hyp-nope/decompose"
|
||
).status_code == 404
|
||
hyp = _seed_card(client)
|
||
from sanguo_portfolio import pipeline_store
|
||
pipeline_store.update_hypothesis_state(
|
||
pipeline_store.default_pipeline_db_path(), hyp, "graveyard",
|
||
death_reason="x")
|
||
assert client.post(f"/api/v1/pipeline/hypotheses/{hyp}/decompose"
|
||
).status_code == 422 # 决议O④墓园终态422(原409通用态细化)
|
||
monkeypatch.delenv("SANGUO_LLM_API_KEY")
|
||
hyp2 = _seed_card(client)
|
||
r = client.post(f"/api/v1/pipeline/hypotheses/{hyp2}/decompose")
|
||
assert r.status_code == 503
|
||
|
||
def test_decompose_llm_error_job_failed_no_leak(self, client, fake_llm):
|
||
_, queue = fake_llm
|
||
from sanguo_api.llm import LLMError
|
||
queue.append(LLMError("LLM 重试耗尽(HTTP 429): 上游体片段 sk-secret123"))
|
||
hyp = _seed_card(client)
|
||
job = _wait_job(client, hyp)
|
||
assert job["status"] == "failed"
|
||
# P3-8: job.error 固定文案,不透传上游响应体任何片段
|
||
assert job["error"] == "LLM 上游返回异常"
|
||
assert "sk-secret123" not in job["error"]
|
||
|
||
def test_decompose_twice_building_is_incremental(self, client, fake_llm,
|
||
tmp_path, monkeypatch):
|
||
# building 态二次分解:I-1 修复——D3 矩阵无 building→building 自迁移边,
|
||
# 二次分解=纯增量注册,卡片保持 building(I-1 前为 500 半成功)
|
||
import yaml as _yaml
|
||
reg = tmp_path / "fr_twice.yaml"
|
||
monkeypatch.setenv("SANGUO_FACTOR_REGISTRY", str(reg))
|
||
_, queue = fake_llm
|
||
queue.append(DECOMP_GOOD)
|
||
hyp = _seed_card(client)
|
||
job1 = _wait_job(client, hyp)
|
||
assert job1["status"] == "completed"
|
||
items = client.get("/api/v1/pipeline/hypotheses").json()["items"]
|
||
assert next(i for i in items if i["id"] == hyp)["state"] == "building"
|
||
queue.append({"factors": [
|
||
{"name": "llm_vol5", "expression": "cs_rank(ts_std(close, 5))",
|
||
"description": "5 日波动率因子自述",
|
||
"justification": "5 日波动承载假设"}]})
|
||
job2 = _wait_job(client, hyp)
|
||
assert job2["status"] == "completed"
|
||
assert job2["registered"][0]["name"] == "llm_vol5"
|
||
assert job2["registered"][0]["description"] == "5 日波动率因子自述"
|
||
# codex review MEDIUM-2:回读 registry 验证 upsert_factor 真落盘(非仅回传)
|
||
import os as _os
|
||
import yaml as _yaml
|
||
with open(_os.environ["SANGUO_FACTOR_REGISTRY"], encoding="utf-8") as f:
|
||
reg_disk = _yaml.safe_load(f)
|
||
assert reg_disk["factors"]["llm_vol5"]["description"] == "5 日波动率因子自述"
|
||
items = client.get("/api/v1/pipeline/hypotheses").json()["items"]
|
||
assert next(i for i in items if i["id"] == hyp)["state"] == "building"
|
||
factors = _yaml.safe_load(reg.read_text())["factors"]
|
||
assert "llm_mom20" in factors and "llm_vol5" in factors # 两次注册都在
|
||
|
||
def test_decompose_lock_exists(self):
|
||
# P2-6:decompose 全程持 _DECOMPOSE_LOCK(仿 _GRADUATE_LOCK,进程内
|
||
# 串行化同卡并发,registry 整文件覆盖丢更新窗口闭合)。并发真测不做
|
||
# ——线程池+async 端点组合测不划算;锁内注册+转态的行为回归由本类
|
||
# 既有各例钉死,此处只钉锁存在防回退。
|
||
import threading
|
||
|
||
from sanguo_api import routes_pipeline as rp
|
||
assert isinstance(rp._DECOMPOSE_LOCK, type(threading.Lock()))
|
||
assert isinstance(rp._GRADUATE_LOCK, type(threading.Lock()))
|
||
def test_decompose_similarity_only_on_new_entry(self, client, fake_llm,
|
||
tmp_path, monkeypatch):
|
||
"""codex review 仅建条纪律:条目已存在时 upsert 不重写 similarity
|
||
(强化形态——同名异假设 upsert 会拒,同名同假设幂等返回,此测后者)."""
|
||
import yaml
|
||
# 先建卡拿 hyp id,再预置 registry:fa_dup 绑同一假设(幂等重注册路径)
|
||
hyp = _seed_card(client)
|
||
reg_path = tmp_path / "r_sim2.yaml"
|
||
monkeypatch.setenv("SANGUO_FACTOR_REGISTRY", str(reg_path))
|
||
reg = {"factors": {
|
||
"fa_old": {"name": "fa_old", "hypothesis": "H-1",
|
||
"status": "assessable",
|
||
"versions": [{"v": 1, "commit": "x",
|
||
"params": {"expression":
|
||
"ts_mean(close, 5) * turnover"},
|
||
"effective_from": "2026-09-01"}]},
|
||
"fa_dup": {"name": "fa_dup", "hypothesis": hyp,
|
||
"status": "incubating",
|
||
"versions": [{"v": 1, "commit": "x",
|
||
"params": {"expression": "close + open"},
|
||
"effective_from": "2026-09-01"}]}}}
|
||
with open(reg_path, "w", encoding="utf-8") as f:
|
||
yaml.safe_dump(reg, f, allow_unicode=True)
|
||
# 绕开 LLM 与 name_taken 硬门,直接喂「已存在名」的 passed
|
||
import sanguo_api.routes_pipeline as rp
|
||
|
||
async def fake_run(client_, card, **k):
|
||
return {"passed": [{"name": "fa_dup",
|
||
"expression": "ts_mean(close, 5) / volume",
|
||
"description": "同窗均值除量",
|
||
"justification": "j"}],
|
||
"failed": [], "rounds": 1}
|
||
|
||
monkeypatch.setattr(rp, "_run_decompose", fake_run)
|
||
assert _wait_job(client, hyp)["status"] == "completed"
|
||
with open(reg_path, encoding="utf-8") as f:
|
||
reg2 = yaml.safe_load(f)
|
||
# 已存在条目:similarity 不写入(旧值 None 保持 None)
|
||
assert "similarity" not in reg2["factors"]["fa_dup"]
|
||
|
||
def test_decompose_register_similarity_hint(self, client, fake_llm,
|
||
tmp_path, monkeypatch):
|
||
"""AST 防换皮提示(spec §4.2 2026-10-10):新因子注册时对既有池比对,
|
||
疑似换皮(≥0.6)存 entry.similarity——硬门(dup_subtree)外的部分公共
|
||
子树对仍可注册,提示非硬拒,判定人做."""
|
||
import yaml
|
||
reg_path = tmp_path / "r_sim.yaml"
|
||
monkeypatch.setenv("SANGUO_FACTOR_REGISTRY", str(reg_path))
|
||
# 既有池:fa_old 与候选共享 ts_mean(close,5) 子树(4 节点<硬门阈 8)
|
||
reg = {"factors": {"fa_old": {
|
||
"name": "fa_old", "hypothesis": "H-1", "status": "assessable",
|
||
"versions": [{"v": 1, "commit": "x",
|
||
"params": {"expression":
|
||
"ts_mean(close, 5) * turnover"},
|
||
"effective_from": "2026-09-01"}]}}}
|
||
with open(reg_path, "w", encoding="utf-8") as f:
|
||
yaml.safe_dump(reg, f, allow_unicode=True)
|
||
_, queue = fake_llm
|
||
queue.append({"factors": [
|
||
{"name": "llm_twin", "expression": "ts_mean(close, 5) / volume",
|
||
"description": "同窗均值除量", "justification": "j"}]})
|
||
hyp = _seed_card(client)
|
||
assert _wait_job(client, hyp)["status"] == "completed"
|
||
with open(reg_path, encoding="utf-8") as f:
|
||
reg2 = yaml.safe_load(f)
|
||
entry = reg2["factors"]["llm_twin"]
|
||
assert entry["similarity"] == [{"name": "fa_old", "ratio": 0.6667}]
|
||
d = client.get("/api/v1/pipeline/factors/llm_twin/detail").json()
|
||
assert d["factor"]["similarity"] == [{"name": "fa_old", "ratio": 0.6667}]
|
||
# 列表端点同带出(工厂板换皮徽标数据源)
|
||
items = client.get("/api/v1/pipeline/factors").json()["items"]
|
||
row = next(i for i in items if i["id"] == "llm_twin")
|
||
assert row["similarity"] == [{"name": "fa_old", "ratio": 0.6667}]
|
||
|
||
|
||
|
||
"""P3-7: save 端点 source 白名单+上限(此前零白名单零上限,任意串入库)."""
|
||
|
||
|
||
|
||
|
||
class TestDecomposeJobs:
|
||
"""job 化新增语义(10-09,QA 范式):202 立返/在飞 409/台账跨重启/僵尸判死."""
|
||
|
||
def test_post_returns_202_running_immediately(self, client, fake_llm,
|
||
tmp_path, monkeypatch):
|
||
import asyncio
|
||
|
||
monkeypatch.setenv("SANGUO_FACTOR_REGISTRY", str(tmp_path / "r.yaml"))
|
||
hyp = _seed_card(client)
|
||
|
||
async def slow(client_, card, **k):
|
||
await asyncio.sleep(0.4)
|
||
return {"passed": [], "failed": [], "rounds": 1}
|
||
|
||
import sanguo_api.routes_pipeline as rp
|
||
monkeypatch.setattr(rp, "_run_decompose", slow)
|
||
r = client.post(f"/api/v1/pipeline/hypotheses/{hyp}/decompose")
|
||
assert r.status_code == 202, r.text
|
||
job = r.json()["job"]
|
||
assert job["status"] == "running" and job["jobId"].startswith("job-")
|
||
# 在飞期间二次 POST → 409(同卡互斥,治重复点击堆积)
|
||
r2 = client.post(f"/api/v1/pipeline/hypotheses/{hyp}/decompose")
|
||
assert r2.status_code == 409
|
||
assert _poll_job(client, hyp)["status"] == "completed"
|
||
|
||
def test_result_survives_memory_clear(self, client, fake_llm,
|
||
tmp_path, monkeypatch):
|
||
"""台账跨重启:内存清空(=服务重启)后已完成结果仍可读(run 台账先于页面)."""
|
||
monkeypatch.setenv("SANGUO_FACTOR_REGISTRY", str(tmp_path / "r2.yaml"))
|
||
_, queue = fake_llm
|
||
queue.append(DECOMP_GOOD)
|
||
hyp = _seed_card(client)
|
||
assert _wait_job(client, hyp)["status"] == "completed"
|
||
from sanguo_api import decompose_jobs
|
||
decompose_jobs._JOBS.clear()
|
||
decompose_jobs._LATEST.clear()
|
||
items = client.get("/api/v1/pipeline/hypotheses").json()["items"]
|
||
cur = next(i for i in items if i["id"] == hyp)["decomposeJob"]
|
||
assert cur["status"] == "completed"
|
||
assert cur["registered"][0]["name"] == "llm_mom20"
|
||
|
||
def test_zombie_running_reconciled_failed(self, client, tmp_path,
|
||
monkeypatch):
|
||
"""台账僵尸 running(重启打断)无内存主→读侧判 failed 并回写."""
|
||
from sanguo_portfolio import pipeline_store
|
||
hyp = _seed_card(client)
|
||
pipeline_store.upsert_decompose_job(
|
||
pipeline_store.default_pipeline_db_path(),
|
||
{"jobId": "job-zombie", "hypId": hyp, "status": "running",
|
||
"rounds": None, "registered": [], "failed": [], "error": None,
|
||
# 未来时间戳保证它是最新的(started_at DESC 取这条)
|
||
"startedAt": "2099-01-01T00:00:00", "finishedAt": None})
|
||
items = client.get("/api/v1/pipeline/hypotheses").json()["items"]
|
||
cur = next(i for i in items if i["id"] == hyp)["decomposeJob"]
|
||
assert cur["status"] == "failed"
|
||
assert "中断" in cur["error"]
|
||
# 回写台账:再读仍 failed(一处真相,非每次读侧重算)
|
||
row = pipeline_store.latest_decompose_job(
|
||
pipeline_store.default_pipeline_db_path(), hyp)
|
||
assert row["status"] == "failed"
|
||
|
||
|
||
class TestD7Visibility:
|
||
"""D7 落成边亮灯(10-09 拍板):三件套数据面+工厂 origin+琥珀待办."""
|
||
|
||
def test_decompose_jobs_history_endpoint(self, client, tmp_path):
|
||
"""件①1b:批次历史端点=台账新→旧(QA tasks/list 同形状 {items})."""
|
||
from sanguo_portfolio import pipeline_store
|
||
db = pipeline_store.default_pipeline_db_path()
|
||
hyp = _seed_card(client)
|
||
for i, (jid, st) in enumerate([("job-old", "completed"),
|
||
("job-new", "completed")]):
|
||
pipeline_store.upsert_decompose_job(db, {
|
||
"jobId": jid, "hypId": hyp, "status": st, "rounds": 1,
|
||
"registered": [{"name": f"f{i}", "source": "bars_daily",
|
||
"expression": "e", "justification": "j"}],
|
||
"failed": [], "error": None,
|
||
"startedAt": f"2026-10-09T10:1{i}:00", "finishedAt": None})
|
||
r = client.get(f"/api/v1/pipeline/hypotheses/{hyp}/decompose-jobs")
|
||
assert r.status_code == 200
|
||
items = r.json()["items"]
|
||
assert [i["jobId"] for i in items] == ["job-new", "job-old"]
|
||
assert items[0]["registered"][0]["name"] == "f1"
|
||
# 不存在的卡 404
|
||
assert client.get("/api/v1/pipeline/hypotheses/hyp-x/decompose-jobs"
|
||
).status_code == 404
|
||
|
||
def test_hypotheses_list_includes_factors(self, client, fake_llm,
|
||
tmp_path, monkeypatch):
|
||
"""件①1a:列表项内联血统反查因子清单(名/表达式/origin/落成日)."""
|
||
monkeypatch.setenv("SANGUO_FACTOR_REGISTRY", str(tmp_path / "r1.yaml"))
|
||
_, queue = fake_llm
|
||
queue.append(DECOMP_GOOD)
|
||
hyp = _seed_card(client)
|
||
job = _wait_job(client, hyp)
|
||
assert job["status"] == "completed"
|
||
items = client.get("/api/v1/pipeline/hypotheses").json()["items"]
|
||
item = next(i for i in items if i["id"] == hyp)
|
||
assert [f["name"] for f in item["factors"]] == ["llm_mom20"]
|
||
assert item["factors"][0]["origin"] == "decomposer"
|
||
assert item["factors"][0]["expression"].startswith("cs_rank")
|
||
assert item["factors"][0]["createdAt"] # 落成日=首版 effective_from
|
||
# 批次归属(分割线数据源):registered 含该名的最新 job
|
||
assert item["factors"][0]["batch"]["jobId"] == job["jobId"]
|
||
|
||
def test_progress_reports_rounds_while_running(self, client, fake_llm,
|
||
tmp_path, monkeypatch):
|
||
"""件①1c:在飞卡片 decomposeJob.progress=进程内真轮数(非 QA 摆设)."""
|
||
import asyncio
|
||
|
||
monkeypatch.setenv("SANGUO_FACTOR_REGISTRY", str(tmp_path / "r3.yaml"))
|
||
hyp = _seed_card(client)
|
||
seen = []
|
||
|
||
async def slow(client_, card, **k):
|
||
cb = k.get("on_progress")
|
||
if cb:
|
||
cb(round_no=1, total_rounds=3, passed=3, regen=1,
|
||
message="第 1/3 轮校验完成")
|
||
await asyncio.sleep(0.15)
|
||
cb(round_no=2, total_rounds=3, passed=6, regen=1,
|
||
message="第 2/3 轮失败者反馈重生成中")
|
||
await asyncio.sleep(0.15)
|
||
seen.append(True)
|
||
return {"passed": [], "failed": [], "rounds": 3}
|
||
|
||
import sanguo_api.routes_pipeline as rp
|
||
monkeypatch.setattr(rp, "_run_decompose", slow)
|
||
r = client.post(f"/api/v1/pipeline/hypotheses/{hyp}/decompose")
|
||
assert r.status_code == 202
|
||
job = _poll_job(client, hyp)
|
||
assert job["status"] == "completed"
|
||
# 进度曾可观测(轮询窗口内读到 progress;0.15s×2>轮询间隔 50ms)
|
||
items = client.get("/api/v1/pipeline/hypotheses").json()["items"]
|
||
cur = next(i for i in items if i["id"] == hyp)["decomposeJob"]
|
||
assert cur.get("progress") is None # 终态不带过程态
|
||
|
||
def test_progress_visible_mid_flight(self, client, fake_llm, tmp_path,
|
||
monkeypatch):
|
||
"""件①1c 真断言:running 期间列表项 progress.currentRound 递增可见."""
|
||
import asyncio
|
||
import time
|
||
|
||
monkeypatch.setenv("SANGUO_FACTOR_REGISTRY", str(tmp_path / "r4.yaml"))
|
||
hyp = _seed_card(client)
|
||
|
||
async def slow(client_, card, **k):
|
||
cb = k.get("on_progress")
|
||
if cb:
|
||
cb(round_no=1, total_rounds=3, passed=3, regen=1,
|
||
message="第 1/3 轮校验完成")
|
||
await asyncio.sleep(0.4)
|
||
return {"passed": [], "failed": [], "rounds": 1}
|
||
|
||
import sanguo_api.routes_pipeline as rp
|
||
monkeypatch.setattr(rp, "_run_decompose", slow)
|
||
assert client.post(
|
||
f"/api/v1/pipeline/hypotheses/{hyp}/decompose").status_code == 202
|
||
got = None
|
||
for _ in range(80):
|
||
items = client.get("/api/v1/pipeline/hypotheses").json()["items"]
|
||
cur = next(i for i in items if i["id"] == hyp)["decomposeJob"]
|
||
if cur and cur.get("progress"):
|
||
got = cur["progress"]
|
||
break
|
||
time.sleep(0.02)
|
||
assert got and got["currentRound"] == 1 and got["totalRounds"] == 3
|
||
assert got["passed"] == 3 and got["regen"] == 1
|
||
assert _poll_job(client, hyp)["status"] == "completed"
|
||
|
||
def test_todos_amber_until_decomposed(self, client, fake_llm,
|
||
tmp_path, monkeypatch):
|
||
"""件③:queued 卡=琥珀待办;分解转 building 即消行."""
|
||
monkeypatch.setenv("SANGUO_FACTOR_REGISTRY", str(tmp_path / "r5.yaml"))
|
||
_, queue = fake_llm
|
||
hyp = _seed_card(client)
|
||
|
||
def _amber_titles():
|
||
return [t["title"] for t in client.get(
|
||
"/api/v1/pipeline/todos").json()["items"]
|
||
if t["id"] == f"decompose-{hyp}"]
|
||
|
||
assert _amber_titles(), "queued 卡应上琥珀待办"
|
||
queue.append(DECOMP_GOOD)
|
||
assert _wait_job(client, hyp)["status"] == "completed"
|
||
assert not _amber_titles(), "building 卡不应再上待办"
|
||
|
||
def test_factors_include_origin_and_hypothesis(self, client, fake_llm,
|
||
tmp_path, monkeypatch):
|
||
"""件②:工厂列表带 origin 身份戳+来源卡血统(manual 缺省)."""
|
||
monkeypatch.setenv("SANGUO_FACTOR_REGISTRY", str(tmp_path / "r6.yaml"))
|
||
_, queue = fake_llm
|
||
queue.append(DECOMP_GOOD)
|
||
hyp = _seed_card(client)
|
||
assert _wait_job(client, hyp)["status"] == "completed"
|
||
items = client.get("/api/v1/pipeline/factors").json()["items"]
|
||
row = next(i for i in items if i["id"] == "llm_mom20")
|
||
assert row["origin"] == "decomposer" and row["hypothesis"] == hyp
|
||
|
||
def test_factor_detail_decomposer_birth(self, client, fake_llm,
|
||
tmp_path, monkeypatch):
|
||
"""验收反馈①:decomposer 因子详情带出生档案(批次=台账含该名最新 job)."""
|
||
monkeypatch.setenv("SANGUO_FACTOR_REGISTRY", str(tmp_path / "r7.yaml"))
|
||
_, queue = fake_llm
|
||
queue.append(DECOMP_GOOD)
|
||
hyp = _seed_card(client)
|
||
job = _wait_job(client, hyp)
|
||
r = client.get("/api/v1/pipeline/factors/llm_mom20/detail")
|
||
assert r.status_code == 200
|
||
d = r.json()
|
||
f = d["factor"]
|
||
assert f["origin"] == "decomposer" and f["hypothesis"] == hyp
|
||
assert f["expression"].startswith("cs_rank")
|
||
assert f["createdAt"] and f["justification"]
|
||
assert d["birth"]["jobId"] == job["jobId"]
|
||
assert d["birth"]["rounds"] == 1 and d["birth"]["hypId"] == hyp
|
||
# 未知因子 404
|
||
assert client.get("/api/v1/pipeline/factors/nope/detail"
|
||
).status_code == 404
|
||
|
||
class TestSaveSourceGuard:
|
||
"""P3-7: save 端点 source 白名单+上限(此前零白名单零上限,任意串入库)."""
|
||
|
||
def test_save_source_off_whitelist_400(self, client, fake_llm):
|
||
body = {"title": "t", "logic": "If a,则 b", "expectedSign": "positive",
|
||
"falsifiable": "若 a<=0 则证伪", "dataNeeds": [],
|
||
"source": "任意自由文本来源"}
|
||
r = client.post("/api/v1/pipeline/hypotheses", json=body)
|
||
assert r.status_code == 400 and "source" in r.json()["detail"]
|
||
|
||
def test_save_source_manual_accepted(self, client, fake_llm):
|
||
body = {"title": "t", "logic": "If a,则 b", "expectedSign": "positive",
|
||
"falsifiable": "若 a<=0 则证伪", "dataNeeds": [],
|
||
"source": "manual"}
|
||
r = client.post("/api/v1/pipeline/hypotheses", json=body)
|
||
assert r.status_code == 201
|
||
assert r.json()["item"]["source"] == "manual"
|
||
|
||
|
||
class TestCardVerdict:
|
||
"""决议O②:人工判卡片死(强断言想法死)——活因子陪葬,cause 带(随卡判死)."""
|
||
|
||
@pytest.fixture()
|
||
def env2(self, client, monkeypatch, tmp_path):
|
||
"""registry 指 tmp(绕种子 merge)+预置一卡两因子(一活一死)."""
|
||
reg = tmp_path / "reg.yaml"
|
||
reg.write_text(
|
||
"factors:\n"
|
||
" fa_live:\n"
|
||
" name: fa_live\n status: incubating\n"
|
||
" hypothesis: hyp-kill1\n origin: decomposer\n"
|
||
" fa_dead:\n"
|
||
" name: fa_dead\n status: graveyard\n"
|
||
" hypothesis: hyp-kill1\n cause: 旧死\n"
|
||
" origin: decomposer\n", encoding="utf-8")
|
||
monkeypatch.setenv("SANGUO_FACTOR_REGISTRY", str(reg))
|
||
return reg
|
||
|
||
def _mk_card(self, state="building"):
|
||
from sanguo_portfolio import pipeline_store as ps
|
||
import os
|
||
db = os.environ["SANGUO_PIPELINE_DB"]
|
||
ps.insert_hypothesis(db, {
|
||
"title": "判死测试卡", "logic": "x",
|
||
"expected_sign": "positive", "falsifiable": "y"})
|
||
import sqlite3
|
||
with sqlite3.connect(db) as conn:
|
||
conn.execute("UPDATE hypotheses SET id='hyp-kill1', "
|
||
f"state='{state}' WHERE title='判死测试卡'")
|
||
return "hyp-kill1"
|
||
|
||
def test_kill_card_buries_live_siblings(self, client, env2):
|
||
import yaml
|
||
hyp = self._mk_card("building")
|
||
r = client.post(f"/api/v1/pipeline/hypotheses/{hyp}/verdict",
|
||
json={"reason": "逻辑前提失效:增持数据口径变更"})
|
||
assert r.status_code == 200
|
||
doc = r.json()
|
||
assert doc["buried"] == ["fa_live"] # 死的不重复陪葬
|
||
with open(env2, encoding="utf-8") as f:
|
||
reg = yaml.safe_load(f)
|
||
assert reg["factors"]["fa_live"]["status"] == "graveyard"
|
||
assert reg["factors"]["fa_live"]["cause"] == \
|
||
"逻辑前提失效:增持数据口径变更(随卡判死)"
|
||
from sanguo_portfolio import pipeline_store as ps
|
||
import os
|
||
card = [c for c in ps.list_hypotheses(os.environ["SANGUO_PIPELINE_DB"])
|
||
if c["id"] == hyp][0]
|
||
assert card["state"] == "graveyard"
|
||
assert card["deathReason"] == "逻辑前提失效:增持数据口径变更"
|
||
|
||
def test_kill_requires_reason(self, client, env2):
|
||
self._mk_card()
|
||
r = client.post("/api/v1/pipeline/hypotheses/hyp-kill1/verdict",
|
||
json={"reason": " "})
|
||
assert r.status_code == 422
|
||
|
||
def test_graveyard_card_reject_422(self, client, env2):
|
||
import sqlite3, os
|
||
self._mk_card("graveyard")
|
||
r = client.post("/api/v1/pipeline/hypotheses/hyp-kill1/verdict",
|
||
json={"reason": "再死一次"})
|
||
assert r.status_code == 422 and "墓园" in r.json()["detail"]
|
||
|
||
def test_promoted_card_reject_422(self, client, env2):
|
||
self._mk_card("promoted")
|
||
r = client.post("/api/v1/pipeline/hypotheses/hyp-kill1/verdict",
|
||
json={"reason": "毕业卡判死"})
|
||
assert r.status_code == 422 and "promoted" in r.json()["detail"]
|
||
|
||
def test_zero_factor_card_pure_idea_kill(self, client, monkeypatch, tmp_path):
|
||
"""0 因子卡=纯想法毙掉,无陪葬也 200."""
|
||
reg = tmp_path / "empty.yaml"
|
||
reg.write_text("factors: {}\n", encoding="utf-8")
|
||
monkeypatch.setenv("SANGUO_FACTOR_REGISTRY", str(reg))
|
||
hyp = self._mk_card("queued")
|
||
r = client.post(f"/api/v1/pipeline/hypotheses/{hyp}/verdict",
|
||
json={"reason": "想法被新文献证伪"})
|
||
assert r.status_code == 200 and r.json()["buried"] == []
|
||
|
||
def test_kill_while_decompose_running_409(self, client, monkeypatch):
|
||
"""codex 终审 C2:在飞分解时判死 409 快拒(防 worker 旧对象覆盖复活墓园)."""
|
||
from sanguo_api import decompose_jobs
|
||
monkeypatch.setattr(decompose_jobs, "is_running", lambda hyp_id: True)
|
||
r = client.post("/api/v1/pipeline/hypotheses/hyp-any/verdict",
|
||
json={"reason": "分解中判死"})
|
||
assert r.status_code == 409 and "分解" in r.json()["detail"]
|
||
|
||
def test_kill_reason_too_long_422(self, client, env2, monkeypatch, tmp_path):
|
||
"""codex 终审 NOTE2:reason 上限 500,超限 422."""
|
||
reg = tmp_path / "empty2.yaml"
|
||
reg.write_text("factors: {}\n", encoding="utf-8")
|
||
monkeypatch.setenv("SANGUO_FACTOR_REGISTRY", str(reg))
|
||
self._mk_card("queued")
|
||
r = client.post("/api/v1/pipeline/hypotheses/hyp-kill1/verdict",
|
||
json={"reason": "长" * 501})
|
||
assert r.status_code == 422 and "reason 过长" in r.json()["detail"]
|
||
|
||
|
||
class TestFactorsCauseAndDecompose422:
|
||
def test_factors_carry_cause(self, client, monkeypatch, tmp_path):
|
||
reg = tmp_path / "reg.yaml"
|
||
reg.write_text(
|
||
"factors:\n"
|
||
" fa_d:\n name: fa_d\n status: graveyard\n"
|
||
" hypothesis: hyp-c1\n cause: IC 两窗反号\n"
|
||
" origin: decomposer\n"
|
||
" fa_l:\n name: fa_l\n status: incubating\n"
|
||
" hypothesis: hyp-c1\n origin: decomposer\n",
|
||
encoding="utf-8")
|
||
monkeypatch.setenv("SANGUO_FACTOR_REGISTRY", str(reg))
|
||
from sanguo_portfolio import pipeline_store as ps
|
||
import os, sqlite3
|
||
db = os.environ["SANGUO_PIPELINE_DB"]
|
||
ps.insert_hypothesis(db, {
|
||
"title": "透传测试卡", "logic": "x",
|
||
"expected_sign": "positive", "falsifiable": "y"})
|
||
with sqlite3.connect(db) as conn:
|
||
conn.execute("UPDATE hypotheses SET id='hyp-c1' "
|
||
"WHERE title='透传测试卡'")
|
||
items = client.get("/api/v1/pipeline/hypotheses").json()["items"]
|
||
fs = next(i for i in items if i["id"] == "hyp-c1")["factors"]
|
||
cmap = {f["name"]: f.get("cause") for f in fs}
|
||
assert cmap["fa_d"] == "IC 两窗反号"
|
||
assert cmap["fa_l"] is None
|
||
|
||
def test_decompose_graveyard_card_422(self, client, monkeypatch, tmp_path):
|
||
from sanguo_portfolio import pipeline_store as ps
|
||
import os, sqlite3
|
||
db = os.environ["SANGUO_PIPELINE_DB"]
|
||
ps.insert_hypothesis(db, {
|
||
"title": "终态分解卡", "logic": "x",
|
||
"expected_sign": "positive", "falsifiable": "y"})
|
||
with sqlite3.connect(db) as conn:
|
||
conn.execute("UPDATE hypotheses SET id='hyp-g1', state='graveyard', "
|
||
"death_reason='全灭' WHERE title='终态分解卡'")
|
||
monkeypatch.setenv("SANGUO_LLM_API_KEY", "sk-test")
|
||
r = client.post("/api/v1/pipeline/hypotheses/hyp-g1/decompose")
|
||
assert r.status_code == 422
|
||
assert "终态" in r.json()["detail"] or "墓园" in r.json()["detail"]
|