49 lines
1.6 KiB
Python
49 lines
1.6 KiB
Python
"""种子数据测试。
|
|
|
|
验证启动种子逻辑会注册节点仓库、创建演示工作流且保持幂等，并验证
|
|
开启种子与调度器后的应用生命周期。
|
|
"""
|
|
|
|
from pathlib import Path
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.db import Database
|
|
from app.main import app
|
|
from app.seed import seed_demo_workflow, seed_nodes
|
|
|
|
WORKSPACE = Path(__file__).resolve().parent.parent.parent
|
|
|
|
|
|
def test_seed_nodes_and_demo_workflow(tmp_path) -> None:
|
|
"""验证 seed_nodes 注册节点、seed_demo_workflow 创建演示流程且幂等。"""
|
|
db = Database(tmp_path / "wov.db")
|
|
seed_nodes(db, WORKSPACE)
|
|
node_ids = [node.id for node in db.list_nodes()]
|
|
assert "echo" in node_ids
|
|
|
|
seed_demo_workflow(db)
|
|
workflow = db.get_workflow("demo")
|
|
assert workflow is not None
|
|
assert workflow["published"] == 1
|
|
latest = db.get_latest_workflow_version("demo")
|
|
assert latest["definition"]["name"] == "视频字幕生成"
|
|
|
|
seed_demo_workflow(db)
|
|
assert len(db.list_workflow_versions("demo")) == 1
|
|
|
|
empty_workspace = tmp_path / "empty"
|
|
(empty_workspace / "wov-node-empty").mkdir(parents=True)
|
|
# 目录存在但没有 manifest 时不应报错。
|
|
seed_nodes(db, empty_workspace)
|
|
|
|
|
|
def test_lifespan_with_seed_and_scheduler(monkeypatch) -> None:
|
|
"""验证启用自动种子与调度器后应用正常启动且 demo 应用可见。"""
|
|
monkeypatch.setenv("WOV_AUTO_SEED", "1")
|
|
monkeypatch.setenv("WOV_SCHEDULER_ENABLED", "1")
|
|
with TestClient(app) as client:
|
|
apps = client.get("/api/apps")
|
|
assert apps.status_code == 200
|
|
assert any(item["id"] == "demo" for item in apps.json())
|