"""src/wov_app/main.py 与 src/wov_app/schemas.py 的模块级测试。 被测模块： - `src/wov_app/main.py`：FastAPI 应用装配（生命周期、路由挂载、静态前端、 健康检查、重启恢复）； - `src/wov_app/schemas.py`：管理端请求模型（Pydantic）。 用例通过真实 TestClient 触发完整生命周期（启动/关闭），验证后台服务被正确 创建与回收、恢复逻辑被调用、静态前端可访问。 """ from __future__ import annotations from pathlib import Path import pytest from fastapi.testclient import TestClient from wov_app.main import app as fastapi_app from wov_app.schemas import BatchJobCreate, WorkflowCreate @pytest.fixture() def client(tmp_path: Path, monkeypatch) -> TestClient: """隔离数据库与存储的真实 TestClient（关闭后台线程便于断言状态）。""" monkeypatch.setattr("wov_app.config.DB_PATH", tmp_path / "wov.db") monkeypatch.setattr("wov_app.config.STORAGE_DIR", tmp_path / "storage") monkeypatch.setattr("wov_app.main.DB_PATH", tmp_path / "wov.db") monkeypatch.setattr("wov_app.main.STORAGE_DIR", tmp_path / "storage") monkeypatch.setattr("wov_app.batch.BATCH_WORK_ROOT", tmp_path / "storage" / "batch") monkeypatch.setenv("WOV_AUTO_SEED", "0") monkeypatch.setenv("WOV_SCHEDULER_ENABLED", "0") monkeypatch.setenv("WOV_CLEANUP_ENABLED", "0") monkeypatch.setenv("WOV_BATCH_ENABLED", "0") with TestClient(fastapi_app) as test_client: yield test_client # --------------------------------------------------------------------------- # 生命周期装配 # --------------------------------------------------------------------------- def test_lifespan_registers_nodes_and_state(client: TestClient) -> None: """启动后节点注册表就绪，且 app.state 上挂好了库/调度器/清理器/批量引擎。""" # 数据：真实应用生命周期。 # 测试过程 state = client.app.state nodes = {m.id for m in __import__("wov_app.registry", fromlist=["list_nodes"]).list_nodes()} # 验证结果 assert "echo" in nodes and "faster-whisper" in nodes assert state.db is not None assert state.scheduler is not None assert state.cleaner is not None assert state.batch is not None def test_lifespan_recovers_interrupted_runs(tmp_path: Path, monkeypatch) -> None: """重启恢复：遗留 RUNNING 任务在启动时被恢复为 QUEUED。""" # 数据：预先在目标库里写入一条 RUNNING 任务（含工作流与版本）。 from wov_app.db import Database db_path = tmp_path / "wov.db" db = Database(db_path) db.upsert_workflow({"id": "wf", "name": "x", "description": "", "published": 1, "latest_version": 1}) db.create_workflow_version("wf", 1, {"nodes": [], "edges": []}) db.create_run({ "id": "run-stale", "workflow_id": "wf", "workflow_version": 1, "status": "RUNNING", "current_node_id": None, "progress": 0.5, "error": None, "input_uri": None, "param_overrides": None, "source": "upload", "created_at": "2026-09-01T00:00:00+00:00", "updated_at": "2026-09-01T00:00:00+00:00", }) monkeypatch.setattr("wov_app.config.DB_PATH", db_path) monkeypatch.setattr("wov_app.main.DB_PATH", db_path) monkeypatch.setattr("wov_app.config.STORAGE_DIR", tmp_path / "storage") monkeypatch.setattr("wov_app.main.STORAGE_DIR", tmp_path / "storage") monkeypatch.setenv("WOV_AUTO_SEED", "0") monkeypatch.setenv("WOV_SCHEDULER_ENABLED", "0") monkeypatch.setenv("WOV_CLEANUP_ENABLED", "0") monkeypatch.setenv("WOV_BATCH_ENABLED", "0") # 测试过程：进入生命周期触发恢复逻辑。 with TestClient(fastapi_app): recovered = Database(db_path).get_run("run-stale") # 验证结果 assert recovered["status"] == "QUEUED" def test_lifespan_seeds_workflows_when_enabled(tmp_path: Path, monkeypatch) -> None: """开启自动种子时启动创建内置工作流（数据驱动）。""" # 数据：目标库 + 开启 seed。 db_path = tmp_path / "wov.db" monkeypatch.setattr("wov_app.config.DB_PATH", db_path) monkeypatch.setattr("wov_app.main.DB_PATH", db_path) monkeypatch.setattr("wov_app.config.STORAGE_DIR", tmp_path / "storage") monkeypatch.setattr("wov_app.main.STORAGE_DIR", tmp_path / "storage") monkeypatch.setenv("WOV_AUTO_SEED", "1") monkeypatch.setenv("WOV_SCHEDULER_ENABLED", "0") monkeypatch.setenv("WOV_CLEANUP_ENABLED", "0") monkeypatch.setenv("WOV_BATCH_ENABLED", "0") # 测试过程 with TestClient(fastapi_app): from wov_app.db import Database as _Db ids = {w["id"] for w in _Db(db_path).list_workflows()} # 验证结果：内置工作流全部就位。 assert {"zh-direct", "ocr-subtitle", "learn-translate"} <= ids def test_lifespan_starts_and_stops_background_services(tmp_path: Path, monkeypatch) -> None: """开启后台服务时启动线程，退出时全部停止（无残留线程）。""" # 数据：全部后台服务开启。 monkeypatch.setattr("wov_app.config.DB_PATH", tmp_path / "wov.db") monkeypatch.setattr("wov_app.main.DB_PATH", tmp_path / "wov.db") monkeypatch.setattr("wov_app.config.STORAGE_DIR", tmp_path / "storage") monkeypatch.setattr("wov_app.main.STORAGE_DIR", tmp_path / "storage") monkeypatch.setattr("wov_app.batch.BATCH_WORK_ROOT", tmp_path / "storage" / "batch") monkeypatch.setenv("WOV_AUTO_SEED", "0") monkeypatch.setenv("WOV_SCHEDULER_ENABLED", "1") monkeypatch.setenv("WOV_CLEANUP_ENABLED", "1") monkeypatch.setenv("WOV_BATCH_ENABLED", "1") # 测试过程 with TestClient(fastapi_app) as test_client: scheduler_thread = test_client.app.state.scheduler._thread cleaner_thread = test_client.app.state.cleaner._thread batch_thread = test_client.app.state.batch._thread assert scheduler_thread is not None and scheduler_thread.is_alive() assert cleaner_thread is not None and cleaner_thread.is_alive() assert batch_thread is not None and batch_thread.is_alive() # 验证结果：退出后线程引用被清空（stop 已执行）。 assert test_client.app.state.scheduler._thread is None assert test_client.app.state.cleaner._thread is None assert test_client.app.state.batch._thread is None # --------------------------------------------------------------------------- # 路由与静态前端 # --------------------------------------------------------------------------- def test_health_endpoint(client: TestClient) -> None: """健康检查返回 ok 与模式标识（部署探针依赖）。""" # 数据：无。 # 测试过程 response = client.get("/health") # 验证结果 assert response.status_code == 200 body = response.json() assert body["status"] == "ok" assert body["mode"] == "monolith" def test_static_frontend_is_mounted(client: TestClient) -> None: """静态前端挂载在根路径，首页可访问（真实 web/ 目录）。""" # 数据：真实前端文件。 # 测试过程 response = client.get("/") # 验证结果 assert response.status_code == 200 assert "text/html" in response.headers["content-type"] def test_openapi_lists_routers(client: TestClient) -> None: """OpenAPI 文档包含三组路由（管理端/用户端/批量）。""" # 数据：无。 # 测试过程 paths = client.get("/openapi.json").json()["paths"] # 验证结果：每组至少一个端点。 assert any(path.startswith("/api/admin/workflows") for path in paths) assert any(path.startswith("/api/apps") for path in paths) assert any(path.startswith("/api/batch") for path in paths) # --------------------------------------------------------------------------- # 请求模型（schemas） # --------------------------------------------------------------------------- def test_workflow_create_schema_defaults() -> None: """WorkflowCreate：id/description 可省略，definition 必填。""" # 数据：最小合法载荷。 model = WorkflowCreate(name="流程", definition={"nodes": [], "edges": []}) # 测试过程与验证结果 assert model.id is None assert model.description == "" assert model.definition == {"nodes": [], "edges": []} def test_batch_job_create_schema_defaults() -> None: """BatchJobCreate：recursive 默认 True（批量页默认递归扫描）。""" # 数据：最小载荷。 model = BatchJobCreate(folder="/videos", workflow_id="wf") # 测试过程与验证结果 assert model.recursive is True assert model.folder == "/videos" def test_schemas_reject_missing_required_fields() -> None: """缺少必填字段时 Pydantic 校验失败（由 FastAPI 转 422）。""" # 数据：缺少 name / folder。 # 测试过程与验证结果 with pytest.raises(Exception): WorkflowCreate(definition={}) with pytest.raises(Exception): BatchJobCreate(workflow_id="wf")