adminandClaude Fable 5
d12071a4d7
- 前端零打包:归档文件(zip/tar/tar.gz/tgz)原样上传服务端解压; 目录拖入/选择器走 webkitRelativePath 多文件直传(filename=相对路径), 隐藏文件/__pycache__ 过滤 + 公共根剥离(参考 PlantAI-dev 主路径) - 后端 _read_archive 魔数分派(zip/tar.gz/tar)+保留嵌套路径; code_files 多文件字段;结构自适应分类(目录前缀→命令解析→扩展名, 未知归代码,纯数据 422 可读,纯脚本合法);store 嵌套落盘 - 存量 bug 修复:FormData 字段名 file→code_zip;golden File→JSON+test_file; http size hint 兼容 code_zip 与多文件求和 - 测试:后端 334/334(新增 10 个变体用例),前端 122/373 全绿 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
481 lines
18 KiB
Python
481 lines
18 KiB
Python
"""tests/test_extension_api.py —— 扩展端点（L2；契约 §6 全集）。
|
||
|
||
mock LLM（无 key → 规则版回退）/mock 执行器（注入 store.executor）。
|
||
SSE 用例端点直调迭代 body_iterator（同 test_api.py 纪律）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import io
|
||
import json
|
||
import zipfile
|
||
|
||
import pytest
|
||
|
||
pytestmark = pytest.mark.bioflow_api
|
||
|
||
from app.services.bioflow import extensions as ext_mod # noqa: E402
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 传输辅助（同 test_api.py ApiSession 模式）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class ApiSession:
|
||
_instance = None
|
||
|
||
def __init__(self):
|
||
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, url, **kw):
|
||
resp = self.loop.run_until_complete(self._client.request(method, url, **kw))
|
||
return resp
|
||
|
||
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)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# fixtures
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class FakeExtExecutor:
|
||
"""async 执行器：可编程退出码/输出；记录调用。"""
|
||
|
||
def __init__(self, exit_code: int = 0, log: str = "trial ok\n", outputs=("result.txt",)):
|
||
self.exit_code = exit_code
|
||
self.log = log
|
||
self.outputs = outputs
|
||
self.calls = []
|
||
|
||
async def execute(self, spec_in):
|
||
from pathlib import Path
|
||
|
||
self.calls.append(spec_in)
|
||
out_dir = Path(spec_in.output_dir)
|
||
out_dir.mkdir(parents=True, exist_ok=True)
|
||
for name in self.outputs:
|
||
(out_dir / name).write_text("data", encoding="utf-8")
|
||
if spec_in.log_path:
|
||
Path(spec_in.log_path).parent.mkdir(parents=True, exist_ok=True)
|
||
Path(spec_in.log_path).write_text(self.log, encoding="utf-8")
|
||
return ext_mod.ExtensionRunResult(
|
||
run_id="", exit_code=self.exit_code, log_tail=self.log,
|
||
outputs=list(self.outputs),
|
||
error="" if self.exit_code == 0 else "命令失败",
|
||
)
|
||
|
||
|
||
def _make_zip(entries: dict[str, bytes]) -> bytes:
|
||
buf = io.BytesIO()
|
||
with zipfile.ZipFile(buf, "w") as zf:
|
||
for name, blob in entries.items():
|
||
zf.writestr(name, blob)
|
||
return buf.getvalue()
|
||
|
||
|
||
@pytest.fixture()
|
||
def ext_api(test_settings, monkeypatch):
|
||
"""扩展 API 会话：settings 指 tmp、store 重置、无 LLM key。"""
|
||
monkeypatch.delenv("DOCGEN_LLM_API_KEY", raising=False)
|
||
monkeypatch.delenv("BIOFLOW_EXTENSION_API_KEY", raising=False)
|
||
ext_mod.reset_store()
|
||
session = ApiSession.shared()
|
||
yield session
|
||
ext_mod.reset_store()
|
||
|
||
|
||
def _store():
|
||
return ext_mod.get_store()
|
||
|
||
|
||
def _post_ext(session, name="测序质量统计", command="python quality_stats.py --fastq input.fastq --out results/",
|
||
code=None, test_blob=None, golden=None, runtime_image="test4py311"):
|
||
files = {"code_zip": ("code.zip", _make_zip(code or {"quality_stats.py": b"import sys\nprint('hi')\n"}), "application/zip")}
|
||
data = {"display_name": name, "command": command, "runtime_image": runtime_image}
|
||
if golden:
|
||
data["golden"] = json.dumps(golden)
|
||
if test_blob is not None:
|
||
files["test_file"] = ("input.fastq", test_blob, "application/octet-stream")
|
||
return session.post("/api/bioflow/extensions", files=files, data=data)
|
||
|
||
|
||
def _body(resp) -> dict:
|
||
try:
|
||
return json.loads(resp.content.decode("utf-8"))
|
||
except Exception: # noqa: BLE001
|
||
return {}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 创建 / 列表 / 详情 / 删除
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_create_multipart_zip(ext_api):
|
||
resp = _post_ext(ext_api, test_blob=b">s1\nACGT\n")
|
||
assert resp.status_code == 201
|
||
body = _body(resp)
|
||
assert body["ok"] is True
|
||
ext = body["data"]
|
||
assert ext["status"] == "draft"
|
||
assert ext["runtime_image"] == "bioflow-gf" # test4py311 映射
|
||
root = _store().ext_dir(ext["ext_id"])
|
||
assert (root / "raw_code" / "quality_stats.py").is_file()
|
||
assert (root / "raw_test_data" / "input.fastq").is_file()
|
||
assert (root / "converted_code" / "quality_stats.py").is_file()
|
||
|
||
|
||
def test_create_json_form(ext_api):
|
||
import base64
|
||
|
||
body = {
|
||
"display_name": "json 扩展",
|
||
"command": "python a.py --out results/",
|
||
"runtime_image": "bioflow-gf",
|
||
"code_files": [{"path": "a.py", "data": base64.b64encode(b"print(1)\n").decode()}],
|
||
}
|
||
resp = ext_api.post("/api/bioflow/extensions", json=body)
|
||
assert resp.status_code == 201
|
||
assert _body(resp)["data"]["display_name"] == "json 扩展"
|
||
|
||
|
||
def test_create_bad_zip(ext_api):
|
||
files = {"code_zip": ("bad.zip", b"not a zip", "application/zip")}
|
||
resp = ext_api.post(
|
||
"/api/bioflow/extensions",
|
||
files=files,
|
||
data={"display_name": "x", "command": "python a.py"},
|
||
)
|
||
assert resp.status_code == 422
|
||
assert "zip" in _body(resp)["error"]["message"]
|
||
|
||
|
||
def test_create_missing_fields(ext_api):
|
||
resp = ext_api.post(
|
||
"/api/bioflow/extensions",
|
||
data={"display_name": "", "command": ""},
|
||
)
|
||
assert resp.status_code in (422, 400)
|
||
|
||
|
||
def test_get_404(ext_api):
|
||
resp = ext_api.get("/api/bioflow/extensions/ext_nope")
|
||
assert resp.status_code == 404
|
||
|
||
|
||
def test_list_published_first(ext_api):
|
||
a = _body(_post_ext(ext_api, name="扩展 A"))["data"]
|
||
b = _body(_post_ext(ext_api, name="扩展 B"))["data"]
|
||
_publish_quick(ext_api, b["ext_id"], executor=FakeExtExecutor(0))
|
||
resp = ext_api.get("/api/bioflow/extensions")
|
||
items = _body(resp)["data"]["items"]
|
||
assert items[0]["ext_id"] == b["ext_id"]
|
||
|
||
|
||
def test_delete(ext_api):
|
||
e = _body(_post_ext(ext_api))["data"]
|
||
resp = ext_api.delete(f"/api/bioflow/extensions/{e['ext_id']}")
|
||
assert resp.status_code == 204
|
||
assert ext_api.get(f"/api/bioflow/extensions/{e['ext_id']}").status_code == 404
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 生命周期动作（build/trial/validate/repair/confirm/publish）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _publish_quick(session, ext_id, *, executor=None):
|
||
"""build → trial → validate → confirm → publish 快速链。"""
|
||
if executor is not None:
|
||
_store().executor = executor
|
||
r = session.post(f"/api/bioflow/extensions/{ext_id}/build")
|
||
assert r.status_code == 200, _body(r)
|
||
r = session.post(f"/api/bioflow/extensions/{ext_id}/trial-run")
|
||
assert r.status_code == 200, _body(r)
|
||
r = session.post(f"/api/bioflow/extensions/{ext_id}/validate")
|
||
assert r.status_code == 200, _body(r)
|
||
r = session.post(f"/api/bioflow/extensions/{ext_id}/confirm", json={"nl_triggers": ["质量统计"]})
|
||
assert r.status_code == 200, _body(r)
|
||
r = session.post(f"/api/bioflow/extensions/{ext_id}/publish")
|
||
assert r.status_code == 200, _body(r)
|
||
return _body(r)["data"]
|
||
|
||
|
||
def test_build_rules_fallback(ext_api):
|
||
e = _body(_post_ext(ext_api))["data"]
|
||
resp = ext_api.post(f"/api/bioflow/extensions/{e['ext_id']}/build")
|
||
assert resp.status_code == 200
|
||
ext = _body(resp)["data"]
|
||
assert ext["status"] == "built"
|
||
assert ext["last_build"]["source"] == "rules"
|
||
assert "{code_dir}" in ext["command_template"]
|
||
assert "{output_dir}" in ext["command_template"]
|
||
|
||
|
||
def test_build_wrong_state_409(ext_api):
|
||
e = _body(_post_ext(ext_api))["data"]
|
||
# draft → publish 非法（须 build/trial/validate/confirm 链）
|
||
resp = ext_api.post(f"/api/bioflow/extensions/{e['ext_id']}/publish")
|
||
assert resp.status_code == 409
|
||
assert _body(resp)["error"]["code"] == "EXT_STATUS"
|
||
|
||
|
||
def test_trial_without_executor_503(ext_api):
|
||
e = _body(_post_ext(ext_api))["data"]
|
||
ext_api.post(f"/api/bioflow/extensions/{e['ext_id']}/build")
|
||
resp = ext_api.post(f"/api/bioflow/extensions/{e['ext_id']}/trial-run")
|
||
assert resp.status_code == 503
|
||
|
||
|
||
def test_full_lifecycle(ext_api):
|
||
e = _body(_post_ext(ext_api, test_blob=b">s\nACGT\n",
|
||
golden={"test_input": "input.fastq", "expected_outputs": ["result.txt"]}))["data"]
|
||
ext_id = e["ext_id"]
|
||
_publish_quick(ext_api, ext_id, executor=FakeExtExecutor(0, "ok\n", ("result.txt",)))
|
||
ext = _body(ext_api.get(f"/api/bioflow/extensions/{ext_id}"))["data"]
|
||
assert ext["status"] == "published"
|
||
assert ext["nl_triggers"] == ["质量统计"]
|
||
# 执行器收到 ext_dir/output_dir（Dev-B 契约字段）
|
||
assert _store().executor.calls[-1].ext_dir.endswith(ext_id)
|
||
assert _store().executor.calls[-1].output_dir
|
||
|
||
|
||
def test_trial_fail_then_repair(ext_api):
|
||
e = _body(_post_ext(ext_api))["data"]
|
||
ext_id = e["ext_id"]
|
||
_store().executor = FakeExtExecutor(1, "No module named 'pandas'\n", ())
|
||
r = ext_api.post(f"/api/bioflow/extensions/{ext_id}/build")
|
||
assert r.status_code == 200
|
||
r = ext_api.post(f"/api/bioflow/extensions/{ext_id}/trial-run")
|
||
assert _body(r)["data"]["status"] == "trial_failed"
|
||
r = ext_api.post(f"/api/bioflow/extensions/{ext_id}/repair")
|
||
assert r.status_code == 200
|
||
ext = _body(r)["data"]
|
||
assert ext["status"] == "built"
|
||
assert "pandas" in ext["requirements"]["pip"]
|
||
|
||
|
||
def test_repair_limit_reached(ext_api):
|
||
e = _body(_post_ext(ext_api))["data"]
|
||
ext_id = e["ext_id"]
|
||
_store().max_repair_rounds = 1
|
||
_store().executor = FakeExtExecutor(1, "No module named 'pandas'\n", ())
|
||
ext_api.post(f"/api/bioflow/extensions/{ext_id}/build")
|
||
ext_api.post(f"/api/bioflow/extensions/{ext_id}/trial-run")
|
||
assert ext_api.post(f"/api/bioflow/extensions/{ext_id}/repair").status_code == 200
|
||
ext_api.post(f"/api/bioflow/extensions/{ext_id}/trial-run") # 再失败
|
||
resp = ext_api.post(f"/api/bioflow/extensions/{ext_id}/repair")
|
||
assert resp.status_code == 409
|
||
assert "上限" in _body(resp)["error"]["message"]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 运行（run/runs/events）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_run_not_published_409(ext_api):
|
||
e = _body(_post_ext(ext_api))["data"]
|
||
resp = ext_api.post(f"/api/bioflow/extensions/{e['ext_id']}/run", json={"params": {}})
|
||
assert resp.status_code == 409
|
||
|
||
|
||
def test_run_and_events(ext_api):
|
||
e = _body(_post_ext(ext_api))["data"]
|
||
ext_id = e["ext_id"]
|
||
_publish_quick(ext_api, ext_id, executor=FakeExtExecutor(0, "run ok\n", ("out.tsv",)))
|
||
resp = ext_api.post(f"/api/bioflow/extensions/{ext_id}/run", json={"params": {"cpu": "4"}})
|
||
assert resp.status_code == 200
|
||
result = _body(resp)["data"]
|
||
assert result["status"] == "done"
|
||
assert "out.tsv" in result["outputs"]
|
||
|
||
runs = _body(ext_api.get(f"/api/bioflow/extensions/{ext_id}/runs"))["data"]
|
||
assert runs["total"] == 1
|
||
assert runs["items"][0]["run_id"] == result["run_id"]
|
||
|
||
# events SSE（占位实现：已完成 run 的快照帧）
|
||
from app.routers.extension import run_events
|
||
|
||
async def _collect():
|
||
resp = await run_events(ext_id, result["run_id"], request=None)
|
||
chunks = []
|
||
async for chunk in resp.body_iterator:
|
||
chunks.append(chunk)
|
||
return b"".join(c.encode("utf-8") if isinstance(c, str) else c for c in chunks).decode("utf-8")
|
||
|
||
raw = asyncio.get_event_loop().run_until_complete(_collect())
|
||
assert "extension_run_started" in raw
|
||
assert "extension_run_done" in raw
|
||
assert "run ok" in raw
|
||
|
||
|
||
def test_run_failed_events(ext_api):
|
||
e = _body(_post_ext(ext_api))["data"]
|
||
ext_id = e["ext_id"]
|
||
_publish_quick(ext_api, ext_id, executor=FakeExtExecutor(0, "trial ok\n"))
|
||
_store().executor = FakeExtExecutor(3, "output contract failed\n", ())
|
||
resp = ext_api.post(f"/api/bioflow/extensions/{ext_id}/run", json={"params": {}})
|
||
result = _body(resp)["data"]
|
||
assert result["status"] == "failed"
|
||
assert result["exit_code"] == 3
|
||
|
||
from app.routers.extension import run_events
|
||
|
||
async def _collect():
|
||
resp = await run_events(ext_id, result["run_id"], request=None)
|
||
chunks = []
|
||
async for chunk in resp.body_iterator:
|
||
chunks.append(chunk)
|
||
return b"".join(c.encode("utf-8") if isinstance(c, str) else c for c in chunks).decode("utf-8")
|
||
|
||
raw = asyncio.get_event_loop().run_until_complete(_collect())
|
||
assert "extension_run_failed" in raw
|
||
"""后端新增测试:三种形态 + 自适应分类 + 嵌套落盘(追加到 test_extension_api.py 尾部)"""
|
||
import io, json, tarfile, zipfile
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
|
||
from tests.test_extension_api import _body, _post_ext, FakeExtExecutor, ApiSession # noqa: F401
|
||
import app.services.bioflow.extensions as ext_mod
|
||
from app.routers.extension import _read_archive, _classify_items, _safe_rel_path
|
||
|
||
|
||
def _make_tar(entries, mode="w:gz"):
|
||
buf = io.BytesIO()
|
||
with tarfile.open(fileobj=buf, mode=mode) as tf:
|
||
for name, data in entries:
|
||
info = tarfile.TarInfo(name=name)
|
||
info.size = len(data)
|
||
tf.addfile(info, io.BytesIO(data))
|
||
return buf.getvalue()
|
||
|
||
|
||
def test_safe_rel_path_traversal():
|
||
assert _safe_rel_path("a/b/c.py") == "a/b/c.py"
|
||
assert _safe_rel_path("../../evil.py") == "evil.py"
|
||
assert _safe_rel_path("/etc/passwd") == "etc/passwd"
|
||
assert _safe_rel_path("") == ""
|
||
|
||
|
||
def test_read_archive_tar_and_targz():
|
||
blob = _make_tar([("raw_code/run.sh", b"#!/bin/sh\necho hi\n"), ("raw_code/a.py", b"print(1)")], "w:gz")
|
||
items = _read_archive(blob)
|
||
assert sorted(n for n, _ in items) == ["raw_code/a.py", "raw_code/run.sh"]
|
||
blob_tar = _make_tar([("raw_code/run.sh", b"#!/bin/sh\necho hi\n")], "w:")
|
||
items2 = _read_archive(blob_tar)
|
||
assert items2[0][0] == "raw_code/run.sh"
|
||
|
||
|
||
def test_read_archive_bad_tar():
|
||
import app.services.bioflow.extensions as em
|
||
with pytest.raises(em.ExtensionError):
|
||
_read_archive(b"not a tar at all")
|
||
|
||
|
||
def test_classify_standard_structure():
|
||
items = [("raw_code/check.py", b"x"), ("raw_test_data/samples.tsv", b"a\tb")]
|
||
code, data = _classify_items(items, "python check.py --samples samples.tsv", [], [])
|
||
assert [n for n, _ in code] == ["check.py"]
|
||
assert [n for n, _ in data] == ["samples.tsv"]
|
||
|
||
|
||
def test_classify_scattered_files():
|
||
# 散落文件:无目录前缀,靠扩展名分类;未知扩展名归 code
|
||
items = [("count.py", b"x"), ("a.fasta", b">s"), ("b.fastq", b"@r"), ("weird.xyz", b"?")]
|
||
code, data = _classify_items(items, "python count.py", [], [])
|
||
assert sorted(n for n, _ in code) == ["count.py", "weird.xyz"]
|
||
assert sorted(n for n, _ in data) == ["a.fasta", "b.fastq"]
|
||
|
||
|
||
def test_classify_command_param_as_data():
|
||
items = [("tool.py", b"x"), ("samples.tsv", b"a\tb")]
|
||
code, data = _classify_items(items, "python tool.py --samples samples.tsv --out outputs/", [], [])
|
||
assert [n for n, _ in code] == ["tool.py"]
|
||
assert [n for n, _ in data] == ["samples.tsv"]
|
||
|
||
|
||
def test_classify_nested_module_preserved():
|
||
items = [("panfamily/io.py", b"def read_fasta(): pass")]
|
||
code, data = _classify_items(items, "python main.py", [], [])
|
||
assert [n for n, _ in code] == ["panfamily/io.py"]
|
||
|
||
|
||
def test_upload_dir_multipart(ext_api):
|
||
"""目录直传:code_files 多文件 + 相对路径落盘"""
|
||
import app.services.bioflow.extensions as em
|
||
from pathlib import Path
|
||
em.reset_store()
|
||
session = ext_api
|
||
files = [
|
||
("code_files", ("raw_code/run.sh", b"#!/bin/sh\necho hi\n", "application/octet-stream")),
|
||
("code_files", ("raw_code/tool.py", b"print(1)", "application/octet-stream")),
|
||
("code_files", ("raw_test_data/in.fa", b">s\nACGT", "application/octet-stream")),
|
||
]
|
||
data = {"display_name": "目录测试", "command": "bash run.sh", "runtime_image": "bioflow-gf"}
|
||
resp = session.post("/api/bioflow/extensions", files=files, data=data)
|
||
assert resp.status_code == 201, _body(resp)
|
||
body = _body(resp)
|
||
ext_id = body["data"]["ext_id"]
|
||
root = em.get_store().ext_dir(ext_id)
|
||
assert (root / "raw_code" / "run.sh").is_file()
|
||
assert (root / "raw_code" / "tool.py").is_file()
|
||
assert (root / "raw_test_data" / "in.fa").is_file()
|
||
em.reset_store()
|
||
|
||
|
||
def test_upload_tar_multipart(ext_api):
|
||
"""tar 归档上传:服务端解压"""
|
||
import app.services.bioflow.extensions as em
|
||
em.reset_store()
|
||
session = ext_api
|
||
blob = _make_tar([("run.sh", b"#!/bin/sh\necho hi\n"), ("tool.py", b"print(1)")], "w:gz")
|
||
files = {"code_zip": ("code.tar.gz", blob, "application/gzip")}
|
||
data = {"display_name": "tar 测试", "command": "bash run.sh"}
|
||
resp = session.post("/api/bioflow/extensions", files=files, data=data)
|
||
assert resp.status_code == 201, _body(resp)
|
||
root = em.get_store().ext_dir(_body(resp)["data"]["ext_id"])
|
||
assert (root / "raw_code" / "run.sh").is_file()
|
||
em.reset_store()
|
||
|
||
|
||
def test_upload_pure_data_422(ext_api):
|
||
"""纯数据包(无代码文件)→ 422 可读错误"""
|
||
import app.services.bioflow.extensions as em
|
||
em.reset_store()
|
||
session = ext_api
|
||
files = {"code_zip": ("data.zip", None, "application/zip")}
|
||
# 构造纯数据 zip
|
||
buf = io.BytesIO()
|
||
with zipfile.ZipFile(buf, "w") as zf:
|
||
zf.writestr("samples.tsv", b"a\tb")
|
||
files = {"code_zip": ("data.zip", buf.getvalue(), "application/zip")}
|
||
data = {"display_name": "纯数据", "command": "python nothing.py"}
|
||
resp = session.post("/api/bioflow/extensions", files=files, data=data)
|
||
assert resp.status_code == 422
|
||
assert "代码文件" in _body(resp)["error"]["message"]
|
||
em.reset_store()
|