adminandClaude Fable 5
79b4c7c52c
- cli/run_extension.py:容器通用入口(宿主挂载 /opt/bioflow-entry,镜像零改动);
extension.json → 模板渲染(白名单 12 占位符 + param_*,{output_dir} 必含)→
subprocess 执行(无 shell,进程组 SIGTERM→3s→SIGKILL)→ 输出契约校验(空目录退出码 3);
零第三方依赖(镜像内标准库)
- extension_exec.py:服务端 docker 执行器(list-form;三数据卷+入口卷;
--cpus/--memory 白名单独立复制不 import kernel.py;terminate→kill 同约定;
返回 {exit_code,timed_out,log_tail,cmd};run_stream 逐行增量)
- extension_image.py:指纹 sha256(base+tag+排序依赖) 前 12 位 →
bioflow-ext-<fp>:dev;inspect 命中复用;Dockerfile FROM base + apt + /opt/conda/bin/pip;
失败分类(pip/apt/network/timeout/docker_missing)+ LABEL 定位中间层清理;
registry 台账 + 引用计数 + collect_garbage
- extension_streams.py:契约 §4 事件流(trial/run async 生成器,增量日志节流,
EventBus 双发,防御性失败帧,aclose 回收)
- 单测 53 用例(15 unittest + 38 pytest)全绿;现有 bioflow 226 用例回归通过
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
229 lines
8.7 KiB
Python
229 lines
8.7 KiB
Python
"""tests/test_extension_streams.py —— 扩展 SSE 事件流(契约 §4 帧形态 + 增量日志)。"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
from app.services.bioflow.extension_exec import ExtensionExecConfig, ExtensionExecutor, ExtensionRunRequest
|
|
from app.services.bioflow.extension_streams import run_stream, trial_stream
|
|
|
|
pytestmark = pytest.mark.bioflow_unit
|
|
|
|
|
|
class FakeEvents:
|
|
def __init__(self):
|
|
self.published: list[tuple[str, dict]] = []
|
|
|
|
def publish(self, event_type, data):
|
|
self.published.append((event_type, data))
|
|
|
|
|
|
def _cfg(tmp_path):
|
|
entry = tmp_path / "cli"
|
|
entry.mkdir()
|
|
(entry / "run_extension.py").write_text("# entry\n", encoding="utf-8")
|
|
return ExtensionExecConfig(host_uid=1234, entry_script_dir=str(entry))
|
|
|
|
|
|
def _request(tmp_path):
|
|
for d in ("ext", "input"):
|
|
(tmp_path / d).mkdir()
|
|
return ExtensionRunRequest(image="bioflow-gf", ext_dir=tmp_path / "ext",
|
|
input_dir=tmp_path / "input",
|
|
output_dir=tmp_path / "output")
|
|
|
|
|
|
def _run(coro):
|
|
return asyncio.new_event_loop().run_until_complete(coro)
|
|
|
|
|
|
def _patch_stream(monkeypatch, chunk_lines, exit_code=0, timed_out=False):
|
|
"""executor.run_stream 替身:固定 chunk 序列 + final。"""
|
|
|
|
async def fake_stream(self, request):
|
|
for line in chunk_lines:
|
|
await asyncio.sleep(0)
|
|
yield {"chunk": line}
|
|
yield {"final": True, "exit_code": exit_code, "timed_out": timed_out,
|
|
"log_tail": "tail", "full_log": "".join(chunk_lines), "cmd": []}
|
|
|
|
monkeypatch.setattr(ExtensionExecutor, "run_stream", fake_stream)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# trial_stream
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_trial_stream_success_frames(tmp_path, monkeypatch):
|
|
_patch_stream(monkeypatch, ["step 1 ok\n", "step 2 ok\n", "wrote output\n"])
|
|
events = FakeEvents()
|
|
executor = ExtensionExecutor(_cfg(tmp_path))
|
|
|
|
async def collect():
|
|
return [f async for f in trial_stream("ext_1", executor, _request(tmp_path),
|
|
events, log_path=tmp_path / "trial.log")]
|
|
|
|
frames = _run(collect())
|
|
assert frames[0]["event"] == "extension_trial"
|
|
assert frames[0]["data"]["status"] == "running"
|
|
assert frames[-1]["event"] == "extension_trial"
|
|
assert frames[-1]["data"]["status"] == "done"
|
|
assert frames[-1]["data"]["exit_code"] == 0
|
|
assert frames[-1]["data"]["log_tail"]
|
|
# 持久事件同步发布(running + done)
|
|
assert [e[0] for e in events.published] == ["extension_trial", "extension_trial"]
|
|
assert events.published[-1][1]["status"] == "done"
|
|
# 日志落盘
|
|
assert "step 1 ok" in (tmp_path / "trial.log").read_text(encoding="utf-8")
|
|
|
|
|
|
def test_trial_stream_log_throttle(tmp_path, monkeypatch):
|
|
# 超过阈值(80 行)才推中间增量帧
|
|
_patch_stream(monkeypatch, [f"line{i}\n" for i in range(100)])
|
|
events = FakeEvents()
|
|
executor = ExtensionExecutor(_cfg(tmp_path))
|
|
|
|
async def collect():
|
|
return [f async for f in trial_stream("ext_1", executor, _request(tmp_path),
|
|
events, log_path=tmp_path / "trial.log")]
|
|
|
|
frames = _run(collect())
|
|
log_frames = [f for f in frames if f["event"] == "extension_trial" and "log" in f["data"]]
|
|
assert log_frames, "应存在增量日志帧"
|
|
assert frames[-1]["data"]["status"] == "done"
|
|
|
|
|
|
def test_trial_stream_output_contract_failure(tmp_path, monkeypatch):
|
|
_patch_stream(monkeypatch, ["wrote nothing\n"], exit_code=3)
|
|
events = FakeEvents()
|
|
executor = ExtensionExecutor(_cfg(tmp_path))
|
|
|
|
async def collect():
|
|
return [f async for f in trial_stream("ext_1", executor, _request(tmp_path),
|
|
events, log_path=tmp_path / "trial.log")]
|
|
|
|
frames = _run(collect())
|
|
assert frames[-1]["data"]["status"] == "failed"
|
|
assert "输出契约" in frames[-1]["data"]["error"]
|
|
|
|
|
|
def test_trial_stream_failed_nonzero(tmp_path, monkeypatch):
|
|
_patch_stream(monkeypatch, ["boom\n"], exit_code=7)
|
|
events = FakeEvents()
|
|
executor = ExtensionExecutor(_cfg(tmp_path))
|
|
|
|
async def collect():
|
|
return [f async for f in trial_stream("ext_1", executor, _request(tmp_path),
|
|
events, log_path=tmp_path / "trial.log")]
|
|
|
|
frames = _run(collect())
|
|
assert frames[-1]["data"]["status"] == "failed"
|
|
assert frames[-1]["data"]["exit_code"] == 7
|
|
assert "退出码 7" in frames[-1]["data"]["error"]
|
|
|
|
|
|
def test_trial_stream_timeout(tmp_path, monkeypatch):
|
|
_patch_stream(monkeypatch, ["long run\n"], exit_code=137, timed_out=True)
|
|
events = FakeEvents()
|
|
executor = ExtensionExecutor(_cfg(tmp_path))
|
|
|
|
async def collect():
|
|
return [f async for f in trial_stream("ext_1", executor, _request(tmp_path),
|
|
events, log_path=tmp_path / "trial.log")]
|
|
|
|
frames = _run(collect())
|
|
assert frames[-1]["data"]["status"] == "failed"
|
|
assert "超时" in frames[-1]["data"]["error"]
|
|
|
|
|
|
def test_trial_stream_executor_exception(tmp_path, monkeypatch):
|
|
def boom(self, request):
|
|
async def gen():
|
|
raise RuntimeError("executor broken")
|
|
yield # pragma: no cover
|
|
return gen()
|
|
|
|
monkeypatch.setattr(ExtensionExecutor, "run_stream", boom)
|
|
events = FakeEvents()
|
|
executor = ExtensionExecutor(_cfg(tmp_path))
|
|
|
|
async def collect():
|
|
return [f async for f in trial_stream("ext_1", executor, _request(tmp_path),
|
|
events, log_path=tmp_path / "trial.log")]
|
|
|
|
frames = _run(collect())
|
|
assert frames[-1]["data"]["status"] == "failed"
|
|
assert "中断" in frames[-1]["data"]["error"]
|
|
assert frames[-1]["data"]["exit_code"] == -1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# run_stream
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def test_run_stream_success_frames(tmp_path, monkeypatch):
|
|
_patch_stream(monkeypatch, ["run line 1\n", "run line 2\n"])
|
|
events = FakeEvents()
|
|
executor = ExtensionExecutor(_cfg(tmp_path))
|
|
request = _request(tmp_path)
|
|
(request.output_dir).mkdir(parents=True)
|
|
(request.output_dir / "result.tsv").write_text("a\tb\n", encoding="utf-8")
|
|
|
|
async def collect():
|
|
return [f async for f in run_stream("run_1", "ext_1", executor, request,
|
|
events, log_path=tmp_path / "run.log")]
|
|
|
|
frames = _run(collect())
|
|
assert frames[0]["event"] == "extension_run_started"
|
|
assert frames[0]["data"] == {"run_id": "run_1", "ext_id": "ext_1"}
|
|
assert frames[-1]["event"] == "extension_run_done"
|
|
assert frames[-1]["data"]["run_id"] == "run_1"
|
|
assert any(o["path"].endswith("result.tsv") for o in frames[-1]["data"]["outputs"])
|
|
assert [e[0] for e in events.published] == ["extension_run_started", "extension_run_done"]
|
|
|
|
|
|
def test_run_stream_log_throttle(tmp_path, monkeypatch):
|
|
_patch_stream(monkeypatch, [f"line{i}\n" for i in range(100)])
|
|
events = FakeEvents()
|
|
executor = ExtensionExecutor(_cfg(tmp_path))
|
|
|
|
async def collect():
|
|
return [f async for f in run_stream("run_1", "ext_1", executor, _request(tmp_path),
|
|
events, log_path=tmp_path / "run.log")]
|
|
|
|
frames = _run(collect())
|
|
log_frames = [f for f in frames if f["event"] == "extension_run_log"]
|
|
assert log_frames
|
|
assert frames[-1]["event"] == "extension_run_done"
|
|
|
|
|
|
def test_run_stream_failed_contract(tmp_path, monkeypatch):
|
|
_patch_stream(monkeypatch, ["empty\n"], exit_code=3)
|
|
events = FakeEvents()
|
|
executor = ExtensionExecutor(_cfg(tmp_path))
|
|
|
|
async def collect():
|
|
return [f async for f in run_stream("run_1", "ext_1", executor, _request(tmp_path),
|
|
events, log_path=tmp_path / "run.log")]
|
|
|
|
frames = _run(collect())
|
|
assert frames[-1]["event"] == "extension_run_failed"
|
|
assert frames[-1]["data"]["exit_code"] == 3
|
|
assert "输出契约" in frames[-1]["data"]["error"]
|
|
|
|
|
|
def test_run_stream_failed_nonzero(tmp_path, monkeypatch):
|
|
_patch_stream(monkeypatch, ["boom\n"], exit_code=7)
|
|
events = FakeEvents()
|
|
executor = ExtensionExecutor(_cfg(tmp_path))
|
|
|
|
async def collect():
|
|
return [f async for f in run_stream("run_1", "ext_1", executor, _request(tmp_path),
|
|
events, log_path=tmp_path / "run.log")]
|
|
|
|
frames = _run(collect())
|
|
assert frames[-1]["event"] == "extension_run_failed"
|
|
assert frames[-1]["data"]["exit_code"] == 7
|
|
assert "退出码 7" in frames[-1]["data"]["error"]
|