cat-shark
dcdc5e8604
批量引擎改为「分块流水线」：视频按 WOV_BATCH_STAGE_GROUP_SIZE（默认 8）分组， 组内按 DAG 拓扑序跑完全部视频（全部 extract → 全部 ASR → 全部翻译 → 全部 ASS） 再进入下一组，本地模型每组只加载一次、卸载一次，而不是每个视频来回加载卸载； 产物仍按组增量落到视频旁。调度器新增 execute_run(run_id, stop_after=节点)： 该节点完成后任务保持 RUNNING 不收尾，下一次调用从产物表跳过已完成节点继续， 用于实现阶段边界。 - nodes/llm.py：翻译节点结束释放本机 Ollama 显存（node 参数 unload_after > LLM_UNLOAD_AFTER > 本机 loopback 端点默认卸载，云端端点不卸载；卸载失败只告警）， 新增 keep_model.flag 语义（阶段内保持常驻）与 release_local_model()； 新增节点内暂停（按批 20 行检查 paused.flag，抛 PauseRequested，调度器保持 PAUSED）。 - src/wov_app/batch.py：分组阶段执行与阶段末统一释放显存；失败视频只在它失败 节点的那个阶段重试（避免 LLM 已常驻时重跑 ASR 抢显存）；任务没有明细时保持 QUEUED 等登记完成、仍有未完成视频时置回 QUEUED 自愈（原先留 RUNNING 会卡死： 引擎只拾取 QUEUED，任务停在“运行中但没人推进”）；无失败视频时删除任务级空目录； 每个阶段开始前清理 paused.flag / keep_model.flag，避免强杀残留影响后续阶段。 - src/wov_app/config.py：新增 WOV_BATCH_STAGE_GROUP_SIZE（设为 1 即旧的每视频全链路）。 - 任务列表与批量页分工：GET /api/runs 默认排除 source=batch（一个批量任务会产生 N 条单视频 run，会把 20 条窗口占满；且任务管理页的暂停/重试/删除对批量 run 语义不成立），需要排查时用 include_batch=1；作为补偿批量页详情新增阶段列 （阶段 i/N · 中文标签，由该视频 run 的 current_node_id 在 DAG 拓扑序中的位置 推导，节点类型映射中文标签）。阶段只有节点边界粒度，句级进度不落库、只在日志。 - 顺带纳入此前未提交的批量僵尸状态恢复：recover_interrupted_batch_jobs 除 RUNNING 外也把「COMPLETED 但仍含未结束视频」的任务置回 QUEUED；fix_zombie_batch_jobs.py 改为按条件扫描并支持 --apply 预览；批量页明细只列本批真正处理过的视频。 测试新增/更新：分块流水线调用顺序（组内按节点跑完再下一组）、每组只释放一次模型、 阶段内保持常驻标志、翻译按批暂停、失败视频不跨阶段推进、任务无明细/中途登记视频时 置回 QUEUED、任务工作空间与残留信号清理、任务列表默认过滤批量 run、详情阶段字段、 前端阶段列渲染；全量 507 passed（唯一失败为既有素材缺失的 integration 用例）。
561 lines
21 KiB
Python
561 lines
21 KiB
Python
"""src/wov_app/scheduler.py 的模块级测试（数据 → 测试过程 → 验证结果）。
|
||
|
||
被测模块：`src/wov_app/scheduler.py`（DAG 拓扑调度、断点续跑、暂停语义），
|
||
可独立调用。用例使用真实 SQLite、真实存储目录与真实节点（echo/echo 派生），
|
||
仅对需要外部服务的节点不做测试。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from wov_app import registry
|
||
from wov_app.db import Database
|
||
from wov_app.scheduler import WorkflowScheduler, topological_sort
|
||
from wov_sdk.models import WorkflowDefinition
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _isolate_registry():
|
||
"""用例前清空注册表、用例后恢复快照：保证用例看到的是干净基线，
|
||
不受其他模块（如 main 生命周期 register_all）的注册结果影响。"""
|
||
snapshot = dict(registry._registry)
|
||
registry._registry.clear()
|
||
yield
|
||
registry._registry.clear()
|
||
registry._registry.update(snapshot)
|
||
|
||
|
||
def _definition(nodes: list[dict], edges: list[dict], **extra) -> WorkflowDefinition:
|
||
"""构造真实 WorkflowDefinition（节点 ID/类型/输入与边）。"""
|
||
payload = {
|
||
"name": "测试流程",
|
||
"version": 1,
|
||
"nodes": nodes,
|
||
"edges": edges,
|
||
"entry_inputs": {"video_uri": "file"},
|
||
"final_outputs": extra.pop("final_outputs", {}),
|
||
**extra,
|
||
}
|
||
return WorkflowDefinition.from_dict(payload)
|
||
|
||
|
||
def _db_with_workflow(tmp_path: Path, definition: WorkflowDefinition, workflow_id: str = "wf") -> Database:
|
||
"""建好工作流 + 版本记录的临时库（任务表有外键约束）。"""
|
||
db = Database(tmp_path / "wov.db")
|
||
db.upsert_workflow({"id": workflow_id, "name": "流程", "description": ""})
|
||
db.create_workflow_version(workflow_id, 1, definition.to_dict())
|
||
return db
|
||
|
||
|
||
def _run(run_id: str, input_uri: str, **overrides) -> dict:
|
||
"""构造真实任务记录。"""
|
||
record = {
|
||
"id": run_id, "workflow_id": "wf", "workflow_version": 1, "status": "QUEUED",
|
||
"current_node_id": None, "progress": 0.0, "error": None, "input_uri": input_uri,
|
||
"param_overrides": None, "source": "upload",
|
||
"created_at": "2026-09-01T00:00:00+00:00", "updated_at": "2026-09-01T00:00:00+00:00",
|
||
}
|
||
record.update(overrides)
|
||
return record
|
||
|
||
|
||
def _scheduler(db: Database, storage: Path) -> WorkflowScheduler:
|
||
"""构造调度器（不启动后台线程，直接调用 execute_run）。"""
|
||
return WorkflowScheduler(db, storage, interval_seconds=999)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 拓扑排序
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_topological_sort_linear_chain() -> None:
|
||
"""线性链按依赖顺序返回。"""
|
||
# 数据：a → b → c。
|
||
definition = _definition(
|
||
nodes=[
|
||
{"id": "a", "node_type": "echo", "inputs": {}},
|
||
{"id": "b", "node_type": "echo", "inputs": {"file_uri": "a.file_uri"}},
|
||
{"id": "c", "node_type": "echo", "inputs": {"file_uri": "b.file_uri"}},
|
||
],
|
||
edges=[{"from": "a", "to": "b"}, {"from": "b", "to": "c"}],
|
||
)
|
||
|
||
# 测试过程
|
||
order = topological_sort(definition)
|
||
|
||
# 验证结果
|
||
assert order == ["a", "b", "c"]
|
||
|
||
|
||
def test_topological_sort_diamond() -> None:
|
||
"""菱形依赖中，汇合节点排在其全部前驱之后。"""
|
||
# 数据：a → (b, c) → d。
|
||
definition = _definition(
|
||
nodes=[
|
||
{"id": "a", "node_type": "echo", "inputs": {}},
|
||
{"id": "b", "node_type": "echo", "inputs": {"file_uri": "a.file_uri"}},
|
||
{"id": "c", "node_type": "echo", "inputs": {"file_uri": "a.file_uri"}},
|
||
{"id": "d", "node_type": "echo", "inputs": {"file_uri": "b.file_uri"}},
|
||
],
|
||
edges=[
|
||
{"from": "a", "to": "b"}, {"from": "a", "to": "c"},
|
||
{"from": "b", "to": "d"}, {"from": "c", "to": "d"},
|
||
],
|
||
)
|
||
|
||
# 测试过程
|
||
order = topological_sort(definition)
|
||
|
||
# 验证结果：a 最先、d 最后，b/c 在中间。
|
||
assert order[0] == "a"
|
||
assert order[-1] == "d"
|
||
assert set(order[1:3]) == {"b", "c"}
|
||
|
||
|
||
def test_execute_run_stop_after_leaves_run_running_and_resumes(tmp_path: Path) -> None:
|
||
"""分阶段执行：stop_after 指定阶段节点后停下（保持 RUNNING、不收尾），再次调用续跑完成。"""
|
||
# 数据：a → b → c 三段 echo 链 + 最终别名。
|
||
definition = _definition(
|
||
nodes=[
|
||
{"id": "a", "node_type": "echo", "inputs": {"file_uri": "input.video_uri"}},
|
||
{"id": "b", "node_type": "echo", "inputs": {"file_uri": "a.file_uri"}},
|
||
{"id": "c", "node_type": "echo", "inputs": {"file_uri": "b.file_uri"}},
|
||
],
|
||
edges=[{"from": "a", "to": "b"}, {"from": "b", "to": "c"}],
|
||
final_outputs={"result": "c.file_uri"},
|
||
)
|
||
storage = tmp_path / "storage"
|
||
db = _db_with_workflow(tmp_path, definition)
|
||
source = tmp_path / "input.txt"
|
||
source.write_text("分阶段内容", encoding="utf-8")
|
||
db.create_run(_run("run-stage", str(source)))
|
||
registry.register_all()
|
||
scheduler = _scheduler(db, storage)
|
||
|
||
# 测试过程：第一阶段只执行到 b 为止。
|
||
scheduler.execute_run("run-stage", stop_after="b")
|
||
|
||
# 验证结果：任务保持 RUNNING 未收尾，a/b 产物已登记，c 与最终别名都没有。
|
||
staged = db.get_run("run-stage")
|
||
assert staged["status"] == "RUNNING"
|
||
assert staged["current_node_id"] == "b"
|
||
names = {artifact["name"] for artifact in db.list_artifacts("run-stage")}
|
||
assert {"a.file_uri", "b.file_uri"} <= names
|
||
assert "c.file_uri" not in names
|
||
assert "result" not in names
|
||
|
||
# 测试过程：不传 stop_after 时整条 DAG 跑完并收尾。
|
||
scheduler.execute_run("run-stage")
|
||
|
||
# 验证结果：完成、进度 1.0、最终别名登记。
|
||
finished = db.get_run("run-stage")
|
||
assert finished["status"] == "COMPLETED"
|
||
assert finished["progress"] == 1.0
|
||
assert "result" in {artifact["name"] for artifact in db.list_artifacts("run-stage")}
|
||
|
||
|
||
def test_execute_run_stop_after_unknown_node_marks_failed(tmp_path: Path) -> None:
|
||
"""stop_after 指向不存在的节点时任务标 FAILED（不留下永远 RUNNING 的任务）。"""
|
||
# 数据：单 echo 节点任务。
|
||
definition = _definition(
|
||
nodes=[{"id": "step", "node_type": "echo", "inputs": {"file_uri": "input.video_uri"}}],
|
||
edges=[],
|
||
final_outputs={"result": "step.file_uri"},
|
||
)
|
||
storage = tmp_path / "storage"
|
||
db = _db_with_workflow(tmp_path, definition)
|
||
source = tmp_path / "input.txt"
|
||
source.write_text("内容", encoding="utf-8")
|
||
db.create_run(_run("run-bad-stage", str(source)))
|
||
registry.register_all()
|
||
|
||
# 测试过程
|
||
_scheduler(db, storage).execute_run("run-bad-stage", stop_after="nope")
|
||
|
||
# 验证结果：FAILED 且错误说明阶段节点不存在。
|
||
stored = db.get_run("run-bad-stage")
|
||
assert stored["status"] == "FAILED"
|
||
assert "stop_after" in (stored["error"] or "")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 执行：成功路径
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_execute_run_completes_and_records_artifacts(tmp_path: Path) -> None:
|
||
"""单节点任务执行成功：状态 COMPLETED、产物登记、进度到位。"""
|
||
# 数据：echo 单节点工作流 + 真实输入文件。
|
||
definition = _definition(
|
||
nodes=[{"id": "step", "node_type": "echo", "inputs": {"file_uri": "input.video_uri"}}],
|
||
edges=[],
|
||
final_outputs={"result": "step.file_uri"},
|
||
)
|
||
storage = tmp_path / "storage"
|
||
db = _db_with_workflow(tmp_path, definition)
|
||
source = tmp_path / "input.txt"
|
||
source.write_text("输入内容", encoding="utf-8")
|
||
db.create_run(_run("run-1", str(source)))
|
||
registry.register_all()
|
||
|
||
# 测试过程
|
||
_scheduler(db, storage).execute_run("run-1")
|
||
|
||
# 验证结果：任务完成、进度 1.0、节点产物与最终别名都已登记。
|
||
stored = db.get_run("run-1")
|
||
assert stored["status"] == "COMPLETED"
|
||
assert stored["progress"] == 1.0
|
||
names = {a["name"] for a in db.list_artifacts("run-1")}
|
||
assert "step.file_uri" in names
|
||
assert "result" in names
|
||
|
||
|
||
def test_execute_run_multi_node_chain_passes_artifacts(tmp_path: Path) -> None:
|
||
"""多节点链：后序节点通过 URI 拿到前序产物（节点间只经产物交换数据）。"""
|
||
# 数据：step1 → step2 两节点链。
|
||
definition = _definition(
|
||
nodes=[
|
||
{"id": "step1", "node_type": "echo", "inputs": {"file_uri": "input.video_uri"}},
|
||
{"id": "step2", "node_type": "echo", "inputs": {"file_uri": "step1.file_uri"}},
|
||
],
|
||
edges=[{"from": "step1", "to": "step2"}],
|
||
final_outputs={"out": "step2.file_uri"},
|
||
)
|
||
storage = tmp_path / "storage"
|
||
db = _db_with_workflow(tmp_path, definition)
|
||
source = tmp_path / "input.txt"
|
||
source.write_text("链式内容", encoding="utf-8")
|
||
db.create_run(_run("run-2", str(source)))
|
||
registry.register_all()
|
||
|
||
# 测试过程
|
||
_scheduler(db, storage).execute_run("run-2")
|
||
|
||
# 验证结果：两节点产物都存在，且 step2 的产物内容来自 step1（传递一致）。
|
||
assert db.get_run("run-2")["status"] == "COMPLETED"
|
||
artifacts = {a["name"]: a["uri"] for a in db.list_artifacts("run-2")}
|
||
assert "step1.file_uri" in artifacts and "step2.file_uri" in artifacts
|
||
assert Path(artifacts["step2.file_uri"]).read_text(encoding="utf-8") == "链式内容"
|
||
assert Path(artifacts["step2.file_uri"]).parent != Path(artifacts["step1.file_uri"]).parent
|
||
|
||
|
||
def test_execute_run_creates_final_alias_with_stable_name(tmp_path: Path) -> None:
|
||
"""最终产物按 上传文件名.标识.时间戳 生成别名，并保留节点原始文件。"""
|
||
# 数据：单节点 + target_language 参数（决定别名标识）。
|
||
definition = _definition(
|
||
nodes=[{
|
||
"id": "step", "node_type": "echo",
|
||
"inputs": {"file_uri": "input.video_uri"},
|
||
"params": {"target_language": "zh-CN"},
|
||
}],
|
||
edges=[],
|
||
final_outputs={"cn_srt_uri": "step.file_uri"},
|
||
)
|
||
storage = tmp_path / "storage"
|
||
db = _db_with_workflow(tmp_path, definition)
|
||
source = tmp_path / "test01.mp4"
|
||
source.write_text("数据", encoding="utf-8")
|
||
db.create_run(_run("run-3", str(source)))
|
||
registry.register_all()
|
||
|
||
# 测试过程
|
||
_scheduler(db, storage).execute_run("run-3")
|
||
|
||
# 验证结果：别名指向 finals 下的稳定路径，含 zh-CN 标识；节点原文件仍在。
|
||
artifacts = {a["name"]: a["uri"] for a in db.list_artifacts("run-3")}
|
||
final = Path(artifacts["cn_srt_uri"])
|
||
assert final.is_file()
|
||
assert "zh-CN" in final.name
|
||
assert "finals" in final.parts
|
||
assert Path(artifacts["step.file_uri"]).is_file()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 失败与无效 DAG
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_execute_run_fails_when_workflow_version_missing(tmp_path: Path) -> None:
|
||
"""工作流版本记录丢失时任务失败（不留 QUEUED 堵塞队列）。"""
|
||
# 数据：工作流存在但没有 v1 版本记录（外键仍满足）。
|
||
db = Database(tmp_path / "wov.db")
|
||
storage = tmp_path / "storage"
|
||
db.upsert_workflow({"id": "wf-no-version", "name": "流程", "description": ""})
|
||
db.create_run(_run("run-x", "input", workflow_id="wf-no-version"))
|
||
|
||
# 测试过程
|
||
_scheduler(db, storage).execute_run("run-x")
|
||
|
||
# 验证结果
|
||
stored = db.get_run("run-x")
|
||
assert stored["status"] == "FAILED"
|
||
assert "workflow version not found" in stored["error"]
|
||
|
||
|
||
def test_execute_run_marks_failed_on_cycle_and_does_not_block_queue(tmp_path: Path) -> None:
|
||
"""环形 DAG（历史无效版本）立即失败且不堵塞后续任务（R04 回归）。"""
|
||
# 数据：A→B→A 的环形定义。
|
||
definition = _definition(
|
||
nodes=[
|
||
{"id": "a", "node_type": "echo", "inputs": {"file_uri": "b.file_uri"}},
|
||
{"id": "b", "node_type": "echo", "inputs": {"file_uri": "a.file_uri"}},
|
||
],
|
||
edges=[{"from": "a", "to": "b"}, {"from": "b", "to": "a"}],
|
||
)
|
||
storage = tmp_path / "storage"
|
||
db = _db_with_workflow(tmp_path, definition)
|
||
db.create_run(_run("run-cycle", "input"))
|
||
registry.register_all()
|
||
|
||
# 测试过程
|
||
_scheduler(db, storage).execute_run("run-cycle")
|
||
|
||
# 验证结果：任务 FAILED，且队首前移（不再返回该任务）。
|
||
assert db.get_run("run-cycle")["status"] == "FAILED"
|
||
assert db.next_queued_run() is None
|
||
|
||
|
||
def test_execute_run_fails_when_input_reference_missing(tmp_path: Path) -> None:
|
||
"""输入引用无法解析（前序产物缺失）时任务失败并记录原因。"""
|
||
# 数据：节点引用不存在的产物。
|
||
definition = _definition(
|
||
nodes=[{"id": "step", "node_type": "echo", "inputs": {"file_uri": "ghost.file_uri"}}],
|
||
edges=[],
|
||
)
|
||
storage = tmp_path / "storage"
|
||
db = _db_with_workflow(tmp_path, definition)
|
||
db.create_run(_run("run-4", "input"))
|
||
registry.register_all()
|
||
|
||
# 测试过程
|
||
_scheduler(db, storage).execute_run("run-4")
|
||
|
||
# 验证结果
|
||
stored = db.get_run("run-4")
|
||
assert stored["status"] == "FAILED"
|
||
assert "missing input" in stored["error"]
|
||
|
||
|
||
def test_execute_run_fails_when_node_returns_failed(tmp_path: Path) -> None:
|
||
"""节点返回 failed 时任务失败并保留错误信息。"""
|
||
# 数据：注册一个总是失败的节点。
|
||
from wov_sdk.models import InvokeResponse, NodeManifest
|
||
|
||
definition = _definition(
|
||
nodes=[{"id": "step", "node_type": "boom", "inputs": {}}],
|
||
edges=[],
|
||
)
|
||
storage = tmp_path / "storage"
|
||
db = _db_with_workflow(tmp_path, definition)
|
||
db.create_run(_run("run-5", "input"))
|
||
registry.register(
|
||
NodeManifest(id="boom", name="失败节点", version="1", capability="c", command=["python"]),
|
||
lambda request: InvokeResponse(status="failed", error="节点内部错误"),
|
||
)
|
||
|
||
# 测试过程
|
||
_scheduler(db, storage).execute_run("run-5")
|
||
|
||
# 验证结果
|
||
stored = db.get_run("run-5")
|
||
assert stored["status"] == "FAILED"
|
||
assert "节点内部错误" in stored["error"]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 暂停语义
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_execute_run_leaves_paused_run_untouched(tmp_path: Path) -> None:
|
||
"""以 PAUSED 进入时直接返回保持暂停（修复"点击暂停反而开始任务"）。"""
|
||
# 数据：PAUSED 状态的任务。
|
||
definition = _definition(
|
||
nodes=[{"id": "step", "node_type": "echo", "inputs": {}}],
|
||
edges=[],
|
||
)
|
||
storage = tmp_path / "storage"
|
||
db = _db_with_workflow(tmp_path, definition)
|
||
db.create_run(_run("run-6", "input", status="PAUSED"))
|
||
registry.register_all()
|
||
|
||
# 测试过程
|
||
_scheduler(db, storage).execute_run("run-6")
|
||
|
||
# 验证结果：仍为 PAUSED，且未产生任何产物。
|
||
assert db.get_run("run-6")["status"] == "PAUSED"
|
||
assert db.list_artifacts("run-6") == []
|
||
|
||
|
||
def test_execute_run_stops_at_node_boundary_when_paused(tmp_path: Path) -> None:
|
||
"""运行中被暂停：在当前节点边界停下保持 PAUSED，不标 FAILED。"""
|
||
# 数据：两节点链；第一个节点执行时把任务置 PAUSED。
|
||
from wov_sdk.models import InvokeResponse, NodeManifest
|
||
|
||
definition = _definition(
|
||
nodes=[
|
||
{"id": "step1", "node_type": "pause-me", "inputs": {}},
|
||
{"id": "step2", "node_type": "echo", "inputs": {}},
|
||
],
|
||
edges=[{"from": "step1", "to": "step2"}],
|
||
)
|
||
storage = tmp_path / "storage"
|
||
db = _db_with_workflow(tmp_path, definition)
|
||
db.create_run(_run("run-7", "input"))
|
||
registry.register_all()
|
||
|
||
def pause_handler(request):
|
||
"""模拟用户在该节点执行期间点击暂停。"""
|
||
db.pause_run("run-7", "t2")
|
||
return InvokeResponse(status="completed", outputs={"file_uri": "/tmp/x"})
|
||
|
||
registry.register(
|
||
NodeManifest(id="pause-me", name="暂停节点", version="1", capability="c", command=["python"]),
|
||
pause_handler,
|
||
)
|
||
|
||
# 测试过程
|
||
_scheduler(db, storage).execute_run("run-7")
|
||
|
||
# 验证结果：保持 PAUSED，step2 未执行。
|
||
assert db.get_run("run-7")["status"] == "PAUSED"
|
||
names = {a["name"] for a in db.list_artifacts("run-7")}
|
||
assert not any(name.startswith("step2") for name in names)
|
||
|
||
|
||
def test_execute_run_clears_stale_pause_flag(tmp_path: Path) -> None:
|
||
"""执行前清理残留的 paused.flag（避免误触发节点内暂停）。"""
|
||
# 数据：单节点任务 + 已存在的 paused.flag。
|
||
definition = _definition(
|
||
nodes=[{"id": "step", "node_type": "echo", "inputs": {}}],
|
||
edges=[],
|
||
)
|
||
storage = tmp_path / "storage"
|
||
db = _db_with_workflow(tmp_path, definition)
|
||
db.create_run(_run("run-8", "input"))
|
||
registry.register_all()
|
||
run_root = storage / "runs" / "run-8"
|
||
run_root.mkdir(parents=True)
|
||
(run_root / "paused.flag").write_text("", encoding="utf-8")
|
||
|
||
# 测试过程
|
||
_scheduler(db, storage).execute_run("run-8")
|
||
|
||
# 验证结果：标志被清除，任务正常完成。
|
||
assert not (run_root / "paused.flag").exists()
|
||
assert db.get_run("run-8")["status"] == "COMPLETED"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 断点续跑与参数覆盖
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_execute_run_resumes_from_existing_artifacts(tmp_path: Path) -> None:
|
||
"""断点续跑：已有产物的节点被跳过，只执行剩余节点。"""
|
||
# 数据：两节点链，step1 产物已登记。
|
||
calls: list[str] = []
|
||
|
||
def tracking_handler(node_id: str):
|
||
def handler(request):
|
||
from wov_sdk.models import InvokeResponse
|
||
|
||
calls.append(node_id)
|
||
return InvokeResponse(status="completed", outputs={"file_uri": f"/tmp/{node_id}"})
|
||
|
||
return handler
|
||
|
||
definition = _definition(
|
||
nodes=[
|
||
{"id": "step1", "node_type": "track1", "inputs": {}},
|
||
{"id": "step2", "node_type": "track2", "inputs": {}},
|
||
],
|
||
edges=[{"from": "step1", "to": "step2"}],
|
||
)
|
||
storage = tmp_path / "storage"
|
||
db = _db_with_workflow(tmp_path, definition)
|
||
db.create_run(_run("run-9", "input"))
|
||
db.create_artifact({
|
||
"run_id": "run-9", "node_id": "step1", "name": "step1.file_uri",
|
||
"uri": "/tmp/step1", "kind": "file",
|
||
})
|
||
from wov_sdk.models import NodeManifest
|
||
|
||
registry.register(
|
||
NodeManifest(id="track1", name="t1", version="1", capability="c", command=["python"]),
|
||
tracking_handler("step1"),
|
||
)
|
||
registry.register(
|
||
NodeManifest(id="track2", name="t2", version="1", capability="c", command=["python"]),
|
||
tracking_handler("step2"),
|
||
)
|
||
|
||
# 测试过程
|
||
_scheduler(db, storage).execute_run("run-9")
|
||
|
||
# 验证结果：只调用了 step2。
|
||
assert calls == ["step2"]
|
||
assert db.get_run("run-9")["status"] == "COMPLETED"
|
||
|
||
|
||
def test_execute_run_applies_param_overrides(tmp_path: Path) -> None:
|
||
"""param_overrides 按节点 ID 合并进节点参数（前端框选 crop 的通道）。"""
|
||
# 数据：节点参数与覆盖值同时存在。
|
||
received: list[dict] = []
|
||
|
||
def handler(request):
|
||
from wov_sdk.models import InvokeResponse
|
||
|
||
received.append(dict(request.params))
|
||
return InvokeResponse(status="completed", outputs={"file_uri": "/tmp/x"})
|
||
|
||
definition = _definition(
|
||
nodes=[{
|
||
"id": "step", "node_type": "param-node",
|
||
"inputs": {}, "params": {"interval_seconds": 0.5, "crop": [0, 0, 1, 1]},
|
||
}],
|
||
edges=[],
|
||
)
|
||
storage = tmp_path / "storage"
|
||
db = _db_with_workflow(tmp_path, definition)
|
||
db.create_run(_run("run-10", "input", param_overrides={"step": {"crop": [0, 0.75, 1, 0.25]}}))
|
||
from wov_sdk.models import NodeManifest
|
||
|
||
registry.register(
|
||
NodeManifest(id="param-node", name="p", version="1", capability="c", command=["python"]),
|
||
handler,
|
||
)
|
||
|
||
# 测试过程
|
||
_scheduler(db, storage).execute_run("run-10")
|
||
|
||
# 验证结果：覆盖值生效，未覆盖的参数保持原样。
|
||
assert received == [{"interval_seconds": 0.5, "crop": [0, 0.75, 1, 0.25]}]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 线程生命周期
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def test_start_and_stop_are_idempotent(tmp_path: Path) -> None:
|
||
"""start 重复调用不产生多余线程；stop 正常结束。"""
|
||
# 数据：空库 + 调度器。
|
||
db = Database(tmp_path / "wov.db")
|
||
scheduler = WorkflowScheduler(db, tmp_path / "storage", interval_seconds=999)
|
||
|
||
# 测试过程
|
||
scheduler.start()
|
||
first = scheduler._thread
|
||
scheduler.start()
|
||
second = scheduler._thread
|
||
scheduler.stop()
|
||
|
||
# 验证结果
|
||
assert first is second
|
||
assert scheduler._thread is None
|