208 lines
8.3 KiB
Python
208 lines
8.3 KiB
Python
"""tests/test_context_wiring.py —— 生产启动链接线（BUG-R1-03 回归锁定）。
|
||
|
||
此前 init_context→build_context 不传 flow_store（恒 None），部署态 ask/
|
||
chat_qa 引用块恒空。本文件锁定：
|
||
1. init_context() 真实接线：临时 flow_docs 目录 + 假嵌入 → context.flow_store
|
||
非 None 且可检索。
|
||
2. ask 端点经接线后的 context 返回引用块（citations 非空）。
|
||
3. flow_docs 缺失时降级 None 不阻断启动（init_flow_store 契约保持）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import sys
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
pytestmark = pytest.mark.bioflow_unit
|
||
|
||
|
||
def _run(coro):
|
||
return asyncio.new_event_loop().run_until_complete(coro)
|
||
|
||
|
||
def _seed_flow_docs(flow_dir: Path) -> list[dict]:
|
||
"""临时 flow_docs：用真实 FlowStore.add_chunk 入库 2 块（假嵌入注入）。
|
||
|
||
假嵌入必须产 1024 维（RetrievalIndex IndexFlatIP 固定维，对齐 vstore
|
||
tests/fake_embed_utils.DIM=1024；词级 shingle 哈希到 1024 维）。
|
||
"""
|
||
import hashlib
|
||
import re
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent))
|
||
from vstore.flow import FlowStore, set_embed_fn
|
||
|
||
DIM = 1024
|
||
|
||
def fake_embed(texts):
|
||
vectors = []
|
||
for t in texts:
|
||
vec = [0.0] * DIM
|
||
# 中文字符 2-gram + ASCII 词 shingle 哈希到维度（主题粗向量替代）
|
||
tokens = re.findall(r"[A-Za-z0-9_\-]+", t)
|
||
for tok in tokens:
|
||
for i in range(max(1, len(tok) - 3)):
|
||
sh = tok[i:i + 4].lower()
|
||
idx = int(hashlib.md5(sh.encode()).hexdigest()[:8], 16) % DIM
|
||
vec[idx] += 1.0
|
||
for m in re.finditer(r"[一-鿿]{2,}", t):
|
||
seg = m.group(0)
|
||
for i in range(len(seg) - 1):
|
||
sh = seg[i:i + 2]
|
||
idx = int(hashlib.md5(sh.encode()).hexdigest()[:8], 16) % DIM
|
||
vec[idx] += 1.0
|
||
norm = sum(v * v for v in vec) ** 0.5
|
||
if norm > 0:
|
||
vec = [v / norm for v in vec]
|
||
vectors.append(vec)
|
||
return vectors
|
||
|
||
set_embed_fn(fake_embed)
|
||
store = FlowStore(flow_dir)
|
||
store.add_chunk(
|
||
{
|
||
"source_id": "contract_identify", "chunk_key": "contract_identify__abc123",
|
||
"source": "stage_contract", "doc_kind": "contract",
|
||
"title": "Identify Stage Contract", "title_path": "Identify | ## 1",
|
||
"heading": "## 1. 输入", "stage_name": "identify",
|
||
"path": "gfa_core/schemas/stages/identify_contract_v1.md",
|
||
"cite": {"anchor": "## 1", "line": 1},
|
||
},
|
||
"Identify 阶段需要全蛋白 FASTA 与 HMM 文件作为输入。",
|
||
vector=None,
|
||
)
|
||
store.add_chunk(
|
||
{
|
||
"source_id": "corpus_main", "chunk_key": "corpus_main__def456",
|
||
"source": "corpus", "doc_kind": "corpus",
|
||
"title": "全流程文档", "title_path": "第一部分 | ## 4. identify",
|
||
"heading": "## 4. identify：家族成员初筛",
|
||
"stage_name": "",
|
||
"path": "docs/bioflow_corpus/PlantAI_现有分析全流程.md",
|
||
"cite": {"anchor": "## 4. identify：家族成员初筛"},
|
||
},
|
||
"identify 阶段对输入蛋白序列进行 HMMER 搜索与结构域确认。",
|
||
vector=None,
|
||
)
|
||
return [
|
||
{"chunk_key": "contract_identify__abc123", "id": "contract_identify__abc123",
|
||
"source": "stage_contract", "doc_kind": "contract", "stage_name": "identify",
|
||
"title": "Identify Stage Contract", "title_path": "Identify | ## 1",
|
||
"path": "gfa_core/schemas/stages/identify_contract_v1.md",
|
||
"cite": {"anchor": "## 1", "line": 1}, "text": "Identify 阶段需要全蛋白 FASTA。"},
|
||
]
|
||
|
||
|
||
def test_init_context_wires_flow_store(tmp_path, monkeypatch):
|
||
"""BUG-R1-03 回归：init_context() 接线后 context.flow_store 非 None。"""
|
||
flow_dir = tmp_path / "flow_docs"
|
||
_seed_flow_docs(flow_dir)
|
||
|
||
import app.config as config_mod
|
||
import app.context as context_mod
|
||
|
||
s = config_mod.Settings(
|
||
_env_file=None,
|
||
data_root=str(tmp_path / "data"),
|
||
jobs_dir=str(tmp_path / "jobs"),
|
||
db_path=str(tmp_path / "bioflow.db"),
|
||
archives_dir=str(tmp_path / "archives"),
|
||
image_map_path=str(tmp_path / "stage_image_map.json"),
|
||
flow_docs_dir=str(flow_dir),
|
||
)
|
||
# 真实映射表避免 load_kernel 依赖测试表：指向仓库 A1 交付物
|
||
s2 = config_mod.Settings(
|
||
_env_file=None,
|
||
data_root=str(tmp_path / "data"),
|
||
jobs_dir=str(tmp_path / "jobs"),
|
||
db_path=str(tmp_path / "bioflow.db"),
|
||
archives_dir=str(tmp_path / "archives"),
|
||
image_map_path=str(Path(__file__).resolve().parent.parent.parent
|
||
/ "bioflow_kernel" / "stage_image_map.json"),
|
||
flow_docs_dir=str(flow_dir),
|
||
)
|
||
monkeypatch.setattr(config_mod, "get_settings", lambda: s2)
|
||
monkeypatch.setattr(context_mod, "get_settings", lambda: s2)
|
||
|
||
ctx = _run(context_mod.init_context(db_path=str(tmp_path / "bioflow.db")))
|
||
try:
|
||
assert ctx.flow_store is not None, "init_context 未接线 flow_store（BUG-R1-03）"
|
||
# 真实检索：引用块可召回
|
||
result = ctx.flow_store.search("identify 需要什么输入", top_k=4)
|
||
assert result.get("items"), "接线后的 FlowStore 检索应为非空"
|
||
finally:
|
||
_run(context_mod.shutdown_context())
|
||
from app.db import close_db
|
||
|
||
_run(close_db())
|
||
context_mod.context = context_mod.Context() # 还原全局（后续用例隔离）
|
||
|
||
|
||
def test_flow_ask_returns_citations_after_wiring(tmp_path, monkeypatch):
|
||
"""BUG-R1-03 回归：接线后 ask 端点返回引用块（citations 非空）。"""
|
||
flow_dir = tmp_path / "flow_docs"
|
||
_seed_flow_docs(flow_dir)
|
||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent))
|
||
from vstore.flow import set_embed_fn
|
||
|
||
import app.config as config_mod
|
||
import app.context as context_mod
|
||
from app.routers import bioflow as r_flow
|
||
from app.schemas import FlowAskRequest
|
||
|
||
s = config_mod.Settings(
|
||
_env_file=None,
|
||
data_root=str(tmp_path / "data"),
|
||
jobs_dir=str(tmp_path / "jobs"),
|
||
db_path=str(tmp_path / "bioflow.db"),
|
||
archives_dir=str(tmp_path / "archives"),
|
||
image_map_path=str(Path(__file__).resolve().parent.parent.parent
|
||
/ "bioflow_kernel" / "stage_image_map.json"),
|
||
flow_docs_dir=str(flow_dir),
|
||
)
|
||
monkeypatch.setattr(config_mod, "get_settings", lambda: s)
|
||
monkeypatch.setattr(context_mod, "get_settings", lambda: s)
|
||
|
||
ctx = _run(context_mod.init_context(db_path=str(tmp_path / "bioflow.db")))
|
||
try:
|
||
monkeypatch.setattr(r_flow, "context", ctx)
|
||
resp = _run(r_flow.flow_ask(FlowAskRequest(
|
||
session_id="s1", question="identify 需要什么输入")))
|
||
assert resp["ok"] is True
|
||
citations = resp["answer"]["citations"]
|
||
assert citations, "接线后 ask 应返回引用块（BUG-R1-03）"
|
||
assert citations[0]["path"]
|
||
assert citations[0]["cite"]
|
||
assert resp["answer"]["degraded"] is False
|
||
finally:
|
||
_run(context_mod.shutdown_context())
|
||
from app.db import close_db
|
||
|
||
_run(close_db())
|
||
context_mod.context = context_mod.Context()
|
||
|
||
|
||
def test_init_flow_store_missing_dir_degrades(tmp_path, monkeypatch):
|
||
"""flow_docs 缺失 → init_flow_store 返回 None（降级不阻断）。"""
|
||
import app.config as config_mod
|
||
import app.context as context_mod
|
||
|
||
s = config_mod.Settings(
|
||
_env_file=None,
|
||
data_root=str(tmp_path / "data"),
|
||
jobs_dir=str(tmp_path / "jobs"),
|
||
db_path=str(tmp_path / "bioflow.db"),
|
||
archives_dir=str(tmp_path / "archives"),
|
||
image_map_path=str(tmp_path / "stage_image_map.json"),
|
||
flow_docs_dir=str(tmp_path / "no_such_flow_docs"),
|
||
)
|
||
monkeypatch.setattr(config_mod, "get_settings", lambda: s)
|
||
monkeypatch.setattr(context_mod, "get_settings", lambda: s)
|
||
store = context_mod.init_flow_store(s.flow_docs_dir_abs)
|
||
# 目录缺失时 FlowStore 可能空载成功也可能抛错降级——两态均不得阻断
|
||
# （关键契约：init_flow_store 绝不抛异常上浮）
|
||
assert store is None or store.stats().get("total_chunks", 0) == 0
|