diff --git a/AGENTS.md b/AGENTS.md index 08182b5..8d63b6c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -36,3 +36,10 @@ uv run uvicorn app.main:app --reload --port 8001 - Python 环境统一使用 uv 管理，不直接使用 pip 修改依赖。 - 测试必须达到 100% 行覆盖率，且只能通过调用真实代码路径覆盖。 - 行覆盖率不包含外部端口占用/权限问题；真实网络启动由 uvicorn 冒烟测试单独覆盖。 + +## 代码注释规范 + +- 本仓库所有源码（Python、TOML 等支持注释的文件）必须配有详细中文注释，说明模块职责、类与函数的作用以及关键逻辑，确保后续维护人员可以快速理解代码工作原理。 +- 新增或修改代码时，必须同步补充或更新对应注释；不得删除已有注释。 +- 测试代码同样必须配有中文注释，说明每条测试验证的行为。 +- JSON 数据文件不支持注释，字段语义以 `wov-sdk` 的 `NodeManifest` 模型注释为准；修改 JSON 字段时须同步更新文档。 diff --git a/app/__init__.py b/app/__init__.py index f113f79..68f310a 100644 --- a/app/__init__.py +++ b/app/__init__.py @@ -1 +1,5 @@ -"""WOV platform API package.""" +"""WOV 平台 API 包。 + +包含 FastAPI 应用入口、SQLite 数据访问、节点生命周期管理、工作流调度器以及 +管理端/用户端路由。 +""" diff --git a/app/config.py b/app/config.py index 362b7ff..2ffd076 100644 --- a/app/config.py +++ b/app/config.py @@ -1,19 +1,31 @@ +"""应用配置中心。 + +集中读取环境变量并推导路径常量，避免业务代码散落魔法值。路径统一使用 +pathlib，Windows 与 Linux 开发环境均可用。 +""" + from __future__ import annotations import os import sys from pathlib import Path +# WOV 工作区根目录：wov-api 位于其下的子目录，父目录即 meta-repo。 WORKSPACE_ROOT = Path(__file__).resolve().parent.parent.parent +# wov-sdk 使用 src 布局，启动时加入 PYTHONPATH 以便直接导入。 SDK_SRC = WORKSPACE_ROOT / "wov-sdk" / "src" if str(SDK_SRC) not in sys.path: sys.path.insert(0, str(SDK_SRC)) +# 数据目录、SQLite 文件与产物存储目录均可通过环境变量覆盖，便于测试隔离。 DATA_DIR = Path(os.getenv("WOV_DATA_DIR", str(WORKSPACE_ROOT / "wov-api" / "data"))) DB_PATH = Path(os.getenv("WOV_DB_PATH", str(DATA_DIR / "wov.db"))) STORAGE_DIR = Path(os.getenv("WOV_STORAGE_DIR", str(DATA_DIR / "storage"))) +# 节点回收线程的轮询间隔。 NODE_REAP_INTERVAL_SECONDS = float(os.getenv("WOV_REAP_INTERVAL_SECONDS", "3")) +# 节点进程启动后等待 WOV_NODE_READY 与健康检查的超时。 NODE_READY_TIMEOUT_SECONDS = float(os.getenv("WOV_READY_TIMEOUT_SECONDS", "12")) +# 单次节点调用（POST /invoke）的 HTTP 超时，LLM 等慢节点需要放宽。 NODE_INVOKE_TIMEOUT_SECONDS = float(os.getenv("WOV_NODE_INVOKE_TIMEOUT_SECONDS", "3600")) diff --git a/app/db.py b/app/db.py index e5b9b4f..36f696d 100644 --- a/app/db.py +++ b/app/db.py @@ -1,3 +1,9 @@ +"""SQLite 数据访问层。 + +所有持久化逻辑集中在本模块，业务代码只依赖 Database 提供的方法。后续切换 +PostgreSQL 时只需替换本层实现，不修改调度器与路由的业务逻辑。 +""" + from __future__ import annotations import json @@ -10,15 +16,21 @@ from wov_sdk.models import NodeManifest class Database: + """SQLite 数据库封装：负责建表以及节点/实例/工作流/任务/产物的 CRUD。""" + def __init__(self, path: Path) -> None: + """打开数据库并确保父目录存在、表结构已初始化。""" self.path = path self.path.parent.mkdir(parents=True, exist_ok=True) self._init_schema() @contextmanager def _connect(self) -> Iterator[sqlite3.Connection]: + """提供带事务提交的数据库连接上下文。""" conn = sqlite3.connect(self.path) + # 按列名读取结果，返回 dict 更直观。 conn.row_factory = sqlite3.Row + # 开启外键约束，保证子表记录引用有效。 conn.execute("PRAGMA foreign_keys = ON") try: yield conn @@ -27,9 +39,11 @@ class Database: conn.close() def _init_schema(self) -> None: + """创建全部业务表；已存在的表保持不变。""" with self._connect() as conn: conn.executescript( """ + -- 节点注册表：保存节点的最新 manifest JSON。 CREATE TABLE IF NOT EXISTS nodes ( id TEXT PRIMARY KEY, name TEXT NOT NULL, @@ -39,6 +53,7 @@ class Database: created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP ); + -- 节点实例表：记录 NodeManager 启动的进程及其状态。 CREATE TABLE IF NOT EXISTS node_instances ( id TEXT PRIMARY KEY, node_id TEXT NOT NULL, @@ -52,6 +67,7 @@ class Database: FOREIGN KEY(node_id) REFERENCES nodes(id) ); + -- 工作流表：只保存概要信息，完整定义存版本表。 CREATE TABLE IF NOT EXISTS workflows ( id TEXT PRIMARY KEY, name TEXT NOT NULL, @@ -61,6 +77,7 @@ class Database: created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP ); + -- 工作流版本表：每个版本保存一份 DAG 定义 JSON。 CREATE TABLE IF NOT EXISTS workflow_versions ( id INTEGER PRIMARY KEY AUTOINCREMENT, workflow_id TEXT NOT NULL, @@ -71,6 +88,7 @@ class Database: FOREIGN KEY(workflow_id) REFERENCES workflows(id) ); + -- 工作流运行表：记录任务从排队到完成/失败的状态机。 CREATE TABLE IF NOT EXISTS workflow_runs ( id TEXT PRIMARY KEY, workflow_id TEXT NOT NULL, @@ -85,6 +103,7 @@ class Database: FOREIGN KEY(workflow_id) REFERENCES workflows(id) ); + -- 产物表：记录每个任务各节点的输出 URI，按名称唯一。 CREATE TABLE IF NOT EXISTS artifacts ( id INTEGER PRIMARY KEY AUTOINCREMENT, run_id TEXT NOT NULL, @@ -101,6 +120,7 @@ class Database: ) def upsert_node(self, manifest: NodeManifest) -> None: + """插入或更新节点注册记录，同名 ID 覆盖为新版本。""" with self._connect() as conn: conn.execute( """ @@ -122,6 +142,7 @@ class Database: ) def get_node(self, node_id: str) -> NodeManifest | None: + """按 ID 读取节点并反序列化为 NodeManifest。""" with self._connect() as conn: row = conn.execute("SELECT manifest_json FROM nodes WHERE id = ?", (node_id,)).fetchone() if row is None: @@ -129,16 +150,20 @@ class Database: return NodeManifest.from_dict(json.loads(row["manifest_json"])) def list_nodes(self) -> list[NodeManifest]: + """按 ID 顺序返回全部注册节点。""" with self._connect() as conn: rows = conn.execute("SELECT manifest_json FROM nodes ORDER BY id").fetchall() return [NodeManifest.from_dict(json.loads(row["manifest_json"])) for row in rows] def delete_node(self, node_id: str) -> None: + """删除节点及其全部实例记录。""" with self._connect() as conn: + # 先删实例再删节点，满足外键约束。 conn.execute("DELETE FROM node_instances WHERE node_id = ?", (node_id,)) conn.execute("DELETE FROM nodes WHERE id = ?", (node_id,)) def upsert_instance(self, instance: dict[str, Any]) -> None: + """插入或更新节点实例状态记录。""" with self._connect() as conn: conn.execute( """ @@ -170,6 +195,7 @@ class Database: ) def list_instances(self) -> list[dict[str, Any]]: + """按启动时间倒序返回全部节点实例。""" with self._connect() as conn: rows = conn.execute( "SELECT * FROM node_instances ORDER BY started_at DESC" @@ -177,10 +203,12 @@ class Database: return [dict(row) for row in rows] def delete_instance(self, instance_id: str) -> None: + """删除指定实例记录。""" with self._connect() as conn: conn.execute("DELETE FROM node_instances WHERE id = ?", (instance_id,)) def upsert_workflow(self, workflow: dict[str, Any]) -> None: + """插入或更新工作流概要信息。""" with self._connect() as conn: conn.execute( """ @@ -202,23 +230,28 @@ class Database: ) def get_workflow(self, workflow_id: str) -> dict[str, Any] | None: + """按 ID 读取工作流概要。""" with self._connect() as conn: row = conn.execute("SELECT * FROM workflows WHERE id = ?", (workflow_id,)).fetchone() return dict(row) if row else None def list_workflows(self) -> list[dict[str, Any]]: + """按创建时间倒序返回全部工作流。""" with self._connect() as conn: rows = conn.execute("SELECT * FROM workflows ORDER BY created_at DESC").fetchall() return [dict(row) for row in rows] def delete_workflow(self, workflow_id: str) -> None: + """级联删除工作流相关的产物、任务、版本和概要记录。""" with self._connect() as conn: + # 外键没有级联删除配置，手动按依赖顺序清理。 conn.execute("DELETE FROM artifacts WHERE run_id IN (SELECT id FROM workflow_runs WHERE workflow_id = ?)", (workflow_id,)) conn.execute("DELETE FROM workflow_runs WHERE workflow_id = ?", (workflow_id,)) conn.execute("DELETE FROM workflow_versions WHERE workflow_id = ?", (workflow_id,)) conn.execute("DELETE FROM workflows WHERE id = ?", (workflow_id,)) def create_workflow_version(self, workflow_id: str, version: int, definition: dict[str, Any]) -> None: + """为工作流新增一个版本，definition 以 JSON 保存。""" with self._connect() as conn: conn.execute( """ @@ -229,6 +262,7 @@ class Database: ) def get_latest_workflow_version(self, workflow_id: str) -> dict[str, Any] | None: + """返回工作流最新版本，并把 definition_json 反序列化为 definition。""" with self._connect() as conn: row = conn.execute( """ @@ -242,10 +276,12 @@ class Database: if row is None: return None result = dict(row) + # 对外统一暴露 definition 字典，隐藏 JSON 存储细节。 result["definition"] = json.loads(result.pop("definition_json")) return result def get_workflow_version(self, workflow_id: str, version: int) -> dict[str, Any] | None: + """按版本号读取指定工作流版本。""" with self._connect() as conn: row = conn.execute( """ @@ -261,6 +297,7 @@ class Database: return result def list_workflow_versions(self, workflow_id: str) -> list[dict[str, Any]]: + """按版本倒序返回工作流全部版本。""" with self._connect() as conn: rows = conn.execute( """ @@ -278,6 +315,7 @@ class Database: return versions def create_run(self, run: dict[str, Any]) -> None: + """创建一条排队中的工作流运行记录。""" with self._connect() as conn: conn.execute( """ @@ -302,11 +340,13 @@ class Database: ) def get_run(self, run_id: str) -> dict[str, Any] | None: + """按 ID 读取任务运行记录。""" with self._connect() as conn: row = conn.execute("SELECT * FROM workflow_runs WHERE id = ?", (run_id,)).fetchone() return dict(row) if row else None def list_runs(self, limit: int = 20) -> list[dict[str, Any]]: + """按创建时间倒序返回最近的运行记录。""" with self._connect() as conn: rows = conn.execute( "SELECT * FROM workflow_runs ORDER BY created_at DESC LIMIT ?", @@ -315,6 +355,8 @@ class Database: return [dict(row) for row in rows] def update_run(self, run_id: str, **fields: Any) -> None: + """更新运行状态字段，同时刷新 updated_at；未知字段会被忽略。""" + # 只允许更新状态机相关字段，防止任意列被改写。 allowed = { "status", "current_node_id", @@ -325,13 +367,16 @@ class Database: if not updates: return updates["updated_at"] = fields.get("updated_at") + # 动态拼接 SET 子句，键来自白名单，不存在 SQL 注入风险。 assignments = ", ".join(f"{key} = ?" for key in updates) values = list(updates.values()) + [run_id] with self._connect() as conn: conn.execute(f"UPDATE workflow_runs SET {assignments} WHERE id = ?", values) def reset_run(self, run_id: str, updated_at: str) -> None: + """把失败任务重置为 QUEUED，并清空进度与旧产物，供重试使用。""" with self._connect() as conn: + # 清空错误和进度，恢复到首次排队时的状态。 conn.execute( """ UPDATE workflow_runs @@ -341,9 +386,11 @@ class Database: """, (updated_at, run_id), ) + # 删除旧产物，避免重试后残留过期下载链接。 conn.execute("DELETE FROM artifacts WHERE run_id = ?", (run_id,)) def next_queued_run(self) -> dict[str, Any] | None: + """按创建时间返回最早一条排队任务，供调度器轮询。""" with self._connect() as conn: row = conn.execute( """ @@ -356,6 +403,7 @@ class Database: return dict(row) if row else None def create_artifact(self, artifact: dict[str, Any]) -> None: + """记录任务产物；同 run 与 name 冲突时覆盖。""" with self._connect() as conn: conn.execute( """ @@ -375,6 +423,7 @@ class Database: ) def list_artifacts(self, run_id: str) -> list[dict[str, Any]]: + """按创建时间返回任务的全部产物。""" with self._connect() as conn: rows = conn.execute( "SELECT * FROM artifacts WHERE run_id = ? ORDER BY created_at", @@ -383,6 +432,7 @@ class Database: return [dict(row) for row in rows] def get_artifact(self, run_id: str, name: str) -> dict[str, Any] | None: + """按任务与产物名读取单个产物记录。""" with self._connect() as conn: row = conn.execute( "SELECT * FROM artifacts WHERE run_id = ? AND name = ?", @@ -391,5 +441,6 @@ class Database: return dict(row) if row else None def delete_run_artifacts(self, run_id: str) -> None: + """删除任务的全部产物记录。""" with self._connect() as conn: conn.execute("DELETE FROM artifacts WHERE run_id = ?", (run_id,)) diff --git a/app/main.py b/app/main.py index 34f61d5..9f588ed 100644 --- a/app/main.py +++ b/app/main.py @@ -1,3 +1,9 @@ +"""FastAPI 应用入口。 + +负责组装数据库、节点管理器、调度器与静态前端，并在应用生命周期内管理 +后台线程的启动与清理。 +""" + from __future__ import annotations import os @@ -18,27 +24,35 @@ from app.seed import seed_demo_workflow, seed_nodes @asynccontextmanager async def lifespan(app: FastAPI): + """应用生命周期：启动时初始化存储、种子数据和后台服务，退出时回收资源。""" + # 确保数据与存储目录存在，避免首次启动写文件失败。 DATA_DIR.mkdir(parents=True, exist_ok=True) STORAGE_DIR.mkdir(parents=True, exist_ok=True) db = Database(DB_PATH) manager = NodeManager(db) + # 启动节点空闲回收线程，负责按 TTL 停止空闲节点进程。 manager.start_reaper() + # 默认自动注册工作区内的节点并创建演示工作流，可关闭便于测试。 if os.getenv("WOV_AUTO_SEED", "1") == "1": seed_nodes(db, WORKSPACE_ROOT) seed_demo_workflow(db) scheduler = WorkflowScheduler(db, manager, STORAGE_DIR) + # 调度器默认开启，处理排队中的任务；测试可关闭后手动执行。 if os.getenv("WOV_SCHEDULER_ENABLED", "1") == "1": scheduler.start() + # 共享对象挂到 app.state，路由通过 Depends 延迟获取。 app.state.db = db app.state.node_manager = manager app.state.scheduler = scheduler yield + # 退出时先停调度器，再关闭全部节点进程，避免残留进程。 scheduler.stop() manager.shutdown() app = FastAPI(title="WOV API", version="0.1.0", lifespan=lifespan) +# MVP 阶段不做鉴权，允许跨域便于本地调试与静态页面访问。 app.add_middleware( CORSMiddleware, allow_origins=["*"], @@ -51,10 +65,13 @@ app.include_router(instances.router) app.include_router(workflows.router) app.include_router(apps.router) + @app.get("/health") def health() -> dict: + """进程存活探针，供部署环境与前端检测后端可用性。""" return {"status": "ok", "service": "wov-api"} +# 静态前端目录位于工作区下的 wov-web，由 FastAPI 直接挂载。 FRONTEND_DIR = Path(__file__).resolve().parent.parent.parent / "wov-web" app.mount("/", StaticFiles(directory=FRONTEND_DIR, html=True), name="frontend") diff --git a/app/node_manager.py b/app/node_manager.py index 0647491..527ceef 100644 --- a/app/node_manager.py +++ b/app/node_manager.py @@ -1,3 +1,10 @@ +"""节点生命周期管理器。 + +NodeManager 是节点实例生命周期的唯一所有者：负责按 Manifest 启动节点进程、 +等待就绪、复用空闲实例、空闲回收和调用转发。API 与调度器不得绕过本模块 +直接启动或杀死节点进程。 +""" + from __future__ import annotations import json @@ -26,25 +33,36 @@ from app.db import Database def _now_iso() -> str: + """返回当前 UTC 时间的 ISO 格式字符串，用于统一时间戳存储。""" return datetime.now(timezone.utc).isoformat() def _idle_seconds(last_used_at: str) -> float: + """计算实例距上次使用的空闲秒数；非法时间按 0 处理。""" if not last_used_at: return 0.0 try: last = datetime.fromisoformat(last_used_at) return max(0.0, (datetime.now(timezone.utc) - last).total_seconds()) except ValueError: + # 时间格式损坏时保守返回 0，避免误回收实例。 return 0.0 def _cuda_library_dirs(repo_dir: Path) -> list[Path]: + """在节点虚拟环境中查找 nvidia cublas/cudnn 动态库目录。 + + faster-whisper 通过 pip 安装的 CUDA 库位于 + .venv/.../site-packages/nvidia/<包>/lib（Linux）或 bin（Windows）， + 需要把这些目录加入动态库搜索路径才能被加载。 + """ venv = repo_dir / ".venv" site_packages: list[Path] = [] + # Windows 的 site-packages 位于 Lib 下。 windows_site = venv / "Lib" / "site-packages" if windows_site.is_dir(): site_packages.append(windows_site) + # Linux 使用 lib/python3.x/site-packages 结构。 site_packages.extend((venv / "lib").glob("python3*/site-packages")) dirs: list[Path] = [] @@ -58,42 +76,61 @@ def _cuda_library_dirs(repo_dir: Path) -> list[Path]: def _with_cuda_library_path(env: dict[str, str], repo_dir: Path) -> dict[str, str]: + """把 CUDA 动态库目录注入进程环境变量并去重。""" dirs = [str(path) for path in _cuda_library_dirs(repo_dir)] if not dirs: return env + # Windows 动态库搜索走 PATH，Linux 走 LD_LIBRARY_PATH。 var = "PATH" if os.name == "nt" else "LD_LIBRARY_PATH" existing = [item for item in env.get(var, "").split(os.pathsep) if item] + # 只追加尚未存在的目录，避免重复路径拖慢加载。 additions = [path for path in dirs if path not in existing] if additions: + # 新目录放在最前，优先使用虚拟环境内的 CUDA 库。 env[var] = os.pathsep.join(additions + existing) return env @dataclass class NodeRuntime: + """运行中的节点实例内存态：进程句柄、状态、地址与使用计数。""" + + # 实例唯一 ID，格式 ni_xxxx。 instance_id: str + # 所属节点注册 ID。 node_id: str + # 启动该实例使用的 manifest 快照。 manifest: NodeManifest + # 生命周期状态：starting / ready / stopping / stopped / error。 status: str = "starting" + # 节点子进程句柄，用于终止与回收。 process: subprocess.Popen | None = None pid: int | None = None + # 节点 HTTP 服务地址，例如 http://127.0.0.1:xxxx。 address: str | None = None started_at: str = field(default_factory=_now_iso) last_used_at: str = field(default_factory=_now_iso) + # 当前并发占用数，超过 manifest.max_concurrency 不再复用。 busy_count: int = 0 error: str | None = None + # 最近 200 行 stderr，便于排查启动失败。 stderr_tail: list[str] = field(default_factory=list) class NodeManager: + """节点进程管理器：启动、复用、调用与回收的唯一入口。""" + def __init__(self, db: Database) -> None: + """保存数据库引用并初始化运行时表与回收线程控制字段。""" self.db = db self._runtimes: dict[str, NodeRuntime] = {} + # 可重入锁保护运行时表，避免回收线程与调用线程竞争。 self._lock = threading.RLock() self._reaper_thread: threading.Thread | None = None self._stopping = False def start_reaper(self) -> None: + """启动后台回收线程，周期性回收空闲节点。""" self._stopping = False self._reaper_thread = threading.Thread( target=self._reaper_loop, @@ -103,18 +140,22 @@ class NodeManager: self._reaper_thread.start() def shutdown(self) -> None: + """停止所有节点实例，供应用退出时清理。""" self._stopping = True with self._lock: for runtime in list(self._runtimes.values()): self._stop_locked(runtime) def _reaper_loop(self) -> None: + """后台循环：每隔固定间隔检查并回收满足 TTL 的空闲实例。""" while not self._stopping: time.sleep(NODE_REAP_INTERVAL_SECONDS) with self._lock: for runtime in list(self._runtimes.values()): + # 未就绪或正在被调用的实例不回收。 if runtime.status != "ready" or runtime.busy_count > 0: continue + # 常驻节点永不自动回收。 if runtime.manifest.keep_warm: continue ttl = runtime.manifest.idle_ttl_seconds @@ -123,9 +164,11 @@ class NodeManager: self._stop_locked(runtime) def _resolve_command(self, manifest: NodeManifest) -> list[str]: + """解析启动命令；python 命令替换为节点仓库虚拟环境解释器。""" command = list(manifest.command) if command and command[0].lower() in {"python", "python3"}: repo_dir = WORKSPACE_ROOT / manifest.repo_dir + # 同时兼容 Windows 与 Linux 的虚拟环境路径。 windows_python = repo_dir / ".venv" / "Scripts" / "python.exe" unix_python = repo_dir / ".venv" / "bin" / "python" if windows_python.is_file(): @@ -133,20 +176,25 @@ class NodeManager: elif unix_python.is_file(): command[0] = str(unix_python) else: + # 虚拟环境不存在时使用 API 自身的 Python，便于演示。 command[0] = sys.executable return command def _node_env(self, manifest: NodeManifest) -> dict[str, str]: + """构造节点进程环境：注入 SDK 路径、manifest 环境变量与 CUDA 路径。""" env = os.environ.copy() current_pythonpath = env.get("PYTHONPATH", "") + # 把 wov-sdk 源码目录放到 PYTHONPATH 最前。 env["PYTHONPATH"] = os.pathsep.join( item for item in [str(SDK_SRC), current_pythonpath] if item ) env.update(manifest.env) + # 为 faster-whisper 等节点补充 CUDA 动态库搜索路径。 repo_dir = WORKSPACE_ROOT / manifest.repo_dir return _with_cuda_library_path(env, repo_dir) def _start_locked(self, manifest: NodeManifest) -> NodeRuntime: + """启动单个节点进程并等待就绪；调用方必须持有锁。""" runtime = NodeRuntime( instance_id=f"ni_{uuid.uuid4().hex[:12]}", node_id=manifest.id, @@ -154,11 +202,13 @@ class NodeManager: ) repo_dir = WORKSPACE_ROOT / manifest.repo_dir if not repo_dir.is_dir(): + # 仓库目录缺失时直接标记错误，避免后续命令误导。 runtime.status = "error" runtime.error = f"repo_dir not found: {repo_dir}" self.db.upsert_instance(self._instance_row(runtime)) return runtime + # 子进程 stdout 读取线程解析到端口后设置事件。 ready_event = threading.Event() ready_port: dict[str, int] = {} @@ -174,6 +224,7 @@ class NodeManager: bufsize=1, ) except Exception as exc: # noqa: BLE001 + # 命令不存在等启动错误统一记录为 error。 runtime.status = "error" runtime.error = str(exc) self.db.upsert_instance(self._instance_row(runtime)) @@ -183,17 +234,21 @@ class NodeManager: runtime.pid = process.pid def read_stdout() -> None: + """逐行读取子进程 stdout，解析 WOV_NODE_READY 端口。""" assert process.stdout is not None for line in iter(process.stdout.readline, ""): line = line.strip() if line.startswith("WOV_NODE_READY"): try: + # 从 "port=<数字>" 中提取端口。 ready_port["port"] = int(line.split("port=", 1)[1]) except (IndexError, ValueError): + # 格式异常时按 0 处理，随后会判定为未就绪。 ready_port["port"] = 0 ready_event.set() def read_stderr() -> None: + """缓存子进程 stderr 尾部，供失败诊断使用。""" assert process.stderr is not None for line in iter(process.stderr.readline, ""): line = line.strip() @@ -204,6 +259,7 @@ class NodeManager: threading.Thread(target=read_stdout, daemon=True).start() threading.Thread(target=read_stderr, daemon=True).start() + # 等待就绪行，超时或端口无效则终止进程并标记失败。 ready = ready_event.wait(timeout=NODE_READY_TIMEOUT_SECONDS) if not ready or "port" not in ready_port or not ready_port["port"]: self._stop_process(process) @@ -212,12 +268,14 @@ class NodeManager: self.db.upsert_instance(self._instance_row(runtime)) return runtime + # 端口有效后做一次真实 HTTP 健康检查。 address = f"http://127.0.0.1:{ready_port['port']}" try: with urllib.request.urlopen(f"{address}/health", timeout=2) as response: if response.status != 200: raise RuntimeError(f"health check returned {response.status}") except Exception as exc: # noqa: BLE001 + # 健康检查失败说明进程虽然打印了就绪行但服务不可用。 self._stop_process(process) runtime.status = "error" runtime.error = f"health check failed: {exc}" @@ -233,25 +291,31 @@ class NodeManager: return runtime def _stop_process(self, process: subprocess.Popen) -> None: + """先优雅 terminate，超时后强制 kill。""" + # 进程已退出时无需处理。 if process.poll() is not None: return process.terminate() try: process.wait(timeout=5) except subprocess.TimeoutExpired: + # 优雅退出失败时强制结束，避免僵尸进程占用端口。 process.kill() process.wait(timeout=5) def _stop_locked(self, runtime: NodeRuntime) -> None: + """停止实例进程并更新数据库状态；调用方必须持有锁。""" if runtime.process is not None: self._stop_process(runtime.process) runtime.status = "stopped" runtime.pid = None runtime.address = None + # 从内存运行时表中移除，后续调用会重新启动。 self._runtimes.pop(runtime.instance_id, None) self.db.upsert_instance(self._instance_row(runtime)) def _instance_row(self, runtime: NodeRuntime) -> dict: + """把运行时对象转换为数据库实例记录。""" return { "id": runtime.instance_id, "node_id": runtime.node_id, @@ -265,11 +329,13 @@ class NodeManager: } def acquire(self, node_id: str) -> tuple[NodeRuntime, str]: + """获取一个可用节点实例：优先复用空闲实例，否则启动新实例。""" manifest = self.db.get_node(node_id) if manifest is None: raise ValueError(f"node not registered: {node_id}") with self._lock: + # 优先复用未达并发上限的就绪实例。 for runtime in self._runtimes.values(): if ( runtime.node_id == node_id @@ -280,24 +346,30 @@ class NodeManager: runtime.last_used_at = _now_iso() return runtime, runtime.address or "" + # 没有可复用实例时启动新进程。 runtime = self._start_locked(manifest) if runtime.status != "ready": raise RuntimeError(runtime.error or "node failed to start") + # 新实例占用一个并发槽位。 runtime.busy_count += 1 runtime.last_used_at = _now_iso() return runtime, runtime.address or "" def release(self, instance_id: str) -> None: + """释放一次实例占用；由 invoke 的 finally 保证执行。""" with self._lock: runtime = self._runtimes.get(instance_id) if runtime is None: return + # 计数下限为 0，防止重复释放导致负值。 runtime.busy_count = max(0, runtime.busy_count - 1) runtime.last_used_at = _now_iso() self.db.upsert_instance(self._instance_row(runtime)) def invoke(self, node_id: str, request: InvokeRequest) -> InvokeResponse: + """向节点实例发起一次调用，并保证无论成败都释放实例。""" runtime, address = self.acquire(node_id) + # 回填实际实例 ID，供节点与日志追踪。 request.node_instance_id = runtime.instance_id body = json.dumps(request.to_dict(), ensure_ascii=False).encode("utf-8") http_request = urllib.request.Request( @@ -307,21 +379,25 @@ class NodeManager: method="POST", ) try: + # 超时来自配置，避免 LLM 等慢节点被过早中断。 with urllib.request.urlopen(http_request, timeout=NODE_INVOKE_TIMEOUT_SECONDS) as response: payload = json.loads(response.read().decode("utf-8")) return InvokeResponse.from_dict(payload) except urllib.error.HTTPError as exc: try: + # 节点返回 500 时 body 仍是协议 JSON，尝试解析失败原因。 payload = json.loads(exc.read().decode("utf-8")) return InvokeResponse.from_dict(payload) except Exception: # noqa: BLE001 return InvokeResponse(status="failed", error=f"node returned {exc.code}") except Exception as exc: # noqa: BLE001 + # 网络错误、超时等统一转换为 failed。 return InvokeResponse(status="failed", error=str(exc)) finally: self.release(request.node_instance_id) def stop_instance(self, instance_id: str) -> None: + """按实例 ID 停止指定节点，供管理后台使用。""" with self._lock: runtime = self._runtimes.get(instance_id) if runtime is None: @@ -329,6 +405,7 @@ class NodeManager: self._stop_locked(runtime) def stop_all_for_node(self, node_id: str) -> None: + """停止某个节点的全部实例，用于删除节点前清理。""" with self._lock: for runtime in list(self._runtimes.values()): if runtime.node_id == node_id: diff --git a/app/routers/__init__.py b/app/routers/__init__.py index 72d3248..cab2cea 100644 --- a/app/routers/__init__.py +++ b/app/routers/__init__.py @@ -1 +1,4 @@ -"""WOV API routers.""" +"""WOV API 路由包。 + +按职责拆分为节点、实例、工作流和用户应用四组路由，统一由 app.main 挂载。 +""" diff --git a/app/routers/apps.py b/app/routers/apps.py index 55a3083..863541d 100644 --- a/app/routers/apps.py +++ b/app/routers/apps.py @@ -1,3 +1,9 @@ +"""用户端应用路由。 + +面向普通用户暴露“应用中心”能力：列出已发布工作流、上传输入创建任务、 +查询进度、重试失败任务以及下载产物。用户只看到输入 -> 进度 -> 结果。 +""" + from __future__ import annotations import uuid @@ -13,10 +19,12 @@ router = APIRouter(tags=["apps"]) def _now_iso() -> str: + """返回当前 UTC 时间的 ISO 格式字符串。""" return datetime.now(timezone.utc).isoformat() def _get_db() -> Database: + """从 FastAPI 应用状态中延迟获取数据库实例。""" from app.main import app return app.state.db @@ -24,8 +32,10 @@ def _get_db() -> Database: @router.get("/api/apps") def list_apps(db: Database = Depends(_get_db)) -> list[dict]: + """返回全部已发布工作流及其最新版本定义。""" apps = [] for workflow in db.list_workflows(): + # 草稿工作流不对用户端可见。 if not workflow["published"]: continue latest = db.get_latest_workflow_version(workflow["id"]) @@ -47,7 +57,9 @@ async def create_run( file: UploadFile = File(...), db: Database = Depends(_get_db), ) -> dict: + """接收用户上传文件，创建排队中的工作流任务。""" workflow = db.get_workflow(workflow_id) + # 只允许对已发布且存在版本的工作流发起任务。 if workflow is None or not workflow["published"]: raise HTTPException(status_code=404, detail="published workflow not found") @@ -56,9 +68,11 @@ async def create_run( raise HTTPException(status_code=422, detail="workflow has no version") run_id = f"run_{uuid.uuid4().hex[:12]}" + # 使用安全文件名，避免路径穿越。 filename = Path(file.filename or "upload.bin").name from app.config import STORAGE_DIR + # 上传文件按 run 隔离存放，调度器通过 input_uri 引用。 input_dir = STORAGE_DIR / "uploads" / run_id input_dir.mkdir(parents=True, exist_ok=True) input_uri = input_dir / filename @@ -88,11 +102,13 @@ async def create_run( @router.get("/api/runs") def list_runs(db: Database = Depends(_get_db)) -> list[dict]: + """返回最近的运行记录，供任务管理页展示。""" return db.list_runs() @router.get("/api/runs/{run_id}") def get_run(run_id: str, db: Database = Depends(_get_db)) -> dict: + """返回任务详情，并附带当前产物列表。""" run = db.get_run(run_id) if run is None: raise HTTPException(status_code=404, detail="run not found") @@ -102,17 +118,20 @@ def get_run(run_id: str, db: Database = Depends(_get_db)) -> dict: @router.post("/api/runs/{run_id}/retry") def retry_run(run_id: str, db: Database = Depends(_get_db)) -> dict: + """重置失败任务为排队状态，清空旧产物后重新执行。""" run = db.get_run(run_id) if run is None: raise HTTPException(status_code=404, detail="run not found") if run["status"] != "FAILED": raise HTTPException(status_code=422, detail="only failed runs can be retried") + # reset_run 会清空进度、错误和旧产物，确保从头开始。 db.reset_run(run_id, _now_iso()) return {"id": run_id, "status": "QUEUED"} @router.get("/api/runs/{run_id}/artifacts") def list_run_artifacts(run_id: str, db: Database = Depends(_get_db)) -> list[dict]: + """返回任务全部产物记录。""" if db.get_run(run_id) is None: raise HTTPException(status_code=404, detail="run not found") return db.list_artifacts(run_id) @@ -124,6 +143,7 @@ def download_artifact( artifact_name: str, db: Database = Depends(_get_db), ) -> FileResponse: + """按任务与产物名下载文件，文件缺失时返回 404。""" artifact = db.get_artifact(run_id, artifact_name) if artifact is None: raise HTTPException(status_code=404, detail="artifact not found") diff --git a/app/routers/instances.py b/app/routers/instances.py index 36b20cd..24f0610 100644 --- a/app/routers/instances.py +++ b/app/routers/instances.py @@ -1,3 +1,9 @@ +"""节点实例管理路由。 + +提供管理后台查看节点实例与手动停止实例的能力。实例生命周期仍由 +NodeManager 控制，路由只转发停止请求。 +""" + from __future__ import annotations from fastapi import APIRouter, Depends, HTTPException @@ -9,12 +15,14 @@ router = APIRouter(prefix="/api/admin/node-instances", tags=["node-instances"]) def _get_db() -> Database: + """从应用状态延迟获取数据库实例。""" from app.main import app return app.state.db def _get_manager() -> NodeManager: + """从应用状态延迟获取节点管理器。""" from app.main import app return app.state.node_manager @@ -22,6 +30,7 @@ def _get_manager() -> NodeManager: @router.get("") def list_instances(db: Database = Depends(_get_db)) -> list[dict]: + """返回全部节点实例记录。""" return db.list_instances() @@ -31,6 +40,7 @@ def stop_instance( db: Database = Depends(_get_db), manager: NodeManager = Depends(_get_manager), ) -> dict: + """请求停止指定实例；停止后实例记录保留为 stopped 状态。""" manager.stop_instance(instance_id) instance = next( (item for item in db.list_instances() if item["id"] == instance_id), diff --git a/app/routers/nodes.py b/app/routers/nodes.py index 5ed7f5c..6ba63a5 100644 --- a/app/routers/nodes.py +++ b/app/routers/nodes.py @@ -1,3 +1,9 @@ +"""节点管理路由。 + +提供节点注册、查询、删除和手动调用接口。注册数据进入 SQLite 节点注册表， +实际进程启动与回收仍由 NodeManager 负责。 +""" + from __future__ import annotations from fastapi import APIRouter, Depends, HTTPException @@ -11,12 +17,14 @@ router = APIRouter(prefix="/api/admin/nodes", tags=["nodes"]) def _get_db() -> Database: + """从应用状态延迟获取数据库实例。""" from app.main import app return app.state.db def _get_manager() -> NodeManager: + """从应用状态延迟获取节点管理器。""" from app.main import app return app.state.node_manager @@ -24,8 +32,10 @@ def _get_manager() -> NodeManager: @router.post("") def register_node(payload: NodeCreate, db: Database = Depends(_get_db)) -> dict: + """校验并注册节点，返回注册后的 manifest。""" manifest = payload.to_manifest() try: + # 协议级校验保证注册表内数据始终合法。 manifest.validate() except ValueError as exc: raise HTTPException(status_code=422, detail=str(exc)) from exc @@ -35,11 +45,13 @@ def register_node(payload: NodeCreate, db: Database = Depends(_get_db)) -> dict: @router.get("") def list_nodes(db: Database = Depends(_get_db)) -> list[dict]: + """返回全部已注册节点。""" return [manifest.to_dict() for manifest in db.list_nodes()] @router.get("/{node_id}") def get_node(node_id: str, db: Database = Depends(_get_db)) -> dict: + """按 ID 返回节点 manifest。""" manifest = db.get_node(node_id) if manifest is None: raise HTTPException(status_code=404, detail="node not found") @@ -52,8 +64,10 @@ def delete_node( db: Database = Depends(_get_db), manager: NodeManager = Depends(_get_manager), ) -> dict: + """删除节点前先停止其全部运行实例。""" if db.get_node(node_id) is None: raise HTTPException(status_code=404, detail="node not found") + # 先回收进程再删注册记录，避免残留孤儿进程。 manager.stop_all_for_node(node_id) db.delete_node(node_id) return {"deleted": node_id} @@ -65,8 +79,10 @@ def invoke_node( payload: InvokePayload, manager: NodeManager = Depends(_get_manager), ) -> dict: + """管理后台手动调用节点，产物写入固定输出目录。""" from app.config import STORAGE_DIR + # 与管理运行共用目录结构，便于调试产物位置。 output_dir = ( STORAGE_DIR / "runs" / payload.run_id / "steps" / node_id ) @@ -86,6 +102,7 @@ def list_node_instances( node_id: str, db: Database = Depends(_get_db), ) -> list[dict]: + """返回指定节点的全部实例记录。""" if db.get_node(node_id) is None: raise HTTPException(status_code=404, detail="node not found") return [ diff --git a/app/routers/workflows.py b/app/routers/workflows.py index 4b30c38..c9fdbe1 100644 --- a/app/routers/workflows.py +++ b/app/routers/workflows.py @@ -1,3 +1,9 @@ +"""工作流管理路由。 + +提供工作流的创建、查询、校验、发布和删除能力。工作流以版本化 DAG 数据保存， +不写死在业务代码中。 +""" + from __future__ import annotations import re @@ -13,17 +19,20 @@ router = APIRouter(prefix="/api/admin/workflows", tags=["workflows"]) def _get_db() -> Database: + """从应用状态延迟获取数据库实例。""" from app.main import app return app.state.db def _slugify(value: str) -> str: + """把工作流名称转换为小写连字符 ID；无有效字符时生成随机 ID。""" slug = re.sub(r"[^a-z0-9]+", "-", value.lower()).strip("-") return slug or uuid.uuid4().hex[:8] def _validate_definition(raw: dict) -> WorkflowDefinition: + """解析并校验 DAG 定义，非法时转换为 422 HTTP 异常。""" try: definition = WorkflowDefinition.from_dict(raw) definition.validate() @@ -34,6 +43,7 @@ def _validate_definition(raw: dict) -> WorkflowDefinition: @router.get("") def list_workflows(db: Database = Depends(_get_db)) -> list[dict]: + """返回全部工作流概要。""" return db.list_workflows() @@ -42,10 +52,13 @@ def create_workflow( payload: WorkflowCreate, db: Database = Depends(_get_db), ) -> dict: + """创建新工作流或为已有工作流追加一个版本。""" definition = _validate_definition(payload.definition) + # 未显式指定 ID 时由名称生成；已有工作流则版本号递增。 workflow_id = payload.id or _slugify(payload.name) existing = db.get_workflow(workflow_id) version = (existing or {}).get("latest_version", 0) + 1 + # 每次创建都保存新版本，发布操作只切换 published 标记。 db.upsert_workflow( { "id": workflow_id, @@ -67,6 +80,7 @@ def create_workflow( @router.get("/{workflow_id}") def get_workflow(workflow_id: str, db: Database = Depends(_get_db)) -> dict: + """返回工作流概要及最新版本定义。""" workflow = db.get_workflow(workflow_id) if workflow is None: raise HTTPException(status_code=404, detail="workflow not found") @@ -77,6 +91,7 @@ def get_workflow(workflow_id: str, db: Database = Depends(_get_db)) -> dict: @router.delete("/{workflow_id}") def delete_workflow(workflow_id: str, db: Database = Depends(_get_db)) -> dict: + """删除工作流及其版本、任务和产物记录。""" if db.get_workflow(workflow_id) is None: raise HTTPException(status_code=404, detail="workflow not found") db.delete_workflow(workflow_id) @@ -89,6 +104,7 @@ def validate_workflow( definition: dict, db: Database = Depends(_get_db), ) -> dict: + """在不保存的情况下校验一份 DAG 定义。""" if db.get_workflow(workflow_id) is None: raise HTTPException(status_code=404, detail="workflow not found") parsed = _validate_definition(definition) @@ -97,11 +113,13 @@ def validate_workflow( @router.post("/{workflow_id}/publish") def publish_workflow(workflow_id: str, db: Database = Depends(_get_db)) -> dict: + """把工作流标记为已发布，使其出现在用户应用中心。""" workflow = db.get_workflow(workflow_id) if workflow is None: raise HTTPException(status_code=404, detail="workflow not found") if workflow["latest_version"] == 0: raise HTTPException(status_code=422, detail="workflow has no version") + # 发布只是状态切换，不修改已保存的版本数据。 db.upsert_workflow( { "id": workflow_id, @@ -116,6 +134,7 @@ def publish_workflow(workflow_id: str, db: Database = Depends(_get_db)) -> dict: @router.get("/{workflow_id}/versions") def list_versions(workflow_id: str, db: Database = Depends(_get_db)) -> list[dict]: + """返回工作流全部版本定义。""" if db.get_workflow(workflow_id) is None: raise HTTPException(status_code=404, detail="workflow not found") return db.list_workflow_versions(workflow_id) diff --git a/app/scheduler.py b/app/scheduler.py index 1b15033..53205c3 100644 --- a/app/scheduler.py +++ b/app/scheduler.py @@ -1,3 +1,9 @@ +"""工作流调度器。 + +轮询 SQLite 中的排队任务，按工作流 DAG 的拓扑顺序依次调用节点，并把节点 +产物登记为任务产物。MVP 使用进程内单线程顺序执行，后续可替换为分布式队列。 +""" + from __future__ import annotations import threading @@ -13,20 +19,25 @@ from app.node_manager import NodeManager def _now_iso() -> str: + """返回当前 UTC 时间的 ISO 格式字符串。""" return datetime.now(timezone.utc).isoformat() def topological_sort(definition: WorkflowDefinition) -> list[str]: + """对工作流 DAG 做拓扑排序，返回可执行的节点 ID 顺序。""" nodes = {node.id: node for node in definition.nodes} + # 统计每个节点的入度，并记录依赖关系。 indegree = {node_id: 0 for node_id in nodes} dependents: dict[str, list[str]] = {node_id: [] for node_id in nodes} for edge in definition.edges: + # 边引用了不存在的节点时直接报错。 if edge.from_node not in nodes or edge.to_node not in nodes: raise ValueError(f"unknown edge: {edge.from_node} -> {edge.to_node}") indegree[edge.to_node] += 1 dependents[edge.from_node].append(edge.to_node) + # Kahn 算法：从入度为 0 的节点开始逐层取出。 queue = [node_id for node_id, degree in indegree.items() if degree == 0] ordered: list[str] = [] while queue: @@ -37,12 +48,15 @@ def topological_sort(definition: WorkflowDefinition) -> list[str]: if indegree[dependent] == 0: queue.append(dependent) + # 排序结果数量不足说明存在环，无法确定执行顺序。 if len(ordered) != len(nodes): raise ValueError("workflow contains a cycle") return ordered class WorkflowScheduler: + """后台任务调度器：单线程轮询并执行排队中的工作流运行。""" + def __init__( self, db: Database, @@ -50,6 +64,7 @@ class WorkflowScheduler: storage_dir: Path, interval_seconds: float = 1.0, ) -> None: + """保存依赖并初始化轮询线程控制字段。""" self.db = db self.node_manager = node_manager self.storage_dir = storage_dir @@ -58,6 +73,7 @@ class WorkflowScheduler: self._stopping = False def start(self) -> None: + """启动调度线程；重复调用无副作用。""" if self._thread is not None: return self._stopping = False @@ -69,12 +85,14 @@ class WorkflowScheduler: self._thread.start() def stop(self) -> None: + """请求停止并等待轮询线程退出。""" self._stopping = True if self._thread is not None: self._thread.join(timeout=5) self._thread = None def _loop(self) -> None: + """轮询循环：有排队任务就立即执行，否则休眠一个间隔。""" while not self._stopping: run = self.db.next_queued_run() if run is not None: @@ -88,18 +106,24 @@ class WorkflowScheduler: run_input_uri: str | None, outputs_by_node: dict[str, dict[str, str]], ) -> str | None: + """解析输入引用：input.xxx 取任务入口，node.key 取前序节点产物。""" + # 入口引用以 input. 为前缀。 if ref.startswith("input."): return run_input_uri + # 其余引用必须形如 "节点ID.输出名"。 node_id, separator, key = ref.partition(".") if not separator: return None return outputs_by_node.get(node_id, {}).get(key) def execute_run(self, run_id: str) -> None: + """执行单个任务：加载 DAG、按拓扑顺序调用节点并登记产物。""" run = self.db.get_run(run_id) + # 任务不存在或不在排队状态时直接返回，避免重复执行。 if run is None or run["status"] != "QUEUED": return + # 工作流或版本记录丢失时把任务标记为失败。 workflow = self.db.get_workflow(run["workflow_id"]) if workflow is None: self.db.update_run(run_id, status="FAILED", error="workflow not found", updated_at=_now_iso()) @@ -110,6 +134,7 @@ class WorkflowScheduler: self.db.update_run(run_id, status="FAILED", error="workflow version not found", updated_at=_now_iso()) return + # 解析并校验 DAG，随后计算拓扑执行顺序。 definition = WorkflowDefinition.from_dict(version["definition"]) definition.validate() ordered = topological_sort(definition) @@ -118,6 +143,7 @@ class WorkflowScheduler: try: for index, node_id in enumerate(ordered): + # 当前节点进度 = 已完成节点数 / 总节点数。 node_spec = next(item for item in definition.nodes if item.id == node_id) self.db.update_run( run_id, @@ -125,6 +151,7 @@ class WorkflowScheduler: progress=index / len(ordered), updated_at=_now_iso(), ) + # 解析节点声明的每个输入引用，缺任一输入即失败。 invoke_inputs: dict[str, str] = {} for input_name, ref in node_spec.inputs.items(): value = self._resolve_ref(ref, run.get("input_uri"), outputs_by_node) @@ -132,6 +159,7 @@ class WorkflowScheduler: raise ValueError(f"missing input {input_name} for node {node_id}") invoke_inputs[input_name] = value + # 每个任务的每个节点使用独立产物目录，避免并发冲突。 output_dir = ( self.storage_dir / "runs" @@ -149,13 +177,16 @@ class WorkflowScheduler: output_dir=str(output_dir), ), ) + # 节点返回非 completed 即视为步骤失败。 if response.status != "completed": raise RuntimeError(response.error or f"node {node_id} failed") + # 记录节点输出，供后续节点引用和最终产物映射使用。 outputs_by_node[node_id] = { str(key): str(value) for key, value in response.outputs.items() } for key, uri in outputs_by_node[node_id].items(): + # 产物名带节点前缀，例如 asr.srt_uri，避免跨节点重名。 artifact = { "run_id": run_id, "node_id": node_id, @@ -166,6 +197,7 @@ class WorkflowScheduler: } self.db.create_artifact(artifact) + # 处理 final_outputs，为用户端提供简洁的下载别名。 for alias, ref in definition.final_outputs.items(): resolved = self._resolve_ref(ref, run.get("input_uri"), outputs_by_node) if resolved is not None: @@ -180,6 +212,7 @@ class WorkflowScheduler: } ) + # 全部节点成功后任务标记为完成。 self.db.update_run( run_id, status="COMPLETED", @@ -188,6 +221,7 @@ class WorkflowScheduler: updated_at=_now_iso(), ) except Exception as exc: # noqa: BLE001 + # 任一步骤异常都结束任务并记录错误，等待用户重试。 self.db.update_run( run_id, status="FAILED", @@ -197,6 +231,7 @@ class WorkflowScheduler: @staticmethod def _mime_type(uri: str) -> str: + """按扩展名推断产物 MIME 类型，未知类型使用通用二进制类型。""" path = Path(uri) suffix = path.suffix.lower() return { @@ -209,6 +244,7 @@ class WorkflowScheduler: @staticmethod def _file_size(uri: str) -> int: + """读取产物文件大小；文件缺失时按 0 处理。""" try: return Path(uri).stat().st_size except OSError: diff --git a/app/schemas.py b/app/schemas.py index 606da03..67f0c13 100644 --- a/app/schemas.py +++ b/app/schemas.py @@ -1,3 +1,9 @@ +"""FastAPI 请求/响应 schema。 + +使用 Pydantic 模型校验管理 API 的 JSON 请求体，并把请求数据转换为 SDK +协议模型。 +""" + from __future__ import annotations from typing import Any @@ -8,6 +14,8 @@ from wov_sdk.models import NodeManifest class NodeCreate(BaseModel): + """节点注册请求体，字段与 NodeManifest 保持一致。""" + id: str = Field(min_length=1) name: str = Field(min_length=1) version: str = Field(min_length=1) @@ -23,6 +31,7 @@ class NodeCreate(BaseModel): keep_warm: bool = False def to_manifest(self) -> NodeManifest: + """转换为 SDK 的 NodeManifest 对象，供注册与校验使用。""" return NodeManifest( id=self.id, name=self.name, @@ -41,13 +50,20 @@ class NodeCreate(BaseModel): class InvokePayload(BaseModel): + """管理后台手动调用节点的请求体。""" + + # 默认 run_id 便于快速联调，正式运行时会由调度器生成。 run_id: str = Field(default_factory=lambda: "run_admin") inputs: dict[str, Any] = Field(default_factory=dict) params: dict[str, Any] = Field(default_factory=dict) class WorkflowCreate(BaseModel): + """创建工作流或新增版本的请求体。""" + + # 缺省时由后端根据名称生成 slug ID。 id: str | None = None name: str = Field(min_length=1) description: str = "" + # DAG 原始字典，后端会解析并校验为 WorkflowDefinition。 definition: dict[str, Any] diff --git a/app/seed.py b/app/seed.py index 885ab62..8df7100 100644 --- a/app/seed.py +++ b/app/seed.py @@ -1,3 +1,9 @@ +"""种子数据模块。 + +启动时把工作区内所有 wov-node-* 子仓库的 manifest 注册为节点，并创建演示 +“视频字幕生成”工作流，方便本地直接体验完整链路。 +""" + from __future__ import annotations from pathlib import Path @@ -8,16 +14,21 @@ from app.db import Database def seed_nodes(db: Database, workspace_root: Path) -> None: + """扫描工作区中的节点仓库并注册其 manifest。""" for node_dir in sorted(workspace_root.glob("wov-node-*")): manifest_path = node_dir / "node.manifest.json" + # 没有 manifest 的目录不是节点仓库，直接跳过。 if not manifest_path.is_file(): continue manifest = NodeManifest.load(str(manifest_path)) + # 以目录名作为 repo_dir，确保 NodeManager 能定位到子仓库。 manifest.repo_dir = node_dir.name db.upsert_node(manifest) def seed_demo_workflow(db: Database) -> None: + """创建演示工作流：提音 -> 转写 -> 翻译 -> ASS。""" + # 已存在同名工作流时不重复创建，保持幂等。 if db.get_workflow("demo") is not None: return definition = WorkflowDefinition( @@ -69,4 +80,5 @@ def seed_demo_workflow(db: Database) -> None: "latest_version": 1, } ) + # 保存第一个版本的 DAG 定义，后续发布流程以版本记录为准。 db.create_workflow_version("demo", 1, definition.to_dict()) diff --git a/pyproject.toml b/pyproject.toml index 324a1d9..626a391 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,8 +1,10 @@ +# WOV API 项目配置：使用 uv 管理环境与依赖。 [project] name = "wov-api" version = "0.1.0" description = "WOV platform API" requires-python = ">=3.11" +# fastapi/uvicorn 提供 Web 服务，python-multipart 支持文件上传。 dependencies = [ "fastapi", "uvicorn", @@ -10,12 +12,15 @@ dependencies = [ "wov-sdk", ] +# 本地路径依赖 wov-sdk。 [tool.uv.sources] wov-sdk = { path = "../wov-sdk" } +# 开发依赖：pytest、覆盖率工具与 httpx（TestClient 依赖）。 [dependency-groups] dev = ["pytest", "pytest-cov", "httpx"] +# pytest 配置：强制 app 包 100% 行覆盖率。 [tool.pytest.ini_options] testpaths = ["tests"] pythonpath = ["."] diff --git a/tests/conftest.py b/tests/conftest.py index 08a569a..f764e05 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,20 +1,31 @@ +"""pytest 全局配置。 + +在测试进程启动时创建独立临时目录，并通过环境变量把应用的数据目录、数据库、 +存储和后台服务全部指向测试环境，避免污染本地开发数据。 +""" + import atexit import os import shutil import tempfile from pathlib import Path +# 每个测试进程使用独立临时根目录，保证测试之间互不干扰。 TEST_ROOT = Path(tempfile.mkdtemp(prefix="wov-api-test-")) os.environ["WOV_DATA_DIR"] = str(TEST_ROOT / "data") os.environ["WOV_DB_PATH"] = str(TEST_ROOT / "data" / "wov.db") os.environ["WOV_STORAGE_DIR"] = str(TEST_ROOT / "storage") +# 回收线程轮询调快，便于空闲回收测试尽快完成。 os.environ["WOV_REAP_INTERVAL_SECONDS"] = "0.1" +# 就绪超时缩短，避免节点启动失败用例等待过久。 os.environ["WOV_READY_TIMEOUT_SECONDS"] = "2" +# 默认关闭自动种子和后台调度，测试显式控制执行时机。 os.environ["WOV_AUTO_SEED"] = "0" os.environ["WOV_SCHEDULER_ENABLED"] = "0" def _cleanup() -> None: + """进程退出时清理临时测试目录。""" shutil.rmtree(TEST_ROOT, ignore_errors=True) diff --git a/tests/fixtures/bad_node.py b/tests/fixtures/bad_node.py index 2b766eb..499d233 100644 --- a/tests/fixtures/bad_node.py +++ b/tests/fixtures/bad_node.py @@ -1,3 +1,8 @@ +"""失败健康检查节点夹具。 + +打印就绪端口但 /health 返回 500，用于验证 NodeManager 的启动失败处理。 +""" + import http.server import threading @@ -12,6 +17,8 @@ class Handler(http.server.BaseHTTPRequestHandler): server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler) +# 打印就绪行让 NodeManager 认为进程已启动。 print(f"WOV_NODE_READY port={server.server_address[1]}", flush=True) threading.Thread(target=server.serve_forever, daemon=True).start() +# 主线程挂起，保持服务进程存活。 threading.Event().wait() diff --git a/tests/fixtures/bad_ready_node.py b/tests/fixtures/bad_ready_node.py index c30490b..17fe0e4 100644 --- a/tests/fixtures/bad_ready_node.py +++ b/tests/fixtures/bad_ready_node.py @@ -1,3 +1,8 @@ +"""错误就绪输出节点夹具。 + +打印不带端口的 WOV_NODE_READY 行，用于验证 NodeManager 对格式异常的处理。 +""" + import time print("WOV_NODE_READY", flush=True) diff --git a/tests/fixtures/bad_status_node.py b/tests/fixtures/bad_status_node.py index f12887e..606903e 100644 --- a/tests/fixtures/bad_status_node.py +++ b/tests/fixtures/bad_status_node.py @@ -1,3 +1,8 @@ +"""异常状态码健康检查节点夹具。 + +/health 返回 201，用于验证 NodeManager 只接受 200 状态码。 +""" + import http.server import threading diff --git a/tests/fixtures/failing_node.py b/tests/fixtures/failing_node.py index 54b2cef..49dc198 100644 --- a/tests/fixtures/failing_node.py +++ b/tests/fixtures/failing_node.py @@ -1,3 +1,9 @@ +"""失败调用节点夹具。 + +健康检查成功但 /invoke 始终返回 500，支持通过 WOV_FAIL_INVALID_JSON 环境变量 +切换为非法 JSON，用于验证 NodeManager 的错误响应解析分支。 +""" + import http.server import json import os @@ -19,6 +25,7 @@ class Handler(http.server.BaseHTTPRequestHandler): if os.environ.get("WOV_FAIL_INVALID_JSON") == "1": body = b"not-json" else: + # 默认返回协议格式的失败 JSON，验证错误信息透传。 body = json.dumps({"status": "failed", "error": "boom"}).encode("utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() diff --git a/tests/fixtures/stderr_node.py b/tests/fixtures/stderr_node.py index 74bb5cd..30249c1 100644 --- a/tests/fixtures/stderr_node.py +++ b/tests/fixtures/stderr_node.py @@ -1,3 +1,8 @@ +"""stderr 捕获节点夹具。 + +启动时向标准错误输出一行警告，用于验证 NodeManager 的 stderr_tail 缓存。 +""" + import http.server import sys import threading diff --git a/tests/test_api.py b/tests/test_api.py index e93cc48..d7b94d2 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -1,3 +1,9 @@ +"""应用级 API 冒烟测试。 + +使用 FastAPI TestClient 验证健康检查、静态页面、节点注册校验以及 +节点生命周期与手动调用等真实链路。 +""" + import json from pathlib import Path @@ -6,12 +12,14 @@ from fastapi.testclient import TestClient from app.main import app WORKSPACE = Path(__file__).resolve().parent.parent.parent +# 直接读取 Echo 节点 manifest 作为注册测试的真实数据。 ECHO_MANIFEST = json.loads( (WORKSPACE / "wov-node-echo" / "node.manifest.json").read_text(encoding="utf-8") ) def test_health() -> None: + """验证静态首页、OpenAPI 文档与健康探针均可访问。""" with TestClient(app) as client: root = client.get("/", follow_redirects=False) assert root.status_code == 200 @@ -26,6 +34,7 @@ def test_health() -> None: def test_register_validation_errors() -> None: + """验证非法节点注册请求会被 Pydantic 或协议校验拒绝。""" with TestClient(app) as client: assert client.post("/api/admin/nodes", json={}).status_code == 422 invalid = dict(ECHO_MANIFEST, id="") @@ -37,6 +46,7 @@ def test_register_validation_errors() -> None: def test_node_lifecycle_and_invoke() -> None: + """验证注册、查询、调用、停止与删除节点的完整生命周期。""" with TestClient(app) as client: registered = client.post("/api/admin/nodes", json=ECHO_MANIFEST) assert registered.status_code == 200 diff --git a/tests/test_apps_api.py b/tests/test_apps_api.py index fd10c78..da981d6 100644 --- a/tests/test_apps_api.py +++ b/tests/test_apps_api.py @@ -1,3 +1,9 @@ +"""用户应用 API 测试。 + +覆盖已发布应用的上传建任务、进度查询、产物下载、失败重试以及 +未发布/无版本工作流的拒绝逻辑。 +""" + import json from pathlib import Path @@ -9,6 +15,7 @@ WORKSPACE = Path(__file__).resolve().parent.parent.parent def _create_published_echo_workflow(client) -> str: + """注册 Echo 节点并创建一个已发布的单节点工作流。""" definition = { "name": "echo-flow", "version": 1, @@ -42,6 +49,7 @@ def _create_published_echo_workflow(client) -> str: def test_upload_run_progress_and_download() -> None: + """验证上传文件建任务、手动执行、查询产物与下载的完整流程。""" with TestClient(app) as client: workflow_id = _create_published_echo_workflow(client) apps = client.get("/api/apps") @@ -101,6 +109,7 @@ def test_upload_run_progress_and_download() -> None: def test_upload_rejects_unpublished_workflow() -> None: + """验证草稿或不存在的工作流不能被用户发起任务。""" with TestClient(app) as client: client.post( "/api/admin/workflows", @@ -129,6 +138,7 @@ def test_upload_rejects_unpublished_workflow() -> None: def test_upload_rejects_workflow_without_version() -> None: + """验证已发布但没有任何版本的工作流返回 422。""" with TestClient(app) as client: db = app.state.db db.upsert_workflow( @@ -142,6 +152,7 @@ def test_upload_rejects_workflow_without_version() -> None: def test_retry_failed_run_requeues_and_reruns() -> None: + """验证失败任务重试会清空旧产物并重新执行成功。""" with TestClient(app) as client: workflow_id = _create_published_echo_workflow(client) uploaded = client.post( @@ -183,6 +194,7 @@ def test_retry_failed_run_requeues_and_reruns() -> None: def test_retry_rejects_non_failed_and_missing_runs() -> None: + """验证只有 FAILED 状态且存在的任务才能重试。""" with TestClient(app) as client: workflow_id = _create_published_echo_workflow(client) uploaded = client.post( diff --git a/tests/test_db.py b/tests/test_db.py index 0fecbe1..f1ff033 100644 --- a/tests/test_db.py +++ b/tests/test_db.py @@ -1,3 +1,9 @@ +"""数据库层单元测试。 + +直接对 Database 方法调用真实 SQLite 路径，覆盖节点、实例、工作流、 +版本、任务与产物的增删改查。 +""" + from pathlib import Path from app.db import Database @@ -5,6 +11,7 @@ from wov_sdk.models import NodeManifest def manifest() -> NodeManifest: + """构造最小合法节点 manifest 供 CRUD 测试复用。""" return NodeManifest( id="echo", name="Echo", @@ -16,6 +23,7 @@ def manifest() -> NodeManifest: def test_node_crud(tmp_path) -> None: + """验证节点的注册、覆盖更新与删除。""" db = Database(tmp_path / "wov.db") assert db.get_node("echo") is None assert db.list_nodes() == [] @@ -34,6 +42,7 @@ def test_node_crud(tmp_path) -> None: def test_instance_crud(tmp_path) -> None: + """验证节点实例记录的插入、状态更新与删除。""" db = Database(tmp_path / "wov.db") db.upsert_node(manifest()) instance = { @@ -59,6 +68,7 @@ def test_instance_crud(tmp_path) -> None: def test_workflow_crud(tmp_path) -> None: + """验证工作流概要的插入、发布标记更新与删除。""" db = Database(tmp_path / "wov.db") workflow = { "id": "demo", @@ -79,6 +89,7 @@ def test_workflow_crud(tmp_path) -> None: def test_workflow_versions(tmp_path) -> None: + """验证工作流版本的写入、最新版本查询与列表。""" db = Database(tmp_path / "wov.db") db.upsert_workflow( {"id": "demo", "name": "Demo", "published": 1, "latest_version": 2} @@ -101,6 +112,7 @@ def test_workflow_versions(tmp_path) -> None: def test_run_and_artifact_crud(tmp_path) -> None: + """验证任务与产物的创建、查询、更新与删除。""" db = Database(tmp_path / "wov.db") db.upsert_workflow({"id": "demo", "name": "Demo", "published": 1, "latest_version": 1}) now = "2026-01-01T00:00:00+00:00" @@ -145,6 +157,7 @@ def test_run_and_artifact_crud(tmp_path) -> None: def test_reset_run_clears_error_and_artifacts(tmp_path) -> None: + """验证 reset_run 会把失败任务恢复到排队状态并清空旧产物。""" db = Database(tmp_path / "wov.db") db.upsert_workflow({"id": "demo", "name": "Demo", "published": 1, "latest_version": 1}) now = "2026-01-01T00:00:00+00:00" diff --git a/tests/test_node_manager.py b/tests/test_node_manager.py index 9a46fa3..9ca7dc3 100644 --- a/tests/test_node_manager.py +++ b/tests/test_node_manager.py @@ -1,3 +1,9 @@ +"""NodeManager 单元测试。 + +覆盖命令解析、环境变量注入、CUDA 路径处理、进程启动失败分支、实例复用、 +调用错误处理、回收线程以及停止清理等真实生命周期路径。 +""" + import os import subprocess import sys @@ -19,6 +25,7 @@ from app.node_manager import ( from wov_sdk.models import InvokeRequest, NodeManifest WORKSPACE = Path(__file__).resolve().parent.parent.parent +# 从真实节点仓库加载 Echo manifest，测试使用真实协议数据。 _BASE_ECHO_MANIFEST = NodeManifest.load(WORKSPACE / "wov-node-echo" / "node.manifest.json") FIXTURES = Path(__file__).resolve().parent / "fixtures" @@ -29,14 +36,17 @@ def db(tmp_path) -> Database: def _register(db: Database, manifest: NodeManifest) -> None: + """把 manifest 写入测试数据库，模拟注册节点。""" db.upsert_node(manifest) def echo_manifest() -> NodeManifest: + """返回 Echo 节点 manifest 的副本，避免测试间共享可变对象。""" return NodeManifest.from_dict(_BASE_ECHO_MANIFEST.to_dict()) def test_time_helpers() -> None: + """验证时间工具对正常、空与非法输入的返回。""" now = _now_iso() assert datetime.fromisoformat(now) assert _idle_seconds(now) < 0.1 @@ -47,6 +57,7 @@ def test_time_helpers() -> None: def test_resolve_command_and_env(db: Database) -> None: + """验证 python 命令解析回退与固定命令保持原样。""" manager = NodeManager(db) python_command = NodeManifest( id="x", @@ -71,6 +82,7 @@ def test_resolve_command_and_env(db: Database) -> None: def test_resolve_command_prefers_node_venv(db: Database, tmp_path, monkeypatch) -> None: + """验证 Windows 与 Linux 虚拟环境解释器优先级。""" monkeypatch.setattr("app.node_manager.WORKSPACE_ROOT", tmp_path) repo = tmp_path / "wov-node-demo" python_exe = repo / ".venv" / "Scripts" / "python.exe" @@ -95,6 +107,7 @@ def test_resolve_command_prefers_node_venv(db: Database, tmp_path, monkeypatch) def test_cuda_library_dirs_finds_unix_nvidia_libs(tmp_path) -> None: + """验证 Linux site-packages 下的 cublas/cudnn lib 目录被发现。""" cublas = ( tmp_path / ".venv" / "lib" / "python3.12" / "site-packages" / "nvidia" / "cublas" / "lib" ) @@ -111,6 +124,7 @@ def test_cuda_library_dirs_finds_unix_nvidia_libs(tmp_path) -> None: def test_cuda_library_dirs_finds_windows_nvidia_bins(tmp_path) -> None: + """验证 Windows Lib/site-packages 下的 bin 目录被发现。""" cublas = tmp_path / ".venv" / "Lib" / "site-packages" / "nvidia" / "cublas" / "bin" cublas.mkdir(parents=True) @@ -120,6 +134,7 @@ def test_cuda_library_dirs_finds_windows_nvidia_bins(tmp_path) -> None: def test_with_cuda_library_path_prepends_on_posix(monkeypatch) -> None: + """验证 Linux 下 CUDA 目录插入 LD_LIBRARY_PATH 开头。""" monkeypatch.setattr(os, "name", "posix") monkeypatch.setattr( "app.node_manager._cuda_library_dirs", @@ -134,6 +149,7 @@ def test_with_cuda_library_path_prepends_on_posix(monkeypatch) -> None: def test_with_cuda_library_path_uses_path_on_windows(monkeypatch) -> None: + """验证 Windows 下 CUDA 目录插入 PATH 开头。""" monkeypatch.setattr(os, "name", "nt") monkeypatch.setattr( "app.node_manager._cuda_library_dirs", @@ -148,6 +164,7 @@ def test_with_cuda_library_path_uses_path_on_windows(monkeypatch) -> None: def test_with_cuda_library_path_without_existing(monkeypatch) -> None: + """验证环境变量原本不存在时直接创建。""" monkeypatch.setattr(os, "name", "posix") monkeypatch.setattr( "app.node_manager._cuda_library_dirs", @@ -160,6 +177,7 @@ def test_with_cuda_library_path_without_existing(monkeypatch) -> None: def test_with_cuda_library_path_deduplicates(monkeypatch) -> None: + """验证已存在的目录不会被重复追加。""" monkeypatch.setattr(os, "name", "posix") monkeypatch.setattr( "app.node_manager._cuda_library_dirs", @@ -176,6 +194,7 @@ def test_with_cuda_library_path_deduplicates(monkeypatch) -> None: def test_with_cuda_library_path_all_present(monkeypatch) -> None: + """验证所有目录都已存在时环境变量保持不变。""" monkeypatch.setattr(os, "name", "posix") monkeypatch.setattr( "app.node_manager._cuda_library_dirs", @@ -189,6 +208,7 @@ def test_with_cuda_library_path_all_present(monkeypatch) -> None: def test_with_cuda_library_path_without_libs(monkeypatch) -> None: + """验证没有任何 CUDA 库时环境变量原样返回。""" monkeypatch.setattr("app.node_manager._cuda_library_dirs", lambda repo_dir: []) env = _with_cuda_library_path({"LD_LIBRARY_PATH": "/usr/lib/foo"}, Path("/repo")) @@ -197,6 +217,7 @@ def test_with_cuda_library_path_without_libs(monkeypatch) -> None: def test_node_env_injects_cuda_library_path(db: Database, tmp_path, monkeypatch) -> None: + """验证 _node_env 会注入 SDK、manifest 变量与 CUDA 路径。""" monkeypatch.setattr("app.node_manager.WORKSPACE_ROOT", tmp_path) repo = tmp_path / "wov-node-whisper" cublas = repo / ".venv" / "lib" / "python3.12" / "site-packages" / "nvidia" / "cublas" / "lib" @@ -219,12 +240,14 @@ def test_node_env_injects_cuda_library_path(db: Database, tmp_path, monkeypatch) def test_acquire_unregistered(db: Database) -> None: + """验证未注册节点无法获取实例。""" manager = NodeManager(db) with pytest.raises(ValueError): manager.acquire("missing") def test_start_missing_repo(db: Database) -> None: + """验证 repo_dir 不存在时启动失败并记录 error。""" manifest = echo_manifest() manifest.repo_dir = "missing-repo" _register(db, manifest) @@ -235,6 +258,7 @@ def test_start_missing_repo(db: Database) -> None: def test_start_command_not_found(db: Database) -> None: + """验证命令不存在时启动失败并记录 error。""" manifest = echo_manifest() manifest.command = ["definitely-not-a-real-wov-command"] _register(db, manifest) @@ -245,6 +269,7 @@ def test_start_command_not_found(db: Database) -> None: def test_start_no_ready_timeout(db: Database, monkeypatch) -> None: + """验证超时未打印就绪行时启动失败。""" monkeypatch.setattr("app.node_manager.NODE_READY_TIMEOUT_SECONDS", 0.2) manifest = echo_manifest() manifest.command = ["python", "-c", "import time; time.sleep(0.5)"] @@ -256,6 +281,7 @@ def test_start_no_ready_timeout(db: Database, monkeypatch) -> None: def test_start_bad_ready_line(db: Database) -> None: + """验证就绪行缺少端口时启动失败。""" manifest = echo_manifest() manifest.command = ["python", "-u", str(FIXTURES / "bad_ready_node.py")] _register(db, manifest) @@ -266,6 +292,7 @@ def test_start_bad_ready_line(db: Database) -> None: def test_start_health_check_failure(db: Database) -> None: + """验证 /health 返回错误状态码时启动失败。""" manifest = echo_manifest() manifest.command = ["python", "-u", str(FIXTURES / "bad_node.py")] _register(db, manifest) @@ -276,6 +303,7 @@ def test_start_health_check_failure(db: Database) -> None: def test_start_health_check_bad_status(db: Database) -> None: + """验证 /health 返回非 200 状态码时启动失败。""" manifest = echo_manifest() manifest.command = ["python", "-u", str(FIXTURES / "bad_status_node.py")] _register(db, manifest) @@ -286,6 +314,7 @@ def test_start_health_check_bad_status(db: Database) -> None: def test_stderr_is_captured(db: Database) -> None: + """验证节点 stderr 会被缓存到运行时供诊断。""" manifest = echo_manifest() manifest.command = ["python", "-u", str(FIXTURES / "stderr_node.py")] _register(db, manifest) @@ -301,6 +330,7 @@ def test_stderr_is_captured(db: Database) -> None: def test_acquire_reuses_ready_instance(db: Database) -> None: + """验证释放后的就绪实例会被再次复用。""" _register(db, echo_manifest()) manager = NodeManager(db) try: @@ -315,6 +345,7 @@ def test_acquire_reuses_ready_instance(db: Database) -> None: def test_invoke_success(db: Database, tmp_path) -> None: + """验证真实节点调用返回完成状态与输出。""" _register(db, echo_manifest()) manager = NodeManager(db) try: @@ -334,6 +365,7 @@ def test_invoke_success(db: Database, tmp_path) -> None: def test_invoke_http_error(db: Database, tmp_path) -> None: + """验证节点返回协议 JSON 错误时透传错误信息。""" manifest = echo_manifest() manifest.command = ["python", "-u", str(FIXTURES / "failing_node.py")] _register(db, manifest) @@ -355,6 +387,7 @@ def test_invoke_http_error(db: Database, tmp_path) -> None: def test_invoke_http_error_invalid_json(db: Database, tmp_path) -> None: + """验证节点返回非法 JSON 时降级为通用错误信息。""" manifest = echo_manifest() manifest.command = ["python", "-u", str(FIXTURES / "failing_node.py")] manifest.env["WOV_FAIL_INVALID_JSON"] = "1" @@ -377,6 +410,7 @@ def test_invoke_http_error_invalid_json(db: Database, tmp_path) -> None: def test_invoke_network_error(db: Database, tmp_path) -> None: + """验证进程退出导致连接失败时返回 failed。""" _register(db, echo_manifest()) manager = NodeManager(db) try: @@ -399,6 +433,7 @@ def test_invoke_network_error(db: Database, tmp_path) -> None: def test_invoke_uses_configured_timeout(db: Database, monkeypatch) -> None: + """验证调用超时取自配置常量。""" _register(db, echo_manifest()) manager = NodeManager(db) captured = {} @@ -444,16 +479,19 @@ def test_invoke_uses_configured_timeout(db: Database, monkeypatch) -> None: def test_release_unknown(db: Database) -> None: + """验证释放未知实例是无害操作。""" manager = NodeManager(db) manager.release("missing") def test_stop_unknown_instance(db: Database) -> None: + """验证停止未知实例是无害操作。""" manager = NodeManager(db) manager.stop_instance("missing") def test_stop_all_for_node(db: Database) -> None: + """验证 stop_all_for_node 会停止该节点全部实例。""" _register(db, echo_manifest()) manager = NodeManager(db) runtime, _ = manager.acquire("echo") @@ -463,6 +501,7 @@ def test_stop_all_for_node(db: Database) -> None: def test_stop_locked_without_process(db: Database) -> None: + """验证无进程句柄的实例也能被标记停止。""" _register(db, echo_manifest()) manager = NodeManager(db) runtime = NodeRuntime( @@ -478,6 +517,7 @@ def test_stop_locked_without_process(db: Database) -> None: def test_stop_process_already_exited() -> None: + """验证进程已退出时停止流程直接返回。""" class FakeProcess: def poll(self): return 0 @@ -487,6 +527,7 @@ def test_stop_process_already_exited() -> None: def test_stop_process_timeout() -> None: + """验证优雅终止超时后强制 kill。""" class FakeProcess: def poll(self): return None @@ -510,6 +551,7 @@ def test_stop_process_timeout() -> None: def test_reaper_recycles_idle(db: Database, tmp_path) -> None: + """验证空闲超过 TTL 的实例会被回收线程停止。""" manifest = echo_manifest() manifest.idle_ttl_seconds = 0 _register(db, manifest) @@ -533,6 +575,7 @@ def test_reaper_recycles_idle(db: Database, tmp_path) -> None: def test_reaper_keeps_warm(db: Database, tmp_path) -> None: + """验证 keep_warm 实例即使空闲也不会被回收。""" manifest = echo_manifest() manifest.idle_ttl_seconds = 0 manifest.keep_warm = True @@ -557,6 +600,7 @@ def test_reaper_keeps_warm(db: Database, tmp_path) -> None: def test_reaper_keeps_busy(db: Database) -> None: + """验证正在被占用的实例不会被回收。""" manifest = echo_manifest() manifest.idle_ttl_seconds = 0 _register(db, manifest) diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py index 34ed02f..6e69157 100644 --- a/tests/test_scheduler.py +++ b/tests/test_scheduler.py @@ -1,3 +1,9 @@ +"""调度器单元测试。 + +覆盖拓扑排序、任务执行成功/失败分支、输入引用解析、MIME 推断以及 +后台轮询线程的启动与停止。 +""" + import time from pathlib import Path @@ -15,10 +21,12 @@ from wov_sdk.models import ( def _db(tmp_path) -> Database: + """在临时目录创建独立数据库。""" return Database(tmp_path / "wov.db") def _echo_definition() -> WorkflowDefinition: + """构造引用 Echo 节点的单步骤工作流定义。""" return WorkflowDefinition( name="echo-flow", version=1, @@ -36,6 +44,7 @@ def _echo_definition() -> WorkflowDefinition: def test_topological_sort() -> None: + """验证 DAG 排序保持依赖顺序，并拒绝环与未知边。""" definition = WorkflowDefinition( name="dag", version=1, @@ -80,6 +89,7 @@ def test_topological_sort() -> None: 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") @@ -117,6 +127,7 @@ def test_execute_echo_workflow(tmp_path) -> None: 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" @@ -142,6 +153,7 @@ def test_execute_run_missing_workflow(tmp_path, monkeypatch) -> None: 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" @@ -166,6 +178,7 @@ def test_execute_run_missing_version(tmp_path) -> None: def test_execute_run_missing_node(tmp_path) -> None: + """验证未注册节点被调用时任务失败。""" db = _db(tmp_path) definition = WorkflowDefinition( name="bad", @@ -204,6 +217,7 @@ def test_execute_run_missing_node(tmp_path) -> None: def test_resolve_ref_and_mime(tmp_path) -> None: + """验证输入引用解析、MIME 推断与文件大小读取。""" db = _db(tmp_path) manager = NodeManager(db) scheduler = WorkflowScheduler(db, manager, tmp_path / "storage") @@ -230,6 +244,7 @@ def test_resolve_ref_and_mime(tmp_path) -> None: 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" @@ -255,6 +270,7 @@ def test_execute_unknown_or_non_queued_run(tmp_path) -> None: def test_execute_missing_input(tmp_path) -> None: + """验证输入引用无法解析时任务失败。""" db = _db(tmp_path) definition = WorkflowDefinition( name="missing-input", @@ -293,6 +309,7 @@ def test_execute_missing_input(tmp_path) -> None: def test_execute_node_failed_response(tmp_path) -> None: + """验证节点返回 failed 时任务被标记为失败。""" db = _db(tmp_path) manifest = NodeManifest( id="fail-node", @@ -340,6 +357,7 @@ def test_execute_node_failed_response(tmp_path) -> None: 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) @@ -353,6 +371,7 @@ def test_scheduler_start_stop_loop(tmp_path) -> 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") diff --git a/tests/test_seed.py b/tests/test_seed.py index 6307132..383850c 100644 --- a/tests/test_seed.py +++ b/tests/test_seed.py @@ -1,3 +1,9 @@ +"""种子数据测试。 + +验证启动种子逻辑会注册节点仓库、创建演示工作流且保持幂等，并验证 +开启种子与调度器后的应用生命周期。 +""" + from pathlib import Path from fastapi.testclient import TestClient @@ -10,6 +16,7 @@ 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()] @@ -27,10 +34,12 @@ def test_seed_nodes_and_demo_workflow(tmp_path) -> None: 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: diff --git a/tests/test_uvicorn_smoke.py b/tests/test_uvicorn_smoke.py index 2038a6f..d4e34e1 100644 --- a/tests/test_uvicorn_smoke.py +++ b/tests/test_uvicorn_smoke.py @@ -1,3 +1,9 @@ +"""uvicorn 冒烟测试。 + +用真实套接字启动 uvicorn 服务并请求 /health，验证应用能脱离 TestClient +在实际 Web 服务环境中正常工作。 +""" + import threading import time import urllib.request @@ -8,6 +14,7 @@ from app.main import app def test_uvicorn_serves_app_over_real_socket() -> None: + """验证 uvicorn 监听真实端口后健康检查可用。""" config = Config(app=app, host="127.0.0.1", port=0, log_level="error") server = Server(config) thread = threading.Thread(target=server.run, daemon=True) diff --git a/tests/test_workflow_api.py b/tests/test_workflow_api.py index 47f6a52..09a9d85 100644 --- a/tests/test_workflow_api.py +++ b/tests/test_workflow_api.py @@ -1,9 +1,15 @@ +"""工作流管理 API 测试。 + +覆盖工作流的创建、查询、校验、发布、版本列表与删除等管理接口。 +""" + from fastapi.testclient import TestClient from app.main import app def definition() -> dict: + """构造一个引用 Echo 节点的合法工作流定义。""" return { "name": "echo-flow", "version": 1, @@ -21,6 +27,7 @@ def definition() -> dict: def test_workflow_crud_and_publish() -> None: + """验证工作流 CRUD、校验、发布与版本列表的完整流程。""" with TestClient(app) as client: created = client.post( "/api/admin/workflows", @@ -64,6 +71,7 @@ def test_workflow_crud_and_publish() -> None: def test_workflow_slug_without_id() -> None: + """验证未提供 ID 时后端会从名称生成 slug。""" with TestClient(app) as client: created = client.post( "/api/admin/workflows", @@ -77,6 +85,7 @@ def test_workflow_slug_without_id() -> None: def test_workflow_validation_error() -> None: + """验证重复节点 ID 的 DAG 会被拒绝。""" with TestClient(app) as client: response = client.post( "/api/admin/workflows", @@ -98,6 +107,7 @@ def test_workflow_validation_error() -> None: def test_publish_workflow_without_version() -> None: + """验证没有版本记录的工作流不能发布。""" with TestClient(app) as client: db = app.state.db db.upsert_workflow(