203 lines
7.5 KiB
Python
203 lines
7.5 KiB
Python
"""工作流管理 API 测试。
|
|
|
|
覆盖工作流的创建、查询、校验、发布、版本列表与删除等管理接口。
|
|
"""
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from wov_app.main import app
|
|
|
|
|
|
def definition() -> dict:
|
|
"""构造一个引用 Echo 节点的合法工作流定义。"""
|
|
return {
|
|
"name": "echo-flow",
|
|
"version": 1,
|
|
"nodes": [
|
|
{
|
|
"id": "step",
|
|
"node_type": "echo",
|
|
"inputs": {"file_uri": "input.video_uri"},
|
|
}
|
|
],
|
|
"edges": [],
|
|
"entry_inputs": {"video_uri": "file"},
|
|
"final_outputs": {"result": "step.file_uri"},
|
|
}
|
|
|
|
|
|
def cyclic_definition() -> dict:
|
|
"""构造带环的 DAG 定义（a → b → a），用于验证保存/发布拒绝无效工作流。
|
|
|
|
环上的两个节点互相引用对方输出，形成了无法拓扑排序的依赖：节点 ID 唯一、
|
|
边引用的节点都存在，因此只有环检测能拦住它（R04 复现定义）。
|
|
"""
|
|
return {
|
|
"name": "cyclic",
|
|
"version": 1,
|
|
"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"}],
|
|
"entry_inputs": {"video_uri": "file"},
|
|
"final_outputs": {"result": "a.file_uri"},
|
|
}
|
|
|
|
|
|
def test_create_workflow_rejects_cycle() -> None:
|
|
"""验证保存环形 DAG 返回 422，不把无效版本写入版本表。"""
|
|
with TestClient(app) as client:
|
|
response = client.post(
|
|
"/api/admin/workflows",
|
|
json={
|
|
"id": "cyclic-flow",
|
|
"name": "Cyclic Flow",
|
|
"definition": cyclic_definition(),
|
|
},
|
|
)
|
|
assert response.status_code == 422
|
|
assert "cycle" in response.json()["detail"]
|
|
# 拒绝保存时不得留下半成品工作流/版本记录。
|
|
db = app.state.db
|
|
assert db.get_workflow("cyclic-flow") is None
|
|
assert db.list_workflow_versions("cyclic-flow") == []
|
|
|
|
# 已有工作流重新提交带环定义：同样拒绝，不追加新版本。
|
|
assert client.post(
|
|
"/api/admin/workflows",
|
|
json={"id": "cycle-check", "name": "Cycle Check", "definition": definition()},
|
|
).status_code == 200
|
|
assert client.post(
|
|
"/api/admin/workflows",
|
|
json={"id": "cycle-check", "name": "Cycle Check", "definition": cyclic_definition()},
|
|
).status_code == 422
|
|
assert db.get_workflow("cycle-check")["latest_version"] == 1
|
|
assert len(db.list_workflow_versions("cycle-check")) == 1
|
|
db.delete_workflow("cycle-check")
|
|
|
|
|
|
def test_validate_workflow_rejects_cycle() -> None:
|
|
"""验证只校验不保存的接口同样拒绝环形 DAG。"""
|
|
with TestClient(app) as client:
|
|
db = app.state.db
|
|
# 独立工作流 ID 并显式清理，避免与其他用例（共用同一个测试数据库）互相影响。
|
|
assert client.post(
|
|
"/api/admin/workflows",
|
|
json={"id": "cycle-check", "name": "Cycle Check", "definition": definition()},
|
|
).status_code == 200
|
|
response = client.post(
|
|
"/api/admin/workflows/cycle-check/validate",
|
|
json=cyclic_definition(),
|
|
)
|
|
assert response.status_code == 422
|
|
assert "cycle" in response.json()["detail"]
|
|
db.delete_workflow("cycle-check")
|
|
|
|
|
|
def test_publish_rejects_cycle_in_latest_version() -> None:
|
|
"""验证历史遗留的环形版本不能被发布（发布前重新校验最新版本定义）。"""
|
|
with TestClient(app) as client:
|
|
db = app.state.db
|
|
# 直接写库模拟修复前已保存的无效版本（保存接口现在会拒绝，只能这样构造）。
|
|
db.upsert_workflow(
|
|
{"id": "legacy-cyclic", "name": "Legacy", "published": 0, "latest_version": 1}
|
|
)
|
|
db.create_workflow_version("legacy-cyclic", 1, cyclic_definition())
|
|
response = client.post("/api/admin/workflows/legacy-cyclic/publish")
|
|
assert response.status_code == 422
|
|
assert "cycle" in response.json()["detail"]
|
|
assert db.get_workflow("legacy-cyclic")["published"] == 0
|
|
db.delete_workflow("legacy-cyclic")
|
|
|
|
|
|
def test_workflow_crud_and_publish() -> None:
|
|
"""验证工作流 CRUD、校验、发布与版本列表的完整流程。"""
|
|
with TestClient(app) as client:
|
|
created = client.post(
|
|
"/api/admin/workflows",
|
|
json={
|
|
"id": "echo-flow",
|
|
"name": "Echo Flow",
|
|
"description": "demo",
|
|
"definition": definition(),
|
|
},
|
|
)
|
|
assert created.status_code == 200
|
|
assert created.json()["id"] == "echo-flow"
|
|
|
|
assert client.get("/api/admin/workflows").status_code == 200
|
|
assert client.get("/api/admin/workflows/echo-flow").status_code == 200
|
|
assert client.get("/api/admin/workflows/missing").status_code == 404
|
|
|
|
validated = client.post(
|
|
"/api/admin/workflows/echo-flow/validate",
|
|
json=definition(),
|
|
)
|
|
assert validated.status_code == 200
|
|
assert validated.json()["valid"] is True
|
|
assert client.post(
|
|
"/api/admin/workflows/missing/validate",
|
|
json=definition(),
|
|
).status_code == 404
|
|
|
|
published = client.post("/api/admin/workflows/echo-flow/publish")
|
|
assert published.status_code == 200
|
|
assert published.json()["published"] == "echo-flow"
|
|
assert client.post("/api/admin/workflows/missing/publish").status_code == 404
|
|
|
|
versions = client.get("/api/admin/workflows/echo-flow/versions")
|
|
assert versions.status_code == 200
|
|
assert len(versions.json()) == 1
|
|
assert client.get("/api/admin/workflows/missing/versions").status_code == 404
|
|
|
|
assert client.delete("/api/admin/workflows/echo-flow").status_code == 200
|
|
assert client.delete("/api/admin/workflows/echo-flow").status_code == 404
|
|
|
|
|
|
def test_workflow_slug_without_id() -> None:
|
|
"""验证未提供 ID 时后端会从名称生成 slug。"""
|
|
with TestClient(app) as client:
|
|
created = client.post(
|
|
"/api/admin/workflows",
|
|
json={
|
|
"name": "Echo Flow",
|
|
"definition": definition(),
|
|
},
|
|
)
|
|
assert created.status_code == 200
|
|
assert created.json()["id"] == "echo-flow"
|
|
|
|
|
|
def test_workflow_validation_error() -> None:
|
|
"""验证重复节点 ID 的 DAG 会被拒绝。"""
|
|
with TestClient(app) as client:
|
|
response = client.post(
|
|
"/api/admin/workflows",
|
|
json={
|
|
"id": "bad",
|
|
"name": "Bad",
|
|
"definition": {
|
|
"name": "Bad",
|
|
"version": 1,
|
|
"nodes": [
|
|
{"id": "a", "node_type": "x"},
|
|
{"id": "a", "node_type": "y"},
|
|
],
|
|
"edges": [],
|
|
},
|
|
},
|
|
)
|
|
assert response.status_code == 422
|
|
|
|
|
|
def test_publish_workflow_without_version() -> None:
|
|
"""验证没有版本记录的工作流不能发布。"""
|
|
with TestClient(app) as client:
|
|
db = app.state.db
|
|
db.upsert_workflow(
|
|
{"id": "empty", "name": "Empty", "published": 0, "latest_version": 0}
|
|
)
|
|
response = client.post("/api/admin/workflows/empty/publish")
|
|
assert response.status_code == 422
|