410 lines
13 KiB
Python
410 lines
13 KiB
Python
"""调度器单元测试。
|
|
|
|
覆盖拓扑排序、任务执行成功/失败分支、输入引用解析、MIME 推断以及
|
|
后台轮询线程的启动与停止。
|
|
"""
|
|
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app.db import Database
|
|
from app.node_manager import NodeManager
|
|
from app.scheduler import WorkflowScheduler, topological_sort
|
|
from wov_sdk.models import (
|
|
NodeManifest,
|
|
WorkflowDefinition,
|
|
WorkflowEdge,
|
|
WorkflowNode,
|
|
)
|
|
|
|
|
|
def _db(tmp_path) -> Database:
|
|
"""在临时目录创建独立数据库。"""
|
|
return Database(tmp_path / "wov.db")
|
|
|
|
|
|
def _echo_definition() -> WorkflowDefinition:
|
|
"""构造引用 Echo 节点的单步骤工作流定义。"""
|
|
return WorkflowDefinition(
|
|
name="echo-flow",
|
|
version=1,
|
|
nodes=[
|
|
WorkflowNode(
|
|
id="step",
|
|
node_type="echo",
|
|
inputs={"file_uri": "input.video_uri"},
|
|
)
|
|
],
|
|
edges=[],
|
|
entry_inputs={"video_uri": "file"},
|
|
final_outputs={"result": "step.file_uri"},
|
|
)
|
|
|
|
|
|
def test_topological_sort() -> None:
|
|
"""验证 DAG 排序保持依赖顺序，并拒绝环与未知边。"""
|
|
definition = WorkflowDefinition(
|
|
name="dag",
|
|
version=1,
|
|
nodes=[
|
|
WorkflowNode(id="a", node_type="x"),
|
|
WorkflowNode(id="b", node_type="x"),
|
|
WorkflowNode(id="c", node_type="x"),
|
|
],
|
|
edges=[
|
|
WorkflowEdge(from_node="a", to_node="b"),
|
|
WorkflowEdge(from_node="a", to_node="c"),
|
|
],
|
|
)
|
|
order = topological_sort(definition)
|
|
assert order.index("a") < order.index("b")
|
|
assert order.index("a") < order.index("c")
|
|
|
|
cycle = WorkflowDefinition(
|
|
name="cycle",
|
|
version=1,
|
|
nodes=[
|
|
WorkflowNode(id="a", node_type="x"),
|
|
WorkflowNode(id="b", node_type="x"),
|
|
],
|
|
edges=[
|
|
WorkflowEdge(from_node="a", to_node="b"),
|
|
WorkflowEdge(from_node="b", to_node="a"),
|
|
],
|
|
)
|
|
with pytest.raises(ValueError, match="cycle"):
|
|
topological_sort(cycle)
|
|
|
|
with pytest.raises(ValueError, match="unknown edge"):
|
|
topological_sort(
|
|
WorkflowDefinition(
|
|
name="bad",
|
|
version=1,
|
|
nodes=[WorkflowNode(id="a", node_type="x")],
|
|
edges=[WorkflowEdge(from_node="a", to_node="missing")],
|
|
)
|
|
)
|
|
|
|
|
|
def test_execute_echo_workflow(tmp_path) -> None:
|
|
"""验证排队任务可被完整执行并登记全部产物。"""
|
|
db = _db(tmp_path)
|
|
input_file = tmp_path / "input.txt"
|
|
input_file.write_text("hello scheduler", encoding="utf-8")
|
|
manifest = NodeManifest.load(
|
|
Path(__file__).resolve().parent.parent.parent / "wov-node-echo" / "node.manifest.json"
|
|
)
|
|
db.upsert_node(manifest)
|
|
db.upsert_workflow({"id": "flow", "name": "Flow", "published": 1, "latest_version": 1})
|
|
db.create_workflow_version("flow", 1, _echo_definition().to_dict())
|
|
now = "2026-01-01T00:00:00+00:00"
|
|
db.create_run(
|
|
{
|
|
"id": "run_1",
|
|
"workflow_id": "flow",
|
|
"workflow_version": 1,
|
|
"status": "QUEUED",
|
|
"progress": 0,
|
|
"input_uri": str(input_file),
|
|
"created_at": now,
|
|
"updated_at": now,
|
|
}
|
|
)
|
|
|
|
manager = NodeManager(db)
|
|
scheduler = WorkflowScheduler(db, manager, tmp_path / "storage")
|
|
try:
|
|
scheduler.execute_run("run_1")
|
|
finally:
|
|
manager.shutdown()
|
|
|
|
run = db.get_run("run_1")
|
|
assert run["status"] == "COMPLETED"
|
|
artifacts = db.list_artifacts("run_1")
|
|
assert {item["name"] for item in artifacts} == {"step.text", "step.file_uri", "result"}
|
|
|
|
|
|
def test_execute_run_missing_workflow(tmp_path, monkeypatch) -> None:
|
|
"""验证工作流记录缺失时任务被标记为失败。"""
|
|
db = _db(tmp_path)
|
|
db.upsert_workflow({"id": "flow", "name": "Flow", "published": 1, "latest_version": 1})
|
|
now = "2026-01-01T00:00:00+00:00"
|
|
db.create_run(
|
|
{
|
|
"id": "run_missing",
|
|
"workflow_id": "flow",
|
|
"workflow_version": 1,
|
|
"status": "QUEUED",
|
|
"progress": 0,
|
|
"created_at": now,
|
|
"updated_at": now,
|
|
}
|
|
)
|
|
monkeypatch.setattr(db, "get_workflow", lambda workflow_id: None)
|
|
manager = NodeManager(db)
|
|
scheduler = WorkflowScheduler(db, manager, tmp_path / "storage")
|
|
try:
|
|
scheduler.execute_run("run_missing")
|
|
finally:
|
|
manager.shutdown()
|
|
assert db.get_run("run_missing")["status"] == "FAILED"
|
|
|
|
|
|
def test_execute_run_missing_version(tmp_path) -> None:
|
|
"""验证版本记录缺失时任务被标记为失败。"""
|
|
db = _db(tmp_path)
|
|
db.upsert_workflow({"id": "flow", "name": "Flow", "published": 1, "latest_version": 1})
|
|
now = "2026-01-01T00:00:00+00:00"
|
|
db.create_run(
|
|
{
|
|
"id": "run_version",
|
|
"workflow_id": "flow",
|
|
"workflow_version": 1,
|
|
"status": "QUEUED",
|
|
"progress": 0,
|
|
"created_at": now,
|
|
"updated_at": now,
|
|
}
|
|
)
|
|
manager = NodeManager(db)
|
|
scheduler = WorkflowScheduler(db, manager, tmp_path / "storage")
|
|
try:
|
|
scheduler.execute_run("run_version")
|
|
finally:
|
|
manager.shutdown()
|
|
assert db.get_run("run_version")["status"] == "FAILED"
|
|
|
|
|
|
def test_execute_run_missing_node(tmp_path) -> None:
|
|
"""验证未注册节点被调用时任务失败。"""
|
|
db = _db(tmp_path)
|
|
definition = WorkflowDefinition(
|
|
name="bad",
|
|
version=1,
|
|
nodes=[
|
|
WorkflowNode(
|
|
id="step",
|
|
node_type="missing-node",
|
|
inputs={"text": "input.video_uri"},
|
|
)
|
|
],
|
|
entry_inputs={"video_uri": "file"},
|
|
)
|
|
db.upsert_workflow({"id": "flow", "name": "Flow", "published": 1, "latest_version": 1})
|
|
db.create_workflow_version("flow", 1, definition.to_dict())
|
|
now = "2026-01-01T00:00:00+00:00"
|
|
db.create_run(
|
|
{
|
|
"id": "run_node",
|
|
"workflow_id": "flow",
|
|
"workflow_version": 1,
|
|
"status": "QUEUED",
|
|
"progress": 0,
|
|
"input_uri": str(tmp_path / "in.txt"),
|
|
"created_at": now,
|
|
"updated_at": now,
|
|
}
|
|
)
|
|
manager = NodeManager(db)
|
|
scheduler = WorkflowScheduler(db, manager, tmp_path / "storage")
|
|
try:
|
|
scheduler.execute_run("run_node")
|
|
finally:
|
|
manager.shutdown()
|
|
assert db.get_run("run_node")["status"] == "FAILED"
|
|
|
|
|
|
def test_resolve_ref_and_mime(tmp_path) -> None:
|
|
"""验证输入引用解析、MIME 推断与文件大小读取。"""
|
|
db = _db(tmp_path)
|
|
manager = NodeManager(db)
|
|
scheduler = WorkflowScheduler(db, manager, tmp_path / "storage")
|
|
assert scheduler._resolve_ref("input.video", "in.mp4", {}) == "in.mp4"
|
|
assert (
|
|
scheduler._resolve_ref(
|
|
"a.out", None, {"a": {"out": "result.txt"}}
|
|
)
|
|
== "result.txt"
|
|
)
|
|
assert scheduler._resolve_ref("a.out", None, {}) is None
|
|
assert scheduler._resolve_ref("nodot", "in.mp4", {}) is None
|
|
assert scheduler._mime_type("x.srt") == "application/x-subrip"
|
|
assert scheduler._mime_type("x.ass") == "text/plain"
|
|
assert scheduler._mime_type("x.wav") == "audio/wav"
|
|
assert scheduler._mime_type("x.mp4") == "video/mp4"
|
|
assert scheduler._mime_type("x.txt") == "text/plain"
|
|
assert scheduler._mime_type("x.bin") == "application/octet-stream"
|
|
existing = tmp_path / "existing.txt"
|
|
existing.write_text("x", encoding="utf-8")
|
|
assert scheduler._file_size(str(existing)) == 1
|
|
missing = tmp_path / "missing.bin"
|
|
assert scheduler._file_size(str(missing)) == 0
|
|
|
|
|
|
def test_execute_unknown_or_non_queued_run(tmp_path) -> None:
|
|
"""验证未知任务或非排队任务会被忽略。"""
|
|
db = _db(tmp_path)
|
|
db.upsert_workflow({"id": "flow", "name": "Flow", "published": 1, "latest_version": 1})
|
|
now = "2026-01-01T00:00:00+00:00"
|
|
db.create_run(
|
|
{
|
|
"id": "run_done",
|
|
"workflow_id": "flow",
|
|
"workflow_version": 1,
|
|
"status": "COMPLETED",
|
|
"progress": 1,
|
|
"created_at": now,
|
|
"updated_at": now,
|
|
}
|
|
)
|
|
manager = NodeManager(db)
|
|
scheduler = WorkflowScheduler(db, manager, tmp_path / "storage")
|
|
try:
|
|
scheduler.execute_run("missing")
|
|
scheduler.execute_run("run_done")
|
|
finally:
|
|
manager.shutdown()
|
|
assert db.get_run("run_done")["status"] == "COMPLETED"
|
|
|
|
|
|
def test_execute_missing_input(tmp_path) -> None:
|
|
"""验证输入引用无法解析时任务失败。"""
|
|
db = _db(tmp_path)
|
|
definition = WorkflowDefinition(
|
|
name="missing-input",
|
|
version=1,
|
|
nodes=[
|
|
WorkflowNode(
|
|
id="step",
|
|
node_type="echo",
|
|
inputs={"text": "missing.output"},
|
|
)
|
|
],
|
|
entry_inputs={"video_uri": "file"},
|
|
)
|
|
db.upsert_workflow({"id": "flow", "name": "Flow", "published": 1, "latest_version": 1})
|
|
db.create_workflow_version("flow", 1, definition.to_dict())
|
|
now = "2026-01-01T00:00:00+00:00"
|
|
db.create_run(
|
|
{
|
|
"id": "run_input",
|
|
"workflow_id": "flow",
|
|
"workflow_version": 1,
|
|
"status": "QUEUED",
|
|
"progress": 0,
|
|
"input_uri": str(tmp_path / "in.txt"),
|
|
"created_at": now,
|
|
"updated_at": now,
|
|
}
|
|
)
|
|
manager = NodeManager(db)
|
|
scheduler = WorkflowScheduler(db, manager, tmp_path / "storage")
|
|
try:
|
|
scheduler.execute_run("run_input")
|
|
finally:
|
|
manager.shutdown()
|
|
assert db.get_run("run_input")["status"] == "FAILED"
|
|
|
|
|
|
def test_execute_node_failed_response(tmp_path) -> None:
|
|
"""验证节点返回 failed 时任务被标记为失败。"""
|
|
db = _db(tmp_path)
|
|
manifest = NodeManifest(
|
|
id="fail-node",
|
|
name="Fail",
|
|
version="1",
|
|
capability="echo",
|
|
repo_dir="wov-node-echo",
|
|
command=["python", "-u", str(Path(__file__).parent / "fixtures" / "failing_node.py")],
|
|
)
|
|
db.upsert_node(manifest)
|
|
definition = WorkflowDefinition(
|
|
name="fail-flow",
|
|
version=1,
|
|
nodes=[
|
|
WorkflowNode(
|
|
id="step",
|
|
node_type="fail-node",
|
|
inputs={"text": "input.video_uri"},
|
|
)
|
|
],
|
|
entry_inputs={"video_uri": "file"},
|
|
)
|
|
db.upsert_workflow({"id": "flow", "name": "Flow", "published": 1, "latest_version": 1})
|
|
db.create_workflow_version("flow", 1, definition.to_dict())
|
|
now = "2026-01-01T00:00:00+00:00"
|
|
db.create_run(
|
|
{
|
|
"id": "run_fail_node",
|
|
"workflow_id": "flow",
|
|
"workflow_version": 1,
|
|
"status": "QUEUED",
|
|
"progress": 0,
|
|
"input_uri": str(tmp_path / "in.txt"),
|
|
"created_at": now,
|
|
"updated_at": now,
|
|
}
|
|
)
|
|
manager = NodeManager(db)
|
|
scheduler = WorkflowScheduler(db, manager, tmp_path / "storage")
|
|
try:
|
|
scheduler.execute_run("run_fail_node")
|
|
finally:
|
|
manager.shutdown()
|
|
assert db.get_run("run_fail_node")["status"] == "FAILED"
|
|
|
|
|
|
def test_scheduler_start_stop_loop(tmp_path) -> None:
|
|
"""验证调度线程可重复启动并正常停止。"""
|
|
db = _db(tmp_path)
|
|
manager = NodeManager(db)
|
|
scheduler = WorkflowScheduler(db, manager, tmp_path / "storage", interval_seconds=0.05)
|
|
scheduler.start()
|
|
try:
|
|
scheduler.start()
|
|
time.sleep(0.15)
|
|
finally:
|
|
scheduler.stop()
|
|
assert scheduler._thread is None
|
|
|
|
|
|
def test_scheduler_background_executes_queued_run(tmp_path) -> None:
|
|
"""验证后台线程会自动执行排队中的任务。"""
|
|
db = _db(tmp_path)
|
|
input_file = tmp_path / "input.txt"
|
|
input_file.write_text("background", encoding="utf-8")
|
|
manifest = NodeManifest.load(
|
|
Path(__file__).resolve().parent.parent.parent / "wov-node-echo" / "node.manifest.json"
|
|
)
|
|
db.upsert_node(manifest)
|
|
db.upsert_workflow({"id": "flow", "name": "Flow", "published": 1, "latest_version": 1})
|
|
db.create_workflow_version("flow", 1, _echo_definition().to_dict())
|
|
now = "2026-01-01T00:00:00+00:00"
|
|
db.create_run(
|
|
{
|
|
"id": "run_bg",
|
|
"workflow_id": "flow",
|
|
"workflow_version": 1,
|
|
"status": "QUEUED",
|
|
"progress": 0,
|
|
"input_uri": str(input_file),
|
|
"created_at": now,
|
|
"updated_at": now,
|
|
}
|
|
)
|
|
manager = NodeManager(db)
|
|
scheduler = WorkflowScheduler(db, manager, tmp_path / "storage", interval_seconds=0.05)
|
|
scheduler.start()
|
|
try:
|
|
deadline = time.monotonic() + 10
|
|
while time.monotonic() < deadline:
|
|
if db.get_run("run_bg")["status"] in {"COMPLETED", "FAILED"}:
|
|
break
|
|
time.sleep(0.1)
|
|
finally:
|
|
scheduler.stop()
|
|
manager.shutdown()
|
|
assert db.get_run("run_bg")["status"] == "COMPLETED"
|