cat-shark
8a715a8064
按"测试规则"重写 tests/：一个模块一个目录、用例按数据→过程→验证三段书写、 不保留全局 conftest.py、测试过程只调用真实生产代码。 结构（73 个文件、30 个模块目录、477 用例）： - tests/nodes/ 15 个模块目录（srt/whisper/ass/ffmpeg/frame_extract/vlm/ subtitle_ocr/llm/llm_filter/subtitle_cleanup/subtitle_correction/ proper_nouns/adaptive_pool/vad_profiler/echo）； - tests/app/ 11 个模块目录（db/scheduler/batch/maintenance/registry/seed/ storage/config/logging/main/routers 三组 API）； - tests/sdk/test_models、tests/web/test_crop、tests/shared（公共设施）。 测试数据随模块目录入库（tests/**/data/），删除根级 testdata/；.gitignore 的 data/ 改为 /data/，否则会连带忽略 tests/**/data/ 导致测试数据无法入库。 顺带发现并修复三个真实缺陷： - nodes/srt.py：相邻条目缺少空行时把下一条时间轴吞进正文（静默错位）， 改为正文行遇时间戳行即报错； - src/wov_app/scheduler.py：_file_size 只捕获 OSError，含 \x00 的产物 URI 抛 ValueError 导致任务误判失败，改为同时捕获； - nodes/subtitle_correction.py：生产代码依赖测试包解析 SRT， 改用生产模块 nodes/srt.py。 真实模型/服务集成测试按外部状态跳过：新增 tests/shared/gpu_memory.py （运行时探测显存、CUDA OOM 转跳过）与 tests/shared/llm_service.py （无 Key / 余额 / 限流转跳过）。全量 477 passed。
217 lines
8.8 KiB
Python
217 lines
8.8 KiB
Python
"""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")
|