admin
0282cfeeeb
- 生信进度:后端新增 stage_progress 事件帧+进度解析白名单(rnaseq_qc/sample_qc), kernel 日志流式落盘(20行/1s flush+fsync),logs 分页端点(offset/limit/eof), 事件 seq 跨重启扫描续增(2026-09-17 断流修复);前端 FlowProgressCard 确定性/ 不确定两态进度条+点击定位到生信任务 TAB,BioJobsPane DAG 行展开滚动日志区 (3s 轮询),log_tail 4000 截断,重跑 stage_started 清零进度 - 图片注入:figures 三源重构(标准DOI图库主路径/遗留体系兜底/PMC兜底), [[FIG:]] 宽松归一化识别 PMC/PMID/DOI 三种形态,引文兜底扩展至 [PMCID:]/[DOI:] 且标记未满配额继续注入,注入预算 90s/10篇(175 实测校准), PMC graphic 无 content-type 宽松正则,空结果不缓存(176 冷连接三连败修复) - QA 启动自愈:start.sh 启动前等 Qdrant readyz(30min)+Neo4j TCP(10min); backend_guard 标签校验改子集判定 - 样本配对:同 inode 去重优先真实文件,R1/R2 归一配对一对一认领 - gitignore 新增 *.bak-*/*.before-*/bioflow_service/bioflow.db 测试:bioflow_service pytest 358/358,前端 bioflow vitest 205/205
906 lines
34 KiB
Python
906 lines
34 KiB
Python
"""tests/test_api.py —— 生信服务端点（L2；mock docker/LLM/vstore）。
|
||
|
||
分层标记 -m bioflow_api。覆盖（TC-BIO-API 系列）：
|
||
health / chat/stream SSE（plan_ready 候选/伪造阶段拒/非生信不建任务/降级）/
|
||
catalog / jobs 全套 / files（上传/路径登记/白名单）/ stages（confirm/rerun/skip/
|
||
pause/resume/manifest+日志尾）/ events SSE（快照+回放）/ archives（存档/列表/
|
||
复用/删除）/ 报告登记。
|
||
|
||
传输层说明：本环境 fastapi TestClient 的 sync stream 包装与 httpx ASGITransport
|
||
对永不结束的 SSE 流均存在首帧不达/缓冲问题（dbg3/dbg6 最小复现实证）——
|
||
SSE 测试端点直调迭代 body_iterator；普通 JSON 请求走 ASGITransport
|
||
（每请求独立 AsyncClient + 新 loop，无跨用例状态纠缠）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import time
|
||
|
||
import pytest
|
||
|
||
pytestmark = pytest.mark.bioflow_api
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 传输辅助
|
||
# ---------------------------------------------------------------------------
|
||
# 架构说明：每个用例一个 ApiSession——独立事件循环 + 持久 AsyncClient（ASGITransport）。
|
||
# asyncio.set_event_loop 保证端点内 asyncio.create_task 派生的后台任务（runner 自动
|
||
# 推进）绑定到本 loop；_wait_until 轮询经同一 client 泵 loop，后台任务随之推进。
|
||
# 规避 TestClient sync stream 与 ASGITransport 对永不结束 SSE 流的缓冲问题
|
||
# （dbg3/dbg6 最小复现实证），SSE 用例端点直调迭代 body_iterator（见模块 docstring）。
|
||
|
||
|
||
def _run_async(coro):
|
||
"""在共享会话 loop 上跑协程（anyio portal 绑定首个 loop，跨 loop 会死锁）。"""
|
||
return ApiSession.shared().loop.run_until_complete(coro)
|
||
|
||
|
||
class ApiSession:
|
||
"""模块级单例会话：全模块共享一个 event loop 与 AsyncClient。
|
||
|
||
原因：anyio asyncio 后端缓存线程首个 event loop（portal），跨测试新建
|
||
loop 会导致 run_until_complete 死锁（实证：test_create_job 后续测试随机
|
||
hang）。单 loop 共享 + 每用例后 drain 后台任务，是确定性方案。
|
||
"""
|
||
|
||
_instance = None
|
||
|
||
def __init__(self) -> None:
|
||
import httpx
|
||
|
||
from app.main import app as _app
|
||
|
||
self.loop = asyncio.new_event_loop()
|
||
asyncio.set_event_loop(self.loop)
|
||
transport = httpx.ASGITransport(app=_app)
|
||
self._client = httpx.AsyncClient(transport=transport, base_url="http://t")
|
||
|
||
@classmethod
|
||
def shared(cls) -> "ApiSession":
|
||
if cls._instance is None:
|
||
cls._instance = cls()
|
||
asyncio.set_event_loop(cls._instance.loop)
|
||
return cls._instance
|
||
|
||
def request(self, method: str, url: str, **kw) -> "_Resp":
|
||
resp = self.loop.run_until_complete(self._client.request(method, url, **kw))
|
||
return _Resp(resp.status_code, resp.content, dict(resp.headers))
|
||
|
||
def get(self, url, **kw):
|
||
return self.request("GET", url, **kw)
|
||
|
||
def post(self, url, **kw):
|
||
return self.request("POST", url, **kw)
|
||
|
||
def delete(self, url, **kw):
|
||
return self.request("DELETE", url, **kw)
|
||
|
||
def drain(self) -> None:
|
||
"""取消全部遗留后台任务（上用例 runner 派生），泵干 loop。"""
|
||
|
||
import app.context as context_mod
|
||
|
||
async def _cancel():
|
||
ctx = context_mod.context
|
||
targets = list(ctx._tasks)
|
||
if ctx.runner is not None:
|
||
targets += list(ctx.runner._tasks)
|
||
for t in targets:
|
||
t.cancel()
|
||
await asyncio.sleep(0.05) # 给取消任务 finally（信号量释放）时间
|
||
return True
|
||
|
||
try:
|
||
self.loop.run_until_complete(asyncio.wait_for(_cancel(), timeout=3))
|
||
except (asyncio.TimeoutError, RuntimeError):
|
||
pass
|
||
|
||
def close(self) -> None:
|
||
self.drain()
|
||
self.loop.run_until_complete(self._client.aclose())
|
||
self.loop.close()
|
||
asyncio.set_event_loop(None)
|
||
ApiSession._instance = None
|
||
|
||
|
||
class _Resp:
|
||
"""httpx 响应最小封装（status_code/json/text）。"""
|
||
|
||
def __init__(self, status_code: int, content: bytes, headers=None) -> None:
|
||
self.status_code = status_code
|
||
self.content = content
|
||
self.headers = headers or {}
|
||
|
||
def json(self):
|
||
return json.loads(self.content.decode("utf-8"))
|
||
|
||
|
||
@pytest.fixture()
|
||
def api(api_context):
|
||
"""API 会话（模块级共享 loop + client；用例后取消遗留后台任务）。"""
|
||
session = ApiSession.shared()
|
||
yield session
|
||
session.drain()
|
||
|
||
|
||
def _wait_until(fn, timeout=5.0, interval=0.02):
|
||
deadline = time.monotonic() + timeout
|
||
while time.monotonic() < deadline:
|
||
value = fn()
|
||
if value:
|
||
return value
|
||
time.sleep(interval)
|
||
return fn()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# SSE 辅助（端点直调；传输层限制见模块 docstring）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _parse_sse_frame(frame_text: str) -> dict | None:
|
||
"""SSE 帧解析（event/data 两行一组；注释帧返回 None）。"""
|
||
lines = frame_text.split("\n")
|
||
event_type = None
|
||
data = None
|
||
for line in lines:
|
||
if line.startswith("event: "):
|
||
event_type = line[7:]
|
||
elif line.startswith("data: "):
|
||
data = line[6:]
|
||
if event_type and data:
|
||
try:
|
||
return {"type": event_type, "data": json.loads(data)}
|
||
except ValueError:
|
||
return {"type": event_type, "data": data}
|
||
if event_type:
|
||
return {"type": event_type, "data": None}
|
||
return None
|
||
|
||
|
||
async def _collect_endpoint_stream(resp, max_events: int = 10, deadline_sec: float = 5.0):
|
||
"""迭代端点返回的 StreamingResponse.body_iterator（限时收集）。"""
|
||
events: list[dict] = []
|
||
buf_holder = [""]
|
||
|
||
async def _iter():
|
||
async for frame in resp.body_iterator:
|
||
chunk = frame if isinstance(frame, (bytes, bytearray)) else str(frame).encode("utf-8")
|
||
buf_holder[0] += chunk.decode("utf-8")
|
||
while "\n\n" in buf_holder[0]:
|
||
frame_text, buf_holder[0] = buf_holder[0].split("\n\n", 1)
|
||
parsed = _parse_sse_frame(frame_text)
|
||
if parsed:
|
||
events.append(parsed)
|
||
if len(events) >= max_events:
|
||
return events
|
||
return events
|
||
|
||
try:
|
||
return await asyncio.wait_for(_iter(), timeout=deadline_sec)
|
||
except asyncio.TimeoutError:
|
||
return events
|
||
|
||
|
||
def _read_sse(endpoint_coro, max_events=10, deadline_sec=5.0):
|
||
"""端点直调版 SSE 读取。"""
|
||
|
||
async def go():
|
||
resp = await endpoint_coro
|
||
assert resp.status_code == 200
|
||
assert resp.media_type == "text/event-stream"
|
||
return await _collect_endpoint_stream(resp, max_events, deadline_sec)
|
||
|
||
return _run_async(go())
|
||
|
||
|
||
def _read_sse_status(endpoint_coro):
|
||
"""端点直调：只取状态码（404 等）。"""
|
||
|
||
async def go():
|
||
from fastapi import HTTPException
|
||
|
||
try:
|
||
resp = await endpoint_coro
|
||
return resp.status_code
|
||
except HTTPException as exc:
|
||
return exc.status_code
|
||
|
||
return _run_async(go())
|
||
|
||
|
||
def _chat_stream_call(message, **kw):
|
||
from app.routers import bioflow as r_flow
|
||
from app.schemas import ChatStreamRequest
|
||
|
||
body = ChatStreamRequest(session_id=kw.get("session_id", "s1"), message=message, **{
|
||
k: v for k, v in kw.items() if k != "session_id"})
|
||
return r_flow.chat_stream(body)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# health
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_health(api):
|
||
"""TC-BIO-API-099：health 200 + db 连通状态。"""
|
||
resp = api.get("/api/bioflow/health")
|
||
assert resp.status_code == 200
|
||
body = resp.json()
|
||
assert body["ok"] is True
|
||
assert body["status"] == "ok"
|
||
assert body["db"]["connected"] is True
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# catalog
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_catalog_all(api):
|
||
"""TC-BIO-API-021：catalog 返回全部阶段（映射表驱动，每项含核心字段）。"""
|
||
resp = api.get("/api/bioflow/catalog")
|
||
assert resp.status_code == 200
|
||
stages = resp.json()["data"]["stages"]
|
||
assert len(stages) >= 7 # 测试映射表 7 阶段（真实表 75）
|
||
item = [s for s in stages if s["name"] == "identify"][0]
|
||
assert {"name", "description", "image", "params", "inputs", "outputs"} <= set(item.keys())
|
||
|
||
|
||
def test_catalog_chain(api):
|
||
"""TC-BIO-API-022：基因家族链。"""
|
||
resp = api.get("/api/bioflow/catalog", params={"chain": "gene_family"})
|
||
assert resp.status_code == 200
|
||
stages = resp.json()["data"]["stages"]
|
||
assert [s["name"] for s in stages][:2] == ["genome_prepare", "identify"]
|
||
|
||
|
||
def test_catalog_stage_detail(api):
|
||
"""TC-BIO-API-023：单阶段参数 schema。"""
|
||
resp = api.get("/api/bioflow/catalog", params={"stage": "identify"})
|
||
assert resp.status_code == 200
|
||
stage = resp.json()["data"]["stage"]
|
||
for p in stage["params"]:
|
||
assert {"name", "type", "default", "source", "help"} <= set(p.keys())
|
||
assert stage["image"] == "bioflow-gf"
|
||
assert stage["description"] == "基因家族鉴定"
|
||
|
||
|
||
def test_catalog_unknown_stage(api):
|
||
"""TC-BIO-API-024：未知阶段 404。"""
|
||
resp = api.get("/api/bioflow/catalog", params={"stage": "nope"})
|
||
assert resp.status_code == 404
|
||
assert resp.json()["error"]["code"] == "NOT_FOUND"
|
||
|
||
|
||
def test_catalog_unknown_chain(api):
|
||
resp = api.get("/api/bioflow/catalog", params={"chain": "nope"})
|
||
assert resp.status_code == 404
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# chat/stream
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_chat_stream_plan_ready(api_context):
|
||
"""TC-BIO-API-001/002：chat/stream SSE，plan_ready 下发候选。"""
|
||
events = _read_sse(_chat_stream_call("帮我跑水稻 SPL 家族鉴定"), max_events=2)
|
||
types = [e["type"] for e in events]
|
||
assert "plan_ready" in types
|
||
plan = [e for e in events if e["type"] == "plan_ready"][0]["data"]
|
||
assert 1 <= len(plan["candidates"]) <= 3
|
||
for c in plan["candidates"]:
|
||
assert c["stages"] and c["purpose"]
|
||
|
||
|
||
def test_chat_stream_unknown_stage_rejected(api_context, monkeypatch):
|
||
"""TC-BIO-API-003：伪造阶段被拒（候选不进候选集）。"""
|
||
from conftest import FakeLLM
|
||
import app.services.bioflow.planner as planner_mod
|
||
monkeypatch.setattr(planner_mod, "complete_json", FakeLLM(candidates=[
|
||
{"purpose": "坏", "stages": ["fake_stage_xyz"]},
|
||
{"purpose": "好", "stages": ["identify"]},
|
||
]))
|
||
events = _read_sse(_chat_stream_call("帮我跑水稻 SPL 家族鉴定"), max_events=2)
|
||
plan = [e for e in events if e["type"] == "plan_ready"][0]["data"]
|
||
for c in plan["candidates"]:
|
||
assert "fake_stage_xyz" not in c["stages"]
|
||
|
||
|
||
def test_chat_stream_non_bioflow_message(api_context):
|
||
"""TC-BIO-API-005：非生信消息不建任务。"""
|
||
events = _read_sse(_chat_stream_call("什么是基因家族"), max_events=1)
|
||
types = [e["type"] for e in events]
|
||
assert "message_done" in types
|
||
assert "job_created" not in types
|
||
done = [e for e in events if e["type"] == "message_done"][0]["data"]
|
||
assert done["kind"] == "qa"
|
||
|
||
|
||
def test_chat_stream_empty_message_422(api):
|
||
"""TC-BIO-API-006/007/008：422 校验。"""
|
||
assert api.post("/api/bioflow/chat/stream",
|
||
json={"session_id": "s1", "message": ""}).status_code == 422
|
||
assert api.post("/api/bioflow/chat/stream",
|
||
json={"session_id": "s1"}).status_code == 422
|
||
assert api.post("/api/bioflow/chat/stream",
|
||
json={"message": "x"}).status_code == 422
|
||
|
||
|
||
def test_chat_stream_overlong_rejected(api):
|
||
"""TC-BIO-API-015：超长消息拒绝，不触发 LLM。"""
|
||
resp = api.post("/api/bioflow/chat/stream",
|
||
json={"session_id": "s1", "message": "x" * 100001})
|
||
assert resp.status_code == 422
|
||
|
||
|
||
def test_chat_stream_llm_timeout_degrade(api_context, monkeypatch):
|
||
"""TC-BIO-API-009：LLM 超时降级（内置兜底候选不中断 SSE）。"""
|
||
from conftest import FakeLLM
|
||
import app.services.bioflow.planner as planner_mod
|
||
monkeypatch.setattr(planner_mod, "complete_json", FakeLLM(error="timeout"))
|
||
events = _read_sse(_chat_stream_call("帮我跑水稻 SPL 家族鉴定"), max_events=2)
|
||
plan = [e for e in events if e["type"] == "plan_ready"][0]["data"]
|
||
assert plan["fallback"] # 降级提示
|
||
assert plan["candidates"]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# jobs
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _create_job(api, stages=("identify",), auto_continue=False, **kw):
|
||
resp = api.post("/api/bioflow/jobs", json={
|
||
"session_id": kw.get("session_id", "s1"),
|
||
"title": kw.get("title", "t"),
|
||
"stages": list(stages),
|
||
"auto_continue": auto_continue,
|
||
})
|
||
return resp
|
||
|
||
|
||
def _job_id(resp):
|
||
return resp.json()["data"]["id"]
|
||
|
||
|
||
def _job_stage_status(api, job_id: str, idx: int = 0):
|
||
return api.get(f"/api/bioflow/jobs/{job_id}").json()["data"]["stages"][idx]["status"]
|
||
|
||
|
||
def test_create_job(api):
|
||
"""TC-BIO-API-027：POST jobs 创建任务（stages 校验 + 返回结构）。"""
|
||
resp = _create_job(api, title="SPL 鉴定")
|
||
assert resp.status_code == 201
|
||
data = resp.json()["data"]
|
||
assert data["id"]
|
||
assert data["title"] == "SPL 鉴定"
|
||
assert data["stages"][0]["stage"] == "identify"
|
||
assert data["status"] in ("created", "running", "done")
|
||
|
||
|
||
def test_create_job_missing_stages(api):
|
||
"""TC-BIO-API-028：缺 stages 422。"""
|
||
assert api.post("/api/bioflow/jobs",
|
||
json={"session_id": "s1", "title": "t"}).status_code == 422
|
||
|
||
|
||
def test_create_job_unknown_stage(api):
|
||
"""TC-BIO-API-030：未知 stage 拒绝。"""
|
||
resp = api.post("/api/bioflow/jobs", json={
|
||
"session_id": "s1", "title": "t", "stages": ["ghost_stage"],
|
||
})
|
||
assert resp.status_code == 422
|
||
assert "ghost_stage" in json.dumps(resp.json(), ensure_ascii=False)
|
||
|
||
|
||
def test_jobs_list_and_filter(api):
|
||
"""TC-BIO-API-032：列表按创建顺序倒序（rowid DESC）。"""
|
||
_create_job(api, title="t1")
|
||
_create_job(api, title="t2")
|
||
resp = api.get("/api/bioflow/jobs")
|
||
assert resp.status_code == 200
|
||
items = resp.json()["data"]["items"]
|
||
assert len(items) >= 2
|
||
assert items[0]["title"] == "t2" # 倒序
|
||
|
||
|
||
def test_job_detail_and_404(api):
|
||
resp = _create_job(api, stages=["identify", "phylogeny"])
|
||
job_id = _job_id(resp)
|
||
detail = api.get(f"/api/bioflow/jobs/{job_id}")
|
||
assert detail.status_code == 200
|
||
assert len(detail.json()["data"]["stages"]) == 2
|
||
assert "progress" in detail.json()["data"]
|
||
not_found = api.get("/api/bioflow/jobs/not-exist")
|
||
assert not_found.status_code == 404
|
||
assert not_found.json()["error"]["code"] == "NOT_FOUND"
|
||
|
||
|
||
def test_delete_job(api):
|
||
resp = _create_job(api)
|
||
job_id = _job_id(resp)
|
||
# 等待任务终态（done；删除只拒 running）
|
||
def terminal():
|
||
return api.get(f"/api/bioflow/jobs/{job_id}").json()["data"]["status"] in ("done", "failed")
|
||
_wait_until(terminal, timeout=5)
|
||
del_resp = api.delete(f"/api/bioflow/jobs/{job_id}")
|
||
assert del_resp.status_code == 204
|
||
assert api.get(f"/api/bioflow/jobs/{job_id}").status_code == 404
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# files
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_files_upload_and_list(api):
|
||
"""TC-BIO-API-039/048：上传 + 文件列表。"""
|
||
resp = _create_job(api)
|
||
job_id = _job_id(resp)
|
||
content = b">gene1\nACGTACGT\n"
|
||
up = api.post(f"/api/bioflow/jobs/{job_id}/files",
|
||
files={"file": ("test.fa", content, "application/octet-stream")})
|
||
assert up.status_code == 200
|
||
data = up.json()["data"]
|
||
assert data["name"] == "test.fa"
|
||
assert data["size"] == len(content)
|
||
listing = api.get(f"/api/bioflow/jobs/{job_id}/files")
|
||
assert listing.status_code == 200
|
||
assert listing.json()["data"]["items"][0]["name"] == "test.fa"
|
||
|
||
|
||
def test_upload_resumes_await_inputs_job(api, api_context, monkeypatch):
|
||
"""BUG-P7-01 端点全链路：await_files 任务上传文件后经端点自动续跑到 done。
|
||
|
||
复现路径（修复前）：POST /jobs（auto_continue=true）→ await_files 停
|
||
await_inputs → 上传成功落盘但无重推（任务卡死）。修复后 upload 成功分支
|
||
调 retry_after_files → _resume_loop 重推 → auto_continue 直跑至 done。
|
||
"""
|
||
from app.context import Context
|
||
from app.services.bioflow.events import EventBus
|
||
from app.services.bioflow.catalog import build_catalog
|
||
from conftest import make_runner
|
||
import app.routers.biojobs as r_jobs
|
||
import app.routers.bioflow as r_flow
|
||
import app.routers.bioarchives as r_arch
|
||
import app.context as context_mod
|
||
|
||
# 专用 runner：开输入检查（api_context 默认关闭）
|
||
events = EventBus(api_context.db.path.parent / "events")
|
||
catalog = build_catalog(api_context.catalog._map)
|
||
runner = make_runner(api_context.db, events, api_context.executor, catalog,
|
||
check_inputs=True)
|
||
ctx2 = Context(db=api_context.db, events=events, kernel=None, catalog=catalog,
|
||
runner=runner, semaphore=runner.semaphore,
|
||
planner=api_context.planner, flow_store=api_context.flow_store,
|
||
executor=api_context.executor)
|
||
monkeypatch.setattr(context_mod, "context", ctx2)
|
||
monkeypatch.setattr(r_jobs, "context", ctx2)
|
||
monkeypatch.setattr(r_flow, "context", ctx2)
|
||
monkeypatch.setattr(r_arch, "context", ctx2)
|
||
|
||
resp = _create_job(api, auto_continue=True)
|
||
job_id = _job_id(resp)
|
||
|
||
def awaiting():
|
||
d = api.get(f"/api/bioflow/jobs/{job_id}").json()["data"]
|
||
return d["status"] in ("await_inputs", "running", "done")
|
||
_wait_until(awaiting, timeout=5)
|
||
# 到达 await_inputs（input 空）
|
||
def at_await():
|
||
d = api.get(f"/api/bioflow/jobs/{job_id}").json()["data"]
|
||
return d["status"] == "await_inputs" and d["stages"][0]["status"] == "await_files"
|
||
assert _wait_until(at_await, timeout=5), "任务应停在 await_inputs（input 空）"
|
||
assert api_context.executor.calls == []
|
||
|
||
# 上传文件（触发 retry_after_files）
|
||
content = b">gene1\nACGTACGT\n"
|
||
up = api.post(f"/api/bioflow/jobs/{job_id}/files",
|
||
files={"file": ("genome.pep.fa", content, "application/octet-stream")})
|
||
assert up.status_code == 200
|
||
|
||
# 自动续跑到 done
|
||
def done():
|
||
d = api.get(f"/api/bioflow/jobs/{job_id}").json()["data"]
|
||
return d["status"] == "done" and d["stages"][0]["status"] == "done"
|
||
assert _wait_until(done, timeout=5), "上传后任务应自动续跑到 done（BUG-P7-01）"
|
||
assert api_context.executor.calls # 执行记录非空
|
||
|
||
|
||
def test_files_upload_bad_extension(api):
|
||
"""TC-BIO-API-046：扩展名白名单外 422。"""
|
||
resp = _create_job(api)
|
||
job_id = _job_id(resp)
|
||
up = api.post(f"/api/bioflow/jobs/{job_id}/files",
|
||
files={"file": ("x.exe", b"MZ", "application/octet-stream")})
|
||
assert up.status_code == 422
|
||
|
||
|
||
def test_files_upload_no_job(api):
|
||
"""TC-BIO-API-120：上传无任务 404。"""
|
||
up = api.post("/api/bioflow/jobs/nope/files",
|
||
files={"file": ("a.fa", b"x", "text/plain")})
|
||
assert up.status_code == 404
|
||
|
||
|
||
def test_path_register_valid_and_whitelist(api, tmp_path):
|
||
"""TC-BIO-API-042/045：路径登记合法 + 敏感黑名单 403。"""
|
||
resp = _create_job(api)
|
||
job_id = _job_id(resp)
|
||
# api_context 的 settings 已指向 tmp_path/data（conftest _isolated_env 派生）
|
||
from app.config import get_settings as _gs
|
||
data_root = _gs().data_root_abs
|
||
import os
|
||
allowed = os.path.join(data_root, "examples", "SPL.fa")
|
||
os.makedirs(os.path.dirname(allowed), exist_ok=True)
|
||
with open(allowed, "w") as f:
|
||
f.write(">x\nACGT")
|
||
ok = api.post(f"/api/bioflow/jobs/{job_id}/files", json={"server_path": allowed})
|
||
assert ok.status_code == 200
|
||
body = ok.json()["data"]
|
||
assert body["items"][0]["source"] == "server_path"
|
||
assert body["items"][0]["name"] == "SPL.fa"
|
||
# 敏感黑名单（.env 文件名形态）
|
||
outside = str(tmp_path / ".env")
|
||
with open(outside, "w") as f:
|
||
f.write("SECRET=1")
|
||
forbidden = api.post(f"/api/bioflow/jobs/{job_id}/files", json={"server_path": outside})
|
||
assert forbidden.status_code == 403
|
||
assert forbidden.json()["error"]["code"] == "PATH_NOT_ALLOWED"
|
||
# 不存在
|
||
missing = api.post(f"/api/bioflow/jobs/{job_id}/files",
|
||
json={"server_path": str(tmp_path / "nope.fa")})
|
||
assert missing.status_code == 400
|
||
assert missing.json()["error"]["code"] == "PATH_NOT_EXIST"
|
||
|
||
|
||
def test_path_register_directory_expand(api, tmp_path):
|
||
"""目录登记：递归展开多文件，逐个 items；挂载清单落 server_registered.json。"""
|
||
resp = _create_job(api)
|
||
job_id = _job_id(resp)
|
||
d = tmp_path / "rnaseq_fastq"
|
||
(d / "s1").mkdir(parents=True)
|
||
(d / "s2").mkdir(parents=True)
|
||
(d / "s1" / "CRR1_1.fastq.gz").write_text("g1", encoding="utf-8")
|
||
(d / "s1" / "CRR1_2.fastq.gz").write_text("g2", encoding="utf-8")
|
||
(d / "s2" / "CRR2_1.fastq.gz").write_text("g3", encoding="utf-8")
|
||
(d / "notes.txt").write_text("n", encoding="utf-8") # 允许扩展名内
|
||
(d / "skip.exe").write_text("x", encoding="utf-8") # 扩展名白名单外,跳过
|
||
ok = api.post(f"/api/bioflow/jobs/{job_id}/files", json={"server_path": str(d)})
|
||
assert ok.status_code == 200
|
||
body = ok.json()["data"]
|
||
names = {it["name"] for it in body["items"]}
|
||
assert names == {"CRR1_1.fastq.gz", "CRR1_2.fastq.gz", "CRR2_1.fastq.gz", "notes.txt"}
|
||
assert body["total"] == 4
|
||
# 挂载清单落盘
|
||
import json as _json
|
||
from pathlib import Path as _Path
|
||
from app.config import get_settings as _gs
|
||
job_dir = _Path(_gs().jobs_dir_abs)
|
||
reg = _json.loads((job_dir / job_id / "server_registered.json").read_text(encoding="utf-8"))
|
||
assert len(reg["registered_server_paths"]) >= 4 # 4 文件 + samples.yaml 引用的父目录
|
||
# 空目录报错
|
||
empty = tmp_path / "empty_dir"
|
||
empty.mkdir()
|
||
err = api.post(f"/api/bioflow/jobs/{job_id}/files", json={"server_path": str(empty)})
|
||
assert err.status_code == 400
|
||
assert err.json()["error"]["code"] == "VALIDATION"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# stages 控制 + manifest
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_confirm_flow(api):
|
||
"""TC-BIO-API-056/057：confirm 发起执行，参数覆盖。"""
|
||
resp = _create_job(api)
|
||
job_id = _job_id(resp)
|
||
|
||
def awaiting():
|
||
return _job_stage_status(api, job_id) in ("waiting_confirm", "done")
|
||
_wait_until(awaiting, timeout=5)
|
||
|
||
confirm = api.post(f"/api/bioflow/jobs/{job_id}/stages/identify/confirm",
|
||
json={"params": {"cpu": "8"}})
|
||
assert confirm.status_code == 200
|
||
assert confirm.json()["data"]["status"] in ("running", "done")
|
||
|
||
def done():
|
||
return _job_stage_status(api, job_id) == "done"
|
||
_wait_until(done, timeout=5)
|
||
assert _job_stage_status(api, job_id) == "done"
|
||
|
||
|
||
def test_confirm_unknown_param(api):
|
||
"""TC-BIO-API-058：拒绝非 schema 参数。"""
|
||
resp = _create_job(api)
|
||
job_id = _job_id(resp)
|
||
|
||
def waiting():
|
||
return _job_stage_status(api, job_id) in ("waiting_confirm", "done")
|
||
_wait_until(waiting, timeout=5)
|
||
confirm = api.post(f"/api/bioflow/jobs/{job_id}/stages/identify/confirm",
|
||
json={"params": {"magic": "x"}})
|
||
# waiting_confirm 时 422；已 done 时 409（两态均可——状态竞争）
|
||
assert confirm.status_code in (409, 422)
|
||
|
||
|
||
def test_confirm_unknown_stage_404(api):
|
||
resp = _create_job(api)
|
||
job_id = _job_id(resp)
|
||
confirm = api.post(f"/api/bioflow/jobs/{job_id}/stages/nope/confirm", json={})
|
||
assert confirm.status_code == 404
|
||
|
||
|
||
def test_manifest_with_log_tail(api):
|
||
"""TC-BIO-API-071：manifest + 日志尾 80 行。"""
|
||
resp = _create_job(api, auto_continue=True)
|
||
job_id = _job_id(resp)
|
||
|
||
def done():
|
||
return _job_stage_status(api, job_id) == "done"
|
||
_wait_until(done, timeout=5)
|
||
m = api.get(f"/api/bioflow/jobs/{job_id}/stages/identify/manifest")
|
||
assert m.status_code == 200
|
||
data = m.json()["data"]
|
||
assert data["manifest"]["status"] == "success"
|
||
assert "log_tail" in data
|
||
|
||
|
||
def test_manifest_unknown_stage(api):
|
||
resp = _create_job(api)
|
||
job_id = _job_id(resp)
|
||
m = api.get(f"/api/bioflow/jobs/{job_id}/stages/nope/manifest")
|
||
assert m.status_code == 404
|
||
|
||
|
||
def test_stage_logs_pagination(api, api_context):
|
||
"""TC-BIO-API-102：logs 分页端点(offset/limit 切片、eof、log_tail、超尾)。"""
|
||
api_context.executor.log = "line0\nline1\nline2\nline3\n"
|
||
resp = _create_job(api, auto_continue=True)
|
||
job_id = _job_id(resp)
|
||
|
||
def done():
|
||
return _job_stage_status(api, job_id) == "done"
|
||
_wait_until(done, timeout=5)
|
||
|
||
r1 = api.get(f"/api/bioflow/jobs/{job_id}/stages/identify/logs?offset=0&limit=1")
|
||
assert r1.status_code == 200
|
||
d1 = r1.json()["data"]
|
||
assert d1["lines"] == ["line0"]
|
||
assert d1["eof"] is False
|
||
# 增量衔接:offset=1&limit=2
|
||
r2 = api.get(f"/api/bioflow/jobs/{job_id}/stages/identify/logs?offset=1&limit=2")
|
||
d2 = r2.json()["data"]
|
||
assert d2["lines"] == ["line1", "line2"]
|
||
assert d2["eof"] is False
|
||
# 到尾:offset=2&limit=500 → eof:true
|
||
r3 = api.get(f"/api/bioflow/jobs/{job_id}/stages/identify/logs?offset=2&limit=500")
|
||
d3 = r3.json()["data"]
|
||
assert d3["lines"] == ["line2", "line3"]
|
||
assert d3["eof"] is True
|
||
# 超尾:offset=1000 → 空 lines + eof:true + log_tail 仍在
|
||
r4 = api.get(f"/api/bioflow/jobs/{job_id}/stages/identify/logs?offset=1000&limit=500")
|
||
d4 = r4.json()["data"]
|
||
assert d4["lines"] == []
|
||
assert d4["eof"] is True
|
||
assert d4["log_tail"]
|
||
|
||
|
||
def test_stage_logs_no_file_and_404(api):
|
||
"""TC-BIO-API-103：log 文件不存在 → 空 lines + eof;未知 job/阶段 404。"""
|
||
resp = _create_job(api) # 手动模式:identify 停在 waiting_confirm,从未执行
|
||
job_id = _job_id(resp)
|
||
|
||
def waiting():
|
||
return _job_stage_status(api, job_id) == "waiting_confirm"
|
||
_wait_until(waiting, timeout=5)
|
||
|
||
r = api.get(f"/api/bioflow/jobs/{job_id}/stages/identify/logs")
|
||
assert r.status_code == 200
|
||
d = r.json()["data"]
|
||
assert d["lines"] == [] and d["eof"] is True and d["total_bytes"] == 0
|
||
# 未知 job 404
|
||
assert api.get("/api/bioflow/jobs/nopejob/stages/identify/logs").status_code == 404
|
||
# 未知阶段 404
|
||
assert api.get(f"/api/bioflow/jobs/{job_id}/stages/nope/logs").status_code == 404
|
||
|
||
|
||
def test_rerun_and_skip_on_failed(api, api_context):
|
||
"""TC-BIO-API-062/065：rerun/skip 失败阶段。"""
|
||
api_context.executor.exit_code = 1
|
||
resp = _create_job(api, auto_continue=True)
|
||
job_id = _job_id(resp)
|
||
|
||
def failed():
|
||
return _job_stage_status(api, job_id) == "failed"
|
||
_wait_until(failed, timeout=5)
|
||
# skip
|
||
skip = api.post(f"/api/bioflow/jobs/{job_id}/stages/identify/skip")
|
||
assert skip.status_code == 200
|
||
assert skip.json()["data"]["status"] == "skipped"
|
||
|
||
|
||
def test_rerun_non_failed_409(api):
|
||
"""TC-BIO-API-064：done 阶段 rerun 409。"""
|
||
resp = _create_job(api, auto_continue=True)
|
||
job_id = _job_id(resp)
|
||
|
||
def done():
|
||
return _job_stage_status(api, job_id) == "done"
|
||
_wait_until(done, timeout=5)
|
||
rerun = api.post(f"/api/bioflow/jobs/{job_id}/stages/identify/rerun", json={})
|
||
assert rerun.status_code == 409
|
||
assert rerun.json()["error"]["code"] == "STAGE_NOT_FAILED"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# events SSE（快照 + 回放）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_events_snapshot_and_replay(api):
|
||
"""TC-BIO-API-017/078：events 首帧快照 + 断线重连回放。"""
|
||
resp = _create_job(api, auto_continue=True)
|
||
job_id = _job_id(resp)
|
||
|
||
def done():
|
||
return _job_stage_status(api, job_id) == "done"
|
||
_wait_until(done, timeout=5)
|
||
|
||
from app.routers import biojobs as r_jobs
|
||
|
||
events = _read_sse(r_jobs.job_events(job_id, stage=None, last_seq=-1), max_events=9)
|
||
frames = [e["type"] for e in events]
|
||
# 快照首帧
|
||
assert frames[0] == "snapshot"
|
||
# 历史事件回放含 job_created/stage_started/stage_done
|
||
assert "job_created" in frames
|
||
assert "stage_started" in frames
|
||
assert "stage_done" in frames
|
||
|
||
|
||
def test_events_unknown_job_404(api):
|
||
"""TC-BIO-API-126：events 未知 job 404。"""
|
||
from app.routers import biojobs as r_jobs
|
||
|
||
status = _read_sse_status(r_jobs.job_events("nope", stage=None, last_seq=-1))
|
||
assert status == 404
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# archives + 报告登记
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _make_archive(api):
|
||
"""创建已 done 任务并存档（返回 (job_id, archive_id)；测试辅助）。"""
|
||
resp = _create_job(api, auto_continue=True)
|
||
job_id = _job_id(resp)
|
||
|
||
def done():
|
||
return _job_stage_status(api, job_id) == "done"
|
||
_wait_until(done, timeout=5)
|
||
arch = api.post("/api/bioflow/archives", json={
|
||
"job_id": job_id, "name": "SPL 水稻鉴定", "description": "d",
|
||
"include_intermediate": True,
|
||
})
|
||
assert arch.status_code == 201
|
||
return job_id, arch.json()["data"]["id"]
|
||
|
||
|
||
def test_archive_create_and_list(api):
|
||
"""TC-BIO-API-087/090：存档（name/disk_path/size）+ 列表。"""
|
||
resp = _create_job(api, auto_continue=True)
|
||
job_id = _job_id(resp)
|
||
|
||
def done():
|
||
return _job_stage_status(api, job_id) == "done"
|
||
_wait_until(done, timeout=5)
|
||
arch = api.post("/api/bioflow/archives", json={
|
||
"job_id": job_id, "name": "SPL 水稻鉴定", "description": "d",
|
||
"include_intermediate": True,
|
||
})
|
||
assert arch.status_code == 201
|
||
data = arch.json()["data"]
|
||
assert data["name"] == "SPL 水稻鉴定"
|
||
assert data["disk_path"]
|
||
assert data["size_bytes"] > 0
|
||
listing = api.get("/api/bioflow/archives")
|
||
assert listing.status_code == 200
|
||
assert listing.json()["data"]["items"][0]["name"] == "SPL 水稻鉴定"
|
||
_last_archive = (job_id, data["id"])
|
||
return None
|
||
|
||
|
||
def test_archive_missing_name(api):
|
||
resp = _create_job(api)
|
||
job_id = _job_id(resp)
|
||
arch = api.post("/api/bioflow/archives", json={"job_id": job_id})
|
||
assert arch.status_code == 422
|
||
|
||
|
||
def test_archive_duplicate_name_409(api):
|
||
job_id, _aid = _make_archive(api)
|
||
arch = api.post("/api/bioflow/archives", json={
|
||
"job_id": job_id, "name": "SPL 水稻鉴定",
|
||
})
|
||
assert arch.status_code == 409
|
||
assert arch.json()["error"]["code"] == "ARCHIVE_EXISTS"
|
||
|
||
|
||
def test_archive_detail_delete(api):
|
||
_job_id_, aid = _make_archive(api)
|
||
detail = api.get(f"/api/bioflow/archives/{aid}")
|
||
assert detail.status_code == 200
|
||
assert detail.json()["data"]["stage_chain"] == ["identify"]
|
||
deleted = api.delete(f"/api/bioflow/archives/{aid}")
|
||
assert deleted.status_code == 204
|
||
assert api.get(f"/api/bioflow/archives/{aid}").status_code == 404
|
||
assert api.delete("/api/bioflow/archives/nope").status_code == 404
|
||
|
||
|
||
def test_archive_run_reuse(api):
|
||
"""TC-BIO-API-094：复用发起（新任务 stages=归档链）。"""
|
||
_job_id_, aid = _make_archive(api)
|
||
run = api.post(f"/api/bioflow/archives/{aid}/run", json={
|
||
"session_id": "s2", "target": "玉米 bZIP",
|
||
})
|
||
assert run.status_code == 201
|
||
data = run.json()["data"]
|
||
assert [s["stage"] for s in data["stages"]] == ["identify"]
|
||
assert "玉米 bZIP" in data["title"]
|
||
# 按名称匹配复用
|
||
by_name = api.post("/api/bioflow/archives/run", json={
|
||
"session_id": "s3", "name": "SPL 水稻鉴定",
|
||
})
|
||
assert by_name.status_code == 201
|
||
# 无匹配 404
|
||
missing = api.post("/api/bioflow/archives/run", json={
|
||
"session_id": "s3", "name": "不存在",
|
||
})
|
||
assert missing.status_code == 404
|
||
assert missing.json()["error"]["code"] == "ARCHIVE_NOT_FOUND"
|
||
|
||
|
||
def test_archive_report(api):
|
||
_job_id_, aid = _make_archive(api)
|
||
report = api.get(f"/api/bioflow/archives/{aid}/report")
|
||
# 该任务无报告（无报告链阶段）→ 404 或 200（有 html 时）
|
||
assert report.status_code in (200, 404)
|
||
|
||
|
||
def test_reports_register(api):
|
||
"""TC-BIO-API-104 服务侧：报告登记归一化 + 图清单回填。"""
|
||
resp = _create_job(api, auto_continue=True)
|
||
job_id = _job_id(resp)
|
||
|
||
def done():
|
||
return _job_stage_status(api, job_id) == "done"
|
||
_wait_until(done, timeout=5)
|
||
reg = api.post("/api/bioflow/reports/register", json={
|
||
"job_id": job_id, "session_id": "s1", "title": "SPL 报告",
|
||
"md_path": "/tmp/x.html", "figures": ["fig1"],
|
||
})
|
||
assert reg.status_code == 200
|
||
data = reg.json()["data"]
|
||
assert data["job_id"] == job_id
|
||
assert data["figures"] == ["fig1"]
|
||
assert data["registered_at"]
|
||
|
||
|
||
def test_report_revise_no_chart_target(api):
|
||
"""TC-BIO-API-144：无绘图目标 422。"""
|
||
resp = _create_job(api)
|
||
job_id = _job_id(resp)
|
||
rev = api.post(f"/api/bioflow/report/{job_id}/revise", json={
|
||
"instruction": "不明确",
|
||
})
|
||
assert rev.status_code == 422
|