118 lines
3.9 KiB
Python
118 lines
3.9 KiB
Python
"""工作流管理 API 测试。
|
|
|
|
覆盖工作流的创建、查询、校验、发布、版本列表与删除等管理接口。
|
|
"""
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from 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 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
|