"""tests/test_runner.py —— runner 状态机全路径（mock docker 执行器）。 覆盖（TC-BIO-RUN 系列）：成功推进/失败/重跑/跳过/暂停/恢复/重启续跑/超时/ 日志截断/并发上限/输入缺失等待/确认等待/自动连续/信号量释放/事件成对。 注意：runner 内部用 asyncio.create_task 派生任务，测试必须「单场景单 loop」—— 每个用例在一个新建 event loop 内完成 create→run→断言前的所有异步步骤。 """ from __future__ import annotations import asyncio import pytest from conftest import FakeExecutor, FakeTimedOutExecutor, make_runner from app.services.bioflow.runner import ( JOB_AWAIT_INPUTS, JOB_DONE, JOB_FAILED, JOB_PAUSED, STAGE_DONE, STAGE_PAUSED, STAGE_QUEUED, STAGE_RUNNING, STAGE_FAILED, STAGE_SKIPPED, STAGE_WAITING_CONFIRM, RunnerError, ) pytestmark = pytest.mark.bioflow_unit def _scenario(coro_factory): """单 loop 跑完整场景。""" loop = asyncio.new_event_loop() try: return loop.run_until_complete(coro_factory()) finally: loop.close() async def _drain(runner, seconds: float = 5.0) -> None: """等待 runner 派生的后台任务全部结束。""" while runner._tasks: await asyncio.wait_for(asyncio.gather(*list(runner._tasks)), timeout=seconds) await asyncio.sleep(0.01) def _collected(events, job_id): return [e for e in events.replay(job_id) if e["type"] != "job_recovered"] def test_full_chain_success(db, events, catalog, fake_executor): """TC-BIO-RUN-001：全链成功推进（自动连续）。""" async def scenario(): runner = make_runner(db, events, fake_executor, catalog) record = await runner.create_job( "s1", "SPL 鉴定", ["identify", "phylogeny", "motif"], auto_continue=True, auto_start=False) await runner.run_job(record) return runner, record runner, record = _scenario(scenario) assert record.status == JOB_DONE ev_types = [e["type"] for e in _collected(events, record.id)] assert ev_types.count("stage_started") == 3 assert ev_types.count("stage_done") == 3 assert ev_types[-1] == "job_done" assert ev_types[0] == "job_created" assert all(runner.stage_status(record, s) == STAGE_DONE for s in record.stage_chain) def test_stage_failure_stops_chain(db, events, catalog): """TC-BIO-RUN-002：阶段失败转 failed，后续阶段不执行。""" executor = FakeExecutor(exit_code=1, log="boom\n") async def scenario(): runner = make_runner(db, events, executor, catalog) record = await runner.create_job( "s1", "t", ["identify", "phylogeny", "motif"], auto_continue=True, auto_start=False) await runner.run_job(record) return runner, record runner, record = _scenario(scenario) assert record.status == JOB_FAILED assert runner.stage_status(record, "identify") == STAGE_FAILED assert runner.stage_status(record, "phylogeny") == "waiting" ev_types = [e["type"] for e in _collected(events, record.id)] assert "stage_failed" in ev_types failed_ev = [e for e in _collected(events, record.id) if e["type"] == "stage_failed"][0] assert failed_ev["data"]["stage"] == "identify" assert failed_ev["data"]["reason"] == "exit_code" assert [c["stage"] for c in executor.calls] == ["identify"] # 后续阶段未执行 def test_rerun_after_failure(db, events, catalog): """TC-BIO-RUN-003：失败后 rerun 成功，attempt=2，继续后续阶段。""" executor = FakeExecutor(exit_code=1) async def scenario(): runner = make_runner(db, events, executor, catalog) record = await runner.create_job( "s1", "t", ["identify", "phylogeny"], auto_continue=True, auto_start=False) await runner.run_job(record) assert record.status == JOB_FAILED executor.exit_code = 0 await runner.rerun_stage(record, "identify", None) await _drain(runner) return runner, record runner, record = _scenario(scenario) assert runner.stage_status(record, "identify") == STAGE_DONE assert record._stage_states["identify"]["attempt"] == 2 assert record.status == JOB_DONE # 后续阶段自动推进 def test_skip_after_failure(db, events, catalog): """TC-BIO-RUN-004：失败后 skip → skipped，后续阶段可执行。""" executor = FakeExecutor(exit_code=1) async def scenario(): runner = make_runner(db, events, executor, catalog) record = await runner.create_job( "s1", "t", ["identify", "phylogeny"], auto_continue=True, auto_start=False) await runner.run_job(record) assert runner.stage_status(record, "identify") == STAGE_FAILED executor.exit_code = 0 await runner.skip_stage(record, "identify") await _drain(runner) return runner, record runner, record = _scenario(scenario) assert runner.stage_status(record, "identify") == STAGE_SKIPPED ev_types = [e["type"] for e in _collected(events, record.id)] assert "stage_skipped" in ev_types assert record.status == JOB_DONE assert runner.stage_status(record, "phylogeny") == STAGE_DONE def test_rerun_with_params_goes_confirm(db, events, catalog): """TC-BIO-RUN-005：带参数重跑 → waiting_confirm 再确认；新参数进入执行。""" executor = FakeExecutor(exit_code=1) async def scenario(): runner = make_runner(db, events, executor, catalog) record = await runner.create_job( "s1", "t", ["identify"], auto_continue=True, auto_start=False) await runner.run_job(record) await runner.rerun_stage(record, "identify", {"cpu": "8"}) assert runner.stage_status(record, "identify") == STAGE_WAITING_CONFIRM executor.exit_code = 0 await runner.confirm_stage(record, "identify", {"cpu": "8"}) await _drain(runner) return runner, record runner, record = _scenario(scenario) assert record._stage_states["identify"]["params"]["cpu"] == "8" assert runner.stage_status(record, "identify") == STAGE_DONE assert executor.calls[-1]["cpu"] == "8" def test_rerun_non_failed_rejected(db, events, catalog, fake_executor): """TC-BIO-RUN-024：done 阶段 rerun/skip 拒绝。""" async def scenario(): runner = make_runner(db, events, fake_executor, catalog) record = await runner.create_job( "s1", "t", ["identify"], auto_continue=True, auto_start=False) await runner.run_job(record) return runner, record runner, record = _scenario(scenario) with pytest.raises(RunnerError, match="不可重跑"): _scenario(lambda: runner.rerun_stage(record, "identify", None)) with pytest.raises(RunnerError, match="不可跳过"): _scenario(lambda: runner.skip_stage(record, "identify")) def test_pause_at_stage_boundary(db, events, catalog): """TC-BIO-RUN-006：运行中暂停立即生效（当前阶段置 paused,不判失败）。""" executor = FakeExecutor(delay=0.05) async def scenario(): runner = make_runner(db, events, executor, catalog) record = await runner.create_job( "s1", "t", ["identify", "phylogeny"], auto_continue=True, auto_start=False) task = asyncio.create_task(runner.run_job(record)) await asyncio.sleep(0.02) # identify 已进入执行 await runner.pause_job(record) await asyncio.wait_for(task, timeout=5) return runner, record runner, record = _scenario(scenario) assert record.status == JOB_PAUSED assert runner.stage_status(record, "identify") == STAGE_PAUSED # 立即暂停语义 assert all(c["stage"] != "phylogeny" for c in executor.calls) def test_confirm_idempotent_while_running(db, events, catalog): """确认幂等：已确认且 running/queued 的重复 confirm 不报 409。""" executor = FakeExecutor(delay=0.1) async def scenario(): runner = make_runner(db, events, executor, catalog) record = await runner.create_job( "s1", "t", ["identify"], auto_continue=False, auto_start=False) await runner.run_job(record) # 手动模式:停 waiting_confirm await runner.confirm_stage(record, "identify", {}) # 立即二次 confirm:阶段已 running 且 _confirmed → 幂等放行,不抛异常 await runner.confirm_stage(record, "identify", {}) await _drain(runner) return runner, record runner, record = _scenario(scenario) assert runner.stage_status(record, "identify") in (STAGE_RUNNING, STAGE_DONE, STAGE_QUEUED) def test_resume_continues_without_rerun(db, events, catalog): """TC-BIO-RUN-007：暂停后恢复续跑，不重跑已完成。""" executor = FakeExecutor(delay=0.02) async def scenario(): runner = make_runner(db, events, executor, catalog) record = await runner.create_job( "s1", "t", ["identify", "phylogeny"], auto_continue=True, auto_start=False) task = asyncio.create_task(runner.run_job(record)) await asyncio.sleep(0.03) await runner.pause_job(record) await asyncio.wait_for(task, timeout=5) assert record.status == JOB_PAUSED await runner.resume_job(record) await _drain(runner) return runner, record runner, record = _scenario(scenario) assert record.status == JOB_DONE identify_calls = [c for c in executor.calls if c["stage"] == "identify"] assert len(identify_calls) == 1 # 不重跑 def test_timeout_fails_stage(db, events, catalog): """TC-BIO-RUN-012：超时 → failed（reason=timeout）。""" executor = FakeTimedOutExecutor() async def scenario(): runner = make_runner(db, events, executor, catalog) record = await runner.create_job( "s1", "t", ["identify"], auto_continue=True, auto_start=False) await runner.run_job(record) return runner, record runner, record = _scenario(scenario) assert runner.stage_status(record, "identify") == STAGE_FAILED failed = [e for e in _collected(events, record.id) if e["type"] == "stage_failed"][0] assert failed["data"]["reason"] == "timeout" def test_log_tail_truncated(db, events, catalog): """TC-BIO-RUN-013：日志截断只留尾 80 行。""" executor = FakeExecutor(log="\n".join(f"L{i}" for i in range(500)) + "\n") async def scenario(): runner = make_runner(db, events, executor, catalog) record = await runner.create_job( "s1", "t", ["identify"], auto_continue=True, auto_start=False) await runner.run_job(record) return runner, record runner, record = _scenario(scenario) tail = record._stage_states["identify"]["log_tail"] assert tail.count("\n") + 1 == 80 assert tail.startswith("L420") assert tail.rstrip().endswith("L499") def test_exit_nonzero_judged_failed(db, events, catalog): """TC-BIO-RUN-014：退出码非 0 判失败，log_tail 含 stderr。""" executor = FakeExecutor(exit_code=3, log="error output\n") async def scenario(): runner = make_runner(db, events, executor, catalog) record = await runner.create_job( "s1", "t", ["identify"], auto_continue=True, auto_start=False) await runner.run_job(record) return runner, record runner, record = _scenario(scenario) assert runner.stage_status(record, "identify") == STAGE_FAILED assert "error output" in record._stage_states["identify"]["log_tail"] def test_manifest_missing_judged_failed(db, events, catalog): """TC-BIO-RUN-015：退出码 0 但 manifest 缺 → failed（reason=no_manifest）。""" executor = FakeExecutor(exit_code=0, manifest_status="missing") async def scenario(): runner = make_runner(db, events, executor, catalog) record = await runner.create_job( "s1", "t", ["identify"], auto_continue=True, auto_start=False) await runner.run_job(record) return runner, record runner, record = _scenario(scenario) assert runner.stage_status(record, "identify") == STAGE_FAILED failed = [e for e in _collected(events, record.id) if e["type"] == "stage_failed"][0] assert failed["data"]["reason"] == "no_manifest" def test_manifest_failed_status(db, events, catalog): """TC-BIO-RUN-016：manifest status=failed 判败（不因退出码 0 误判成功）。""" executor = FakeExecutor(exit_code=0, manifest_status="failed") async def scenario(): runner = make_runner(db, events, executor, catalog) record = await runner.create_job( "s1", "t", ["identify"], auto_continue=True, auto_start=False) await runner.run_job(record) return runner, record runner, record = _scenario(scenario) assert runner.stage_status(record, "identify") == STAGE_FAILED failed = [e for e in _collected(events, record.id) if e["type"] == "stage_failed"][0] assert failed["data"]["reason"] == "manifest_failed" def test_await_files_when_missing(db, events, catalog): """TC-BIO-RUN-017：输入缺失等待文件（force_input_check=True）。""" executor = FakeExecutor() async def scenario(): runner = make_runner(db, events, executor, catalog, check_inputs=True) record = await runner.create_job( "s1", "t", ["identify"], auto_continue=True, auto_start=False) await runner.run_job(record) return runner, record runner, record = _scenario(scenario) assert runner.stage_status(record, "identify") == "await_files" ev = [e for e in _collected(events, record.id) if e["type"] == "stage_await_files"][0] assert ev["data"]["stage"] == "identify" assert ev["data"]["required_files"] assert executor.calls == [] # 不启动 docker def test_await_files_then_resume(db, events, catalog): """TC-BIO-RUN-018：文件补齐自动推进。""" executor = FakeExecutor() async def scenario(): runner = make_runner(db, events, executor, catalog, check_inputs=True) record = await runner.create_job( "s1", "t", ["identify"], auto_continue=True, auto_start=False) await runner.run_job(record) assert runner.stage_status(record, "identify") == "await_files" (record.input_dir / "genome.pep.fa").write_text(">x\nACGT\n", encoding="utf-8") record._stage_states["identify"]["status"] = "waiting" await runner.run_job(record) return runner, record runner, record = _scenario(scenario) assert record.status == JOB_DONE assert runner.stage_status(record, "identify") == STAGE_DONE def test_retry_after_files_auto_continue_runs_to_done(db, events, catalog): """BUG-P7-01①：auto_continue 任务上传后 retry_after_files 自动续跑到 done。""" executor = FakeExecutor() async def scenario(): runner = make_runner(db, events, executor, catalog, check_inputs=True) record = await runner.create_job( "s1", "t", ["identify"], auto_continue=True, auto_start=False) await runner.run_job(record) assert runner.stage_status(record, "identify") == "await_files" assert record.status == JOB_AWAIT_INPUTS assert executor.calls == [] # 模拟上传落盘 (record.input_dir / "genome.pep.fa").write_text(">x\nACGT\n", encoding="utf-8") # 上传端点调用点（BUG-P7-01 修法核心） await runner.retry_after_files(record) await _drain(runner) return runner, record runner, record = _scenario(scenario) assert record.status == JOB_DONE assert runner.stage_status(record, "identify") == STAGE_DONE assert executor.calls # fake executor 记录执行调用 def test_retry_after_files_manual_mode_goes_confirm(db, events, catalog): """BUG-P7-01②：手动模式上传后进入 waiting_confirm 并收到 stage_await_confirm。""" executor = FakeExecutor() async def scenario(): runner = make_runner(db, events, executor, catalog, check_inputs=True) record = await runner.create_job( "s1", "t", ["identify"], auto_continue=False, auto_start=False) await runner.run_job(record) assert runner.stage_status(record, "identify") == "await_files" (record.input_dir / "genome.pep.fa").write_text(">x\nACGT\n", encoding="utf-8") await runner.retry_after_files(record) await _drain(runner) return runner, record runner, record = _scenario(scenario) assert runner.stage_status(record, "identify") == STAGE_WAITING_CONFIRM confirm_events = [e for e in _collected(events, record.id) if e["type"] == "stage_await_confirm"] assert confirm_events assert confirm_events[-1]["data"]["stage"] == "identify" assert executor.calls == [] # 未确认不执行 def test_retry_after_files_non_await_state_is_noop(db, events, catalog): """BUG-P7-01③：非 await_inputs 状态调用为 no-op（不重复 spawn）。""" executor = FakeExecutor() async def scenario(): runner = make_runner(db, events, executor, catalog, check_inputs=True) record = await runner.create_job( "s1", "t", ["identify"], auto_continue=True, auto_start=False) await runner.run_job(record) assert record.status == JOB_AWAIT_INPUTS (record.input_dir / "genome.pep.fa").write_text(">x\nACGT\n", encoding="utf-8") await runner.retry_after_files(record) await _drain(runner) assert record.status == JOB_DONE done_calls = len(executor.calls) tasks_before = len(runner._tasks) # 已完成状态再调用 → no-op：状态不变、不新增任务、无新执行 await runner.retry_after_files(record) await _drain(runner) assert record.status == JOB_DONE assert len(executor.calls) == done_calls # 不重复执行 assert len(runner._tasks) <= tasks_before return runner, record _scenario(scenario) def test_await_confirm_manual_mode(db, events, catalog): """TC-BIO-RUN-019/020：手动模式等待确认；确认后执行。""" executor = FakeExecutor() async def scenario(): runner = make_runner(db, events, executor, catalog) record = await runner.create_job( "s1", "t", ["identify", "phylogeny"], auto_continue=False, auto_start=False) await runner.run_job(record) assert runner.stage_status(record, "identify") == STAGE_WAITING_CONFIRM assert executor.calls == [] # 未确认不执行 await runner.confirm_stage(record, "identify", {}) await _drain(runner) return runner, record runner, record = _scenario(scenario) ev = [e for e in _collected(events, record.id) if e["type"] == "stage_await_confirm"][0] assert ev["data"]["stage"] == "identify" assert ev["data"]["form_defaults"] # 表单默认值 assert runner.stage_status(record, "identify") == STAGE_DONE assert runner.stage_status(record, "phylogeny") == STAGE_WAITING_CONFIRM def test_confirm_unknown_param_rejected(db, events, catalog, fake_executor): """TC-BIO-API-058 对应服务层：confirm 拒绝非 schema 参数。""" async def scenario(): runner = make_runner(db, events, fake_executor, catalog) record = await runner.create_job( "s1", "t", ["identify"], auto_start=False) await runner.run_job(record) with pytest.raises(RunnerError, match="未知参数"): await runner.confirm_stage(record, "identify", {"magic": "x"}) return runner, record _scenario(scenario) def test_confirm_non_confirmable(db, events, catalog, fake_executor): """TC-BIO-API-059：重复 confirm 幂等拒绝。""" async def scenario(): runner = make_runner(db, events, fake_executor, catalog) record = await runner.create_job( "s1", "t", ["identify"], auto_continue=True, auto_start=False) await runner.run_job(record) assert runner.stage_status(record, "identify") == STAGE_DONE with pytest.raises(RunnerError, match="不可确认"): await runner.confirm_stage(record, "identify", {}) return runner, record _scenario(scenario) def test_auto_continue_skips_confirm(db, events, catalog, fake_executor): """TC-BIO-RUN-021：自动连续跳过确认全程推进。""" async def scenario(): runner = make_runner(db, events, fake_executor, catalog) record = await runner.create_job( "s1", "t", ["identify", "phylogeny"], auto_continue=True, auto_start=False) await runner.run_job(record) return runner, record runner, record = _scenario(scenario) assert record.status == JOB_DONE assert "stage_await_confirm" not in [e["type"] for e in _collected(events, record.id)] def test_concurrency_limit(db, events, catalog): """TC-BIO-RUN-010：并发上限排队（4 任务并发，同时 running ≤3）。""" executor = FakeExecutor(delay=0.1) async def scenario(): runner = make_runner(db, events, executor, catalog) records = [] for i in range(4): r = await runner.create_job( f"s{i}", f"t{i}", ["identify"], auto_continue=True, auto_start=False) records.append(r) await asyncio.gather(*[runner.run_job(r) for r in records]) return runner, records runner, records = _scenario(scenario) assert executor.max_running <= 3 assert all(r.status == JOB_DONE for r in records) def test_semaphore_released_on_failure(db, events, catalog): """TC-BIO-RUN-025：执行失败后信号量释放。""" executor = FakeExecutor(exit_code=1) async def scenario(): runner = make_runner(db, events, executor, catalog) record = await runner.create_job( "s1", "t", ["identify"], auto_continue=True, auto_start=False) await runner.run_job(record) return runner, record runner, record = _scenario(scenario) assert record.status == JOB_FAILED assert not runner.semaphore.locked() def test_restart_recovery_no_rerun(db, events, catalog, fake_executor): """TC-BIO-RUN-008：重启续跑不重跑已完成阶段（attempt 不变）。""" async def scenario(): runner = make_runner(db, events, fake_executor, catalog) record = await runner.create_job( "s1", "t", ["identify", "phylogeny"], auto_continue=True, auto_start=False) # 只完成第 1 阶段（模拟中断前状态） await runner._run_stage(record, "identify", 0) assert runner.stage_status(record, "identify") == STAGE_DONE # 模拟重启：新 runner 实例（进程重启后内存清空） runner2 = make_runner(db, events, fake_executor, catalog) recovered = await runner2.recover() await _drain(runner2) return runner2, record, recovered runner2, record, recovered = _scenario(scenario) assert any(r.id == record.id for r in recovered) from app.services.bioflow import store st = _scenario(lambda: store.get_stage(db, record.id, "identify")) assert st["status"] == STAGE_DONE assert st["attempt"] == 1 # 不重跑 def test_restart_recovers_inflight_by_manifest(db, events, catalog, fake_executor): """TC-BIO-RUN-009：重启恢复 in-flight running 阶段按 manifest 判成败。""" async def scenario(): runner = make_runner(db, events, fake_executor, catalog) record = await runner.create_job( "s1", "t", ["identify", "phylogeny"], auto_continue=True, auto_start=False) await runner._run_stage(record, "identify", 0) # 第 1 阶段完成 # phylogeny 标记 running 且无 manifest（模拟崩溃瞬间） record._stage_states["phylogeny"]["status"] = "running" await runner._persist_stage(record, "phylogeny") # 模拟重启 runner2 = make_runner(db, events, fake_executor, catalog) await runner2.recover() await _drain(runner2) # 同一 loop 内校验（aiosqlite 连接绑定本 loop） from app.services.bioflow import store st = await store.get_stage(db, record.id, "phylogeny") assert st["status"] == STAGE_DONE assert st["attempt"] >= 1 # 无 manifest → 重跑；attempt 至少 1 return runner2, record _scenario(scenario) def test_restart_skips_archived(db, events, catalog, fake_executor): """TC-BIO-RUN-030：恢复时跳过已归档任务。""" async def scenario(): runner = make_runner(db, events, fake_executor, catalog) record = await runner.create_job( "s1", "t", ["identify"], auto_continue=True, auto_start=False) await runner.run_job(record) # 归档（状态改为 archived） await runner._persist_job(record) from app.services.bioflow.store import update_job_status await update_job_status(db, record.id, "archived") runner2 = make_runner(db, events, fake_executor, catalog) recovered = await runner2.recover() return recovered recovered = _scenario(scenario) assert recovered == [] def test_empty_stages_done_immediately(db, events, catalog, fake_executor): """TC-BIO-RUN-039：空阶段任务直接 done，无 docker 调用。""" # 注意：创建任务要求 stages 非空（pydantic min_length=1）由路由层保证； # 服务层直接构造空链验证防御 async def scenario(): runner = make_runner(db, events, fake_executor, catalog) record = await runner.create_job( "s1", "t", ["identify"], auto_continue=True, auto_start=False) record.stage_chain = [] record._stage_states = {} await runner.run_job(record) return runner, record runner, record = _scenario(scenario) assert record.status == JOB_DONE assert fake_executor.calls == [] assert record.progress() == 100 def test_progress_calculation(db, events, catalog, fake_executor): """TC-BIO-RUN-029：任务进度数值。""" async def scenario(): runner = make_runner(db, events, fake_executor, catalog) record = await runner.create_job( "s1", "t", ["identify", "phylogeny", "motif"], auto_continue=True, auto_start=False) assert record.progress() == 0 # 全链跑完 → 100（进度公式按完成阶段数/总数） await runner.run_job(record) return runner, record runner, record = _scenario(scenario) assert record.progress() == 100 def test_unknown_stage_create_rejected(db, events, catalog, fake_executor): async def scenario(): runner = make_runner(db, events, fake_executor, catalog) with pytest.raises(RunnerError, match="未知阶段"): await runner.create_job("s1", "t", ["ghost_stage"], auto_start=False) return None _scenario(scenario) def test_prerequisite_autofill(db, events, catalog, fake_executor): """TC-BIO-RUN-034：前置补全后链完整（phylogeny 缺 identify → 自动补）。""" async def scenario(): runner = make_runner(db, events, fake_executor, catalog) record = await runner.create_job("s1", "t", ["phylogeny", "motif"], auto_start=False) return record record = _scenario(scenario) assert record.stage_chain[0] == "identify" assert set(record.stage_chain) == {"identify", "phylogeny", "motif"} def test_multi_job_isolation(db, events, catalog): """TC-BIO-RUN-028：多任务互不串扰。""" executor = FakeExecutor(log="job1 log\n") async def scenario(): runner = make_runner(db, events, executor, catalog) r1 = await runner.create_job("s1", "t1", ["identify"], auto_continue=True, auto_start=False) r2 = await runner.create_job("s2", "t2", ["identify"], auto_continue=True, auto_start=False) await asyncio.gather(runner.run_job(r1), runner.run_job(r2)) return runner, r1, r2 runner, r1, r2 = _scenario(scenario) assert r1.id != r2.id assert r1.status == JOB_DONE and r2.status == JOB_DONE assert r1._stage_states["identify"]["log_tail"] == "job1 log" # tail_lines 去尾换行 def test_stage_log_event(db, events, catalog, fake_executor): """TC-BIO-SSE-004：stage_log 事件随执行发布（log 非空）。""" async def scenario(): runner = make_runner(db, events, fake_executor, catalog) record = await runner.create_job( "s1", "t", ["identify"], auto_continue=True, auto_start=False) await runner.run_job(record) return runner, record runner, record = _scenario(scenario) log_events = [e for e in _collected(events, record.id) if e["type"] == "stage_log"] assert log_events assert all(e["data"]["log"] for e in log_events) def test_event_seq_strictly_increasing(db, events, catalog, fake_executor): """TC-BIO-SSE-013：seq 严格递增。""" async def scenario(): runner = make_runner(db, events, fake_executor, catalog) record = await runner.create_job( "s1", "t", ["identify"], auto_continue=True, auto_start=False) await runner.run_job(record) return runner, record runner, record = _scenario(scenario) seqs = [e["seq"] for e in _collected(events, record.id)] assert seqs == sorted(seqs) assert len(set(seqs)) == len(seqs) # --------------------------------------------------------------------------- # ensure_samples_yaml 配对修复（2026-09-17 水稻 M34 真实缺陷回归） # --------------------------------------------------------------------------- def _mk_fq(dirp, name): p = dirp / name p.write_text(">seq\nACGT\n") return p def test_samples_pairing_symlink_alias_dedup(tmp_path): """TC-BIO-FIX-01：目录含 f1/r2 真实文件 + _1/_2 符号链接别名时， 只产出 N 个配对样本，无 _r2.fq 孤儿条目，无 _2 别名抢配。""" from app.services.bioflow.runner import ensure_samples_yaml src = tmp_path / "fastq" src.mkdir() for sid in ("CRR1", "CRR2", "CRR3"): sub = src / sid sub.mkdir() real_r1 = _mk_fq(sub, f"{sid}_f1.fq.gz") real_r2 = _mk_fq(sub, f"{sid}_r2.fq.gz") # 别名符号链接（指向真实文件；模拟 M34 水稻目录实况） (sub / f"{sid}_1.fq.gz").symlink_to(real_r1) (sub / f"{sid}_2.fq.gz").symlink_to(real_r2) job_in = tmp_path / "job" / "input" job_in.mkdir(parents=True) mounts = ensure_samples_yaml(job_in, source_roots=[src]) import yaml samples = yaml.safe_load((job_in / "samples.yaml").read_text())["samples"] assert len(samples) == 3, f"期望 3 个配对样本,实际 {list(samples)}" assert all(not n.endswith("_r2.fq") for n in samples) for sid in ("CRR1", "CRR2", "CRR3"): s = samples[sid] assert s["r1"].endswith(f"{sid}_f1.fq.gz") assert s["r2"].endswith(f"{sid}_r2.fq.gz") assert mounts def test_samples_pairing_real_r2_not_orphaned(tmp_path): """TC-BIO-FIX-02：真实 _r2 与别名 _2 并存时，r2 必须取真实文件， 且 _2 别名不产生孤儿样本条目。""" from app.services.bioflow.runner import ensure_samples_yaml src = tmp_path / "fastq" src.mkdir() sub = src / "S1" sub.mkdir() real_r1 = _mk_fq(sub, "S1_f1.fq.gz") real_r2 = _mk_fq(sub, "S1_r2.fq.gz") (sub / "S1_1.fq.gz").symlink_to(real_r1) (sub / "S1_2.fq.gz").symlink_to(real_r2) job_in = tmp_path / "job" / "input" job_in.mkdir(parents=True) ensure_samples_yaml(job_in, source_roots=[src]) import yaml samples = yaml.safe_load((job_in / "samples.yaml").read_text())["samples"] assert len(samples) == 1, f"期望 1 个配对样本,实际 {list(samples)}" s = samples["S1"] assert s["r1"].endswith("S1_f1.fq.gz") assert s["r2"].endswith("S1_r2.fq.gz") def test_samples_pairing_unpaired_r2_kept(tmp_path): """TC-BIO-FIX-03：只有单端 r2（无 r1）时保留孤儿条目（原契约）。""" from app.services.bioflow.runner import ensure_samples_yaml src = tmp_path / "fastq" src.mkdir() _mk_fq(src, "WX_alone_r2.fq.gz") job_in = tmp_path / "job" / "input" job_in.mkdir(parents=True) ensure_samples_yaml(job_in, source_roots=[src]) import yaml samples = yaml.safe_load((job_in / "samples.yaml").read_text())["samples"] assert len(samples) == 1 s = next(iter(samples.values())) assert not s.get("r1") assert s["r2"].endswith("WX_alone_r2.fq.gz") # --------------------------------------------------------------------------- # 阶段进度条（2026-09-17：流式日志 + [PROGRESS] 标记解析 + stage_progress 事件） # --------------------------------------------------------------------------- def test_parse_progress_marker_table(): """parse_progress_marker 表驱动:样本/拷贝标记识别,无关行 None。""" from app.services.bioflow.runner import parse_progress_marker cases = [ ("[PROGRESS] Processing: SRR1", {"kind": "sample", "message": "SRR1"}), (" [PROGRESS] Processing: sample A ", {"kind": "sample", "message": "sample A"}), ("[PROGRESS] Copying results to final directory...", {"kind": "copy", "percent": 95}), ("[PROGRESS] Copying results to final", {"kind": "copy", "percent": 95}), ("[INFO] not progress", None), ("Processing: SRR1", None), # 无标记前缀不识别 ("", None), ] for line, expected in cases: assert parse_progress_marker(line) == expected, f"line={line!r}" class FakeStreamingExecutor: """mock docker 流式执行器：记录 on_log 回调并在调用中分块推送进度标记。""" def __init__(self, chunks: list[str], exit_code: int = 0): self.chunks = chunks self.exit_code = exit_code self.calls: list[dict] = [] self.on_log = None async def __call__(self, stage, input_dir, workspace, params=None, log_path=None, **kwargs): self.calls.append({"stage": stage}) self.on_log = kwargs.get("on_log") for chunk in self.chunks: if self.on_log is not None: await self.on_log(chunk) # 写成功 manifest(默认 FakeExecutor 语义:退出码 0 → success) if self.exit_code == 0: from conftest import FakeExecutor FakeExecutor._write_manifest(self, workspace, stage, "success") return {"exit_code": self.exit_code, "timed_out": False, "log_tail": "".join(self.chunks), "cmd": []} def test_stage_progress_events_rnaseq(db, events, catalog): """TC-BIO-PROG-01：rnaseq_qc 白名单阶段收到 [PROGRESS] → 发布 stage_progress(percent=seen/total,单调递增);成功后强制 100。""" chunks = [ "Input validation passed: 2 sample(s) ready\n", "Processing 2 samples...\n", "[PROGRESS] Processing: A\n", "fastp done\n", "[PROGRESS] Processing: B\n", ] executor = FakeStreamingExecutor(chunks) async def scenario(): runner = make_runner(db, events, executor, catalog) record = await runner.create_job( "s1", "t", ["rnaseq_qc"], auto_continue=True, auto_start=False) await runner.run_job(record) return runner, record runner, record = _scenario(scenario) assert runner.stage_status(record, "rnaseq_qc") == STAGE_DONE prog = [e["data"] for e in _collected(events, record.id) if e["type"] == "stage_progress"] assert prog, "应发布 stage_progress 事件" percents = [p["percent"] for p in prog] assert percents[-1] == 100 # 终态强制 100 assert percents[:-1] == sorted(percents[:-1]) # 中间值单调 assert any(p["percent"] > 0 for p in prog) # 有实际进度(非只有 0/100) assert prog[-1]["message"] == "" # 终态清空 message def test_stage_progress_not_for_non_rnaseq(db, events, catalog): """TC-BIO-PROG-02：非白名单阶段(identify)不发布 stage_progress。""" chunks = ["Processing 5 samples...\n", "[PROGRESS] Processing: A\n"] executor = FakeStreamingExecutor(chunks) async def scenario(): runner = make_runner(db, events, executor, catalog) record = await runner.create_job( "s1", "t", ["identify"], auto_continue=True, auto_start=False) await runner.run_job(record) return runner, record runner, record = _scenario(scenario) assert runner.stage_status(record, "identify") == STAGE_DONE prog = [e for e in _collected(events, record.id) if e["type"] == "stage_progress"] assert prog == [] def test_incremental_stage_log_events(db, events, catalog): """TC-BIO-PROG-03：on_log 分块 → 增量 stage_log 发布 + 终态 log_tail 并存。""" executor = FakeStreamingExecutor(["chunk1\n", "chunk2\n"]) async def scenario(): runner = make_runner(db, events, executor, catalog) record = await runner.create_job( "s1", "t", ["identify"], auto_continue=True, auto_start=False) await runner.run_job(record) return runner, record runner, record = _scenario(scenario) log_events = [e["data"] for e in _collected(events, record.id) if e["type"] == "stage_log"] assert log_events joined = "".join(e["log"] for e in log_events) assert "chunk1" in joined and "chunk2" in joined assert record._stage_states["identify"]["log_tail"] # 终态 log_tail 仍在