cat-shark
c063aed1f1
事故（batch_969fabe74b83 等 3 个任务实测）：批量引擎串行处理到 9.9GB 大视频时中断，_run_job 无条件收尾把任务置 COMPLETED，留下"N 个 PENDING 待处理却已完成"的僵尸状态——3 个任务遗留的 PENDING 完全相同（kiwvr-887/ mdvr-413/savr-1077 共 10 个大视频），且均卡在大文件前从未被真正处理。 - batch.py: _run_job 置 COMPLETED 前校验全部非 SKIPPED 视频已结束 （无 PENDING/PAUSED 残留），否则保持 RUNNING 交引擎下一轮续跑 - db.py/main.py: 新增 recover_interrupted_batch_jobs，重启时把 RUNNING 批量任务恢复 QUEUED（否则停在 RUNNING 永远不会被再次拾起） - tests: 新增 2 条 TDD 回归（残留 PENDING 不置 COMPLETED、批量崩溃恢复） - scripts/fix_zombie_batch_jobs.py: 历史僵尸数据修复脚本（置回 QUEUED 续跑） - AGENTS.md: 补充完成任务判定与崩溃恢复约定
465 lines
17 KiB
Python
465 lines
17 KiB
Python
"""数据库层单元测试。
|
||
|
||
直接对 Database 方法调用真实 SQLite 路径，覆盖工作流、版本、任务与产物的
|
||
增删改查。节点注册表已改为进程内内存态，不再落库。
|
||
"""
|
||
|
||
from pathlib import Path
|
||
|
||
from wov_app.db import Database
|
||
|
||
|
||
def test_workflow_crud(tmp_path) -> None:
|
||
"""验证工作流概要的插入、发布标记更新与删除。"""
|
||
db = Database(tmp_path / "wov.db")
|
||
workflow = {
|
||
"id": "demo",
|
||
"name": "Demo",
|
||
"description": "desc",
|
||
"published": 0,
|
||
"latest_version": 0,
|
||
}
|
||
db.upsert_workflow(workflow)
|
||
assert db.get_workflow("demo")["name"] == "Demo"
|
||
assert [item["id"] for item in db.list_workflows()] == ["demo"]
|
||
|
||
db.upsert_workflow({**workflow, "published": 1, "latest_version": 1})
|
||
assert db.get_workflow("demo")["published"] == 1
|
||
|
||
db.delete_workflow("demo")
|
||
assert db.get_workflow("demo") is 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}
|
||
)
|
||
definition = {"name": "Demo", "version": 1, "nodes": [], "edges": []}
|
||
db.create_workflow_version("demo", 1, definition)
|
||
db.create_workflow_version("demo", 2, {**definition, "version": 2})
|
||
|
||
latest = db.get_latest_workflow_version("demo")
|
||
assert latest["version"] == 2
|
||
assert latest["definition"]["version"] == 2
|
||
|
||
version = db.get_workflow_version("demo", 1)
|
||
assert version["version"] == 1
|
||
assert db.get_workflow_version("demo", 99) is None
|
||
assert len(db.list_workflow_versions("demo")) == 2
|
||
|
||
empty_db = Database(tmp_path / "empty.db")
|
||
assert empty_db.get_latest_workflow_version("missing") is 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"
|
||
db.create_run(
|
||
{
|
||
"id": "run_1",
|
||
"workflow_id": "demo",
|
||
"workflow_version": 1,
|
||
"status": "QUEUED",
|
||
"progress": 0,
|
||
"input_uri": "in.txt",
|
||
"created_at": now,
|
||
"updated_at": now,
|
||
}
|
||
)
|
||
assert db.get_run("run_1")["status"] == "QUEUED"
|
||
assert db.next_queued_run()["id"] == "run_1"
|
||
|
||
db.update_run("run_1", status="RUNNING", progress=0.5, updated_at=now)
|
||
db.update_run("run_1")
|
||
assert db.get_run("run_1")["status"] == "RUNNING"
|
||
assert db.get_run("run_1")["progress"] == 0.5
|
||
assert db.next_queued_run() is None
|
||
assert len(db.list_runs()) == 1
|
||
|
||
db.create_artifact(
|
||
{
|
||
"run_id": "run_1",
|
||
"node_id": "echo",
|
||
"name": "result",
|
||
"uri": "out.txt",
|
||
"mime_type": "text/plain",
|
||
"size": 3,
|
||
}
|
||
)
|
||
assert db.get_artifact("run_1", "result")["uri"] == "out.txt"
|
||
assert db.get_artifact("run_1", "missing") is None
|
||
assert len(db.list_artifacts("run_1")) == 1
|
||
|
||
db.delete_run_artifacts("run_1")
|
||
assert db.list_artifacts("run_1") == []
|
||
|
||
|
||
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"
|
||
db.create_run(
|
||
{
|
||
"id": "run_1",
|
||
"workflow_id": "demo",
|
||
"workflow_version": 1,
|
||
"status": "FAILED",
|
||
"progress": 0.75,
|
||
"current_node_id": "translate",
|
||
"error": "timed out",
|
||
"input_uri": "in.txt",
|
||
"created_at": now,
|
||
"updated_at": now,
|
||
}
|
||
)
|
||
db.create_artifact(
|
||
{
|
||
"run_id": "run_1",
|
||
"node_id": "asr",
|
||
"name": "asr.srt_uri",
|
||
"uri": "out.srt",
|
||
"mime_type": "application/x-subrip",
|
||
"size": 3,
|
||
}
|
||
)
|
||
|
||
db.reset_run("run_1", "2026-01-02T00:00:00+00:00")
|
||
|
||
run = db.get_run("run_1")
|
||
assert run["status"] == "QUEUED"
|
||
assert run["progress"] == 0
|
||
assert run["current_node_id"] is None
|
||
assert run["error"] is None
|
||
assert run["updated_at"] == "2026-01-02T00:00:00+00:00"
|
||
assert run["created_at"] == now
|
||
assert db.list_artifacts("run_1") == []
|
||
|
||
|
||
def test_delete_run(tmp_path) -> None:
|
||
"""验证 delete_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"
|
||
db.create_run(
|
||
{
|
||
"id": "run_1",
|
||
"workflow_id": "demo",
|
||
"workflow_version": 1,
|
||
"status": "COMPLETED",
|
||
"progress": 1,
|
||
"input_uri": "in.txt",
|
||
"created_at": now,
|
||
"updated_at": now,
|
||
}
|
||
)
|
||
db.create_artifact(
|
||
{
|
||
"run_id": "run_1",
|
||
"node_id": "asr",
|
||
"name": "asr.srt_uri",
|
||
"uri": "out.srt",
|
||
"mime_type": "application/x-subrip",
|
||
"size": 3,
|
||
}
|
||
)
|
||
db.delete_run("run_1")
|
||
assert db.get_run("run_1") is None
|
||
assert db.list_artifacts("run_1") == []
|
||
|
||
|
||
def test_list_run_ids(tmp_path) -> None:
|
||
"""验证 list_run_ids 返回全部任务 ID，供孤儿清理对照使用。"""
|
||
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"
|
||
assert db.list_run_ids() == []
|
||
for run_id in ("run_a", "run_b"):
|
||
db.create_run(
|
||
{
|
||
"id": run_id,
|
||
"workflow_id": "demo",
|
||
"workflow_version": 1,
|
||
"status": "QUEUED",
|
||
"progress": 0,
|
||
"created_at": now,
|
||
"updated_at": now,
|
||
}
|
||
)
|
||
assert set(db.list_run_ids()) == {"run_a", "run_b"}
|
||
|
||
|
||
def test_run_param_overrides_persist(tmp_path) -> None:
|
||
"""验证 param_overrides 随任务持久化并可读回。"""
|
||
db = Database(tmp_path / "wov.db")
|
||
db.upsert_workflow({"id": "demo", "name": "D", "published": 1, "latest_version": 1})
|
||
now = "2026-01-01T00:00:00+00:00"
|
||
db.create_run(
|
||
{
|
||
"id": "run_ov",
|
||
"workflow_id": "demo",
|
||
"workflow_version": 1,
|
||
"status": "QUEUED",
|
||
"progress": 0,
|
||
"param_overrides": {"extract": {"crop": [0, 0.5, 1, 0.5]}},
|
||
"created_at": now,
|
||
"updated_at": now,
|
||
}
|
||
)
|
||
run = db.get_run("run_ov")
|
||
assert run["param_overrides"] == {"extract": {"crop": [0, 0.5, 1, 0.5]}}
|
||
assert db.next_queued_run()["param_overrides"] == {"extract": {"crop": [0, 0.5, 1, 0.5]}}
|
||
|
||
|
||
def test_db_migration_adds_param_overrides(tmp_path) -> None:
|
||
"""旧库迁移：缺少 param_overrides 列的库打开后自动补列。"""
|
||
import sqlite3
|
||
|
||
db_path = tmp_path / "old.db"
|
||
conn = sqlite3.connect(db_path)
|
||
conn.execute(
|
||
"CREATE TABLE workflow_runs (id TEXT PRIMARY KEY, workflow_id TEXT NOT NULL,"
|
||
" workflow_version INTEGER NOT NULL, status TEXT NOT NULL, current_node_id TEXT,"
|
||
" progress REAL NOT NULL DEFAULT 0, error TEXT, input_uri TEXT,"
|
||
" created_at TEXT NOT NULL, updated_at TEXT NOT NULL)"
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
Database(db_path)
|
||
conn = sqlite3.connect(db_path)
|
||
columns = [row[1] for row in conn.execute("PRAGMA table_info(workflow_runs)")]
|
||
conn.close()
|
||
assert "param_overrides" in columns
|
||
|
||
|
||
def test_pause_resume_run(tmp_path) -> None:
|
||
"""验证 pause_run/resume_run 的状态流转与 PAUSED 任务不被调度器自动拾起。
|
||
|
||
修复回归：PAUSED 任务若被 next_queued_run 取到，execute_run 会把它复活为
|
||
RUNNING 继续执行——"点击暂停反而开始任务"。暂停必须由用户显式 resume
|
||
（PAUSED → QUEUED）后调度器才重新执行。
|
||
"""
|
||
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"
|
||
db.create_run(
|
||
{
|
||
"id": "run_p",
|
||
"workflow_id": "demo",
|
||
"workflow_version": 1,
|
||
"status": "QUEUED",
|
||
"progress": 0,
|
||
"created_at": now,
|
||
"updated_at": now,
|
||
}
|
||
)
|
||
db.pause_run("run_p", now)
|
||
assert db.get_run("run_p")["status"] == "PAUSED"
|
||
# 已暂停的任务不会被调度器拾起（等待用户显式 resume）。
|
||
assert db.next_queued_run() is None
|
||
db.resume_run("run_p", now)
|
||
assert db.get_run("run_p")["status"] == "QUEUED"
|
||
assert db.next_queued_run()["id"] == "run_p"
|
||
|
||
|
||
def test_recover_interrupted_runs(tmp_path) -> None:
|
||
"""重启恢复：遗留 RUNNING 任务恢复为 QUEUED（保留产物供断点续跑）。
|
||
|
||
进程被杀/重启时 RUNNING 任务不会自动收尾，若保持 RUNNING 将永久孤儿
|
||
（next_queued_run 不拾起、暂停后又被 execute_run 复活）。恢复为 QUEUED
|
||
后调度器会从产物表断点续跑；用户主动暂停的 PAUSED 任务保持不变。
|
||
"""
|
||
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"
|
||
for run_id, status in (("run_orphan", "RUNNING"), ("run_paused", "PAUSED"),
|
||
("run_done", "COMPLETED")):
|
||
db.create_run(
|
||
{
|
||
"id": run_id,
|
||
"workflow_id": "demo",
|
||
"workflow_version": 1,
|
||
"status": status,
|
||
"progress": 0.5,
|
||
"created_at": now,
|
||
"updated_at": now,
|
||
}
|
||
)
|
||
recovered = db.recover_interrupted_runs("2026-01-02T00:00:00+00:00")
|
||
assert recovered == 1 # 只有 RUNNING 被恢复。
|
||
assert db.get_run("run_orphan")["status"] == "QUEUED"
|
||
assert db.get_run("run_orphan")["updated_at"] == "2026-01-02T00:00:00+00:00"
|
||
assert db.get_run("run_paused")["status"] == "PAUSED"
|
||
assert db.get_run("run_done")["status"] == "COMPLETED"
|
||
# 恢复后调度器可拾起并断点续跑。
|
||
assert db.next_queued_run()["id"] == "run_orphan"
|
||
|
||
def test_recover_interrupted_batch_jobs(tmp_path) -> None:
|
||
"""重启恢复：遗留 RUNNING 的批量任务恢复为 QUEUED，供引擎重新拾起。
|
||
|
||
批量引擎处理视频（尤其大文件）时进程被杀/重启，批量任务停在 RUNNING。
|
||
若保持 RUNNING，next_queued_batch_job 只拾取 QUEUED，任务永远不会被再次
|
||
驱动 → 未处理完的 PENDING 视频永久残留（batch_969fabe74b83 事故链路之一）。
|
||
恢复为 QUEUED 后引擎重新拾起续跑；PAUSED 批量任务保持不变等待显式 resume。
|
||
"""
|
||
db = Database(tmp_path / "wov.db")
|
||
now = "2026-01-01T00:00:00+00:00"
|
||
|
||
def add_job(job_id: str, status: str) -> None:
|
||
"""创建指定状态的批量任务记录。"""
|
||
db.create_batch_job(
|
||
{
|
||
"id": job_id, "folder_path": "/tmp/videos", "workflow_id": "demo",
|
||
"recursive": 1, "status": status, "progress": 0.5, "total": 3,
|
||
"done": 1, "failed": 0, "error": None,
|
||
"created_at": now, "updated_at": now,
|
||
}
|
||
)
|
||
|
||
add_job("job_orphan", "RUNNING")
|
||
add_job("job_paused", "PAUSED")
|
||
add_job("job_done", "COMPLETED")
|
||
recovered = db.recover_interrupted_batch_jobs("2026-01-02T00:00:00+00:00")
|
||
assert recovered == 1 # 只有 RUNNING 被恢复。
|
||
assert db.get_batch_job("job_orphan")["status"] == "QUEUED"
|
||
assert db.get_batch_job("job_paused")["status"] == "PAUSED"
|
||
assert db.get_batch_job("job_done")["status"] == "COMPLETED"
|
||
# 恢复后批量引擎可拾起并从断点续跑。
|
||
assert db.next_queued_batch_job()["id"] == "job_orphan"
|
||
|
||
def test_restore_run_outputs(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"
|
||
db.create_run(
|
||
{
|
||
"id": "run_r",
|
||
"workflow_id": "demo",
|
||
"workflow_version": 1,
|
||
"status": "PAUSED",
|
||
"progress": 0,
|
||
"created_at": now,
|
||
"updated_at": now,
|
||
}
|
||
)
|
||
db.create_artifact(
|
||
{
|
||
"run_id": "run_r",
|
||
"node_id": "extract",
|
||
"name": "frames_manifest",
|
||
"uri": "frames.json",
|
||
"mime_type": "application/json",
|
||
"size": 1,
|
||
}
|
||
)
|
||
assert db.restore_run_outputs("run_r") == {
|
||
"extract": {"frames_manifest": "frames.json"}
|
||
}
|
||
assert db.restore_run_outputs("run_none") == {}
|
||
|
||
|
||
def test_run_source_column_default_and_next_queued(tmp_path) -> None:
|
||
"""source 列默认 upload；主调度器不拾取 batch 来源的运行。"""
|
||
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"
|
||
db.create_run(
|
||
{
|
||
"id": "run_upload",
|
||
"workflow_id": "demo",
|
||
"workflow_version": 1,
|
||
"status": "QUEUED",
|
||
"progress": 0,
|
||
"created_at": now,
|
||
"updated_at": now,
|
||
}
|
||
)
|
||
db.create_run(
|
||
{
|
||
"id": "run_batch",
|
||
"workflow_id": "demo",
|
||
"workflow_version": 1,
|
||
"status": "QUEUED",
|
||
"progress": 0,
|
||
"source": "batch",
|
||
"created_at": now,
|
||
"updated_at": now,
|
||
}
|
||
)
|
||
# 默认 source 为 upload，可显式指定 batch。
|
||
assert db.get_run("run_upload")["source"] == "upload"
|
||
assert db.get_run("run_batch")["source"] == "batch"
|
||
# 主调度器只取非 batch 运行，批量运行由批量引擎单独拾起。
|
||
assert db.next_queued_run()["id"] == "run_upload"
|
||
|
||
|
||
def test_db_migration_adds_source_column(tmp_path) -> None:
|
||
"""旧库迁移：缺少 source 列的库打开后自动补列并默认 upload。"""
|
||
import sqlite3
|
||
|
||
db_path = tmp_path / "old.db"
|
||
conn = sqlite3.connect(db_path)
|
||
conn.execute(
|
||
"CREATE TABLE workflow_runs (id TEXT PRIMARY KEY, workflow_id TEXT NOT NULL,"
|
||
" workflow_version INTEGER NOT NULL, status TEXT NOT NULL, current_node_id TEXT,"
|
||
" progress REAL NOT NULL DEFAULT 0, error TEXT, input_uri TEXT,"
|
||
" param_overrides TEXT, created_at TEXT NOT NULL, updated_at TEXT NOT NULL)"
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
Database(db_path)
|
||
conn = sqlite3.connect(db_path)
|
||
columns = [row[1] for row in conn.execute("PRAGMA table_info(workflow_runs)")]
|
||
conn.close()
|
||
assert "source" in columns
|
||
|
||
|
||
def test_batch_jobs_and_videos_crud(tmp_path) -> None:
|
||
"""批量任务/视频明细的增删改查与排队查询。"""
|
||
db = Database(tmp_path / "wov.db")
|
||
now = "2026-01-01T00:00:00+00:00"
|
||
db.create_batch_job(
|
||
{
|
||
"id": "batch_1", "folder_path": "/videos", "workflow_id": "demo",
|
||
"recursive": 1, "status": "QUEUED", "progress": 0, "total": 2,
|
||
"done": 0, "failed": 0, "current_video": None, "error": None,
|
||
"created_at": now, "updated_at": now,
|
||
}
|
||
)
|
||
assert db.next_queued_batch_job()["id"] == "batch_1"
|
||
assert db.list_batch_job_ids() == ["batch_1"]
|
||
assert db.list_batch_jobs()[0]["total"] == 2
|
||
|
||
db.create_batch_video(
|
||
{
|
||
"id": "bv_1", "job_id": "batch_1", "video_path": "/videos/a.mp4",
|
||
"work_dir": "/videos/a", "run_id": None, "status": "PENDING",
|
||
"error": None, "created_at": now, "updated_at": now,
|
||
}
|
||
)
|
||
db.update_batch_video("bv_1", status="COMPLETED", updated_at=now)
|
||
assert db.get_batch_video("bv_1")["status"] == "COMPLETED"
|
||
assert len(db.list_batch_videos("batch_1")) == 1
|
||
|
||
# 未知字段更新被忽略（不会报错也不会改状态）。
|
||
db.update_batch_video("bv_1", bogus=1, updated_at=now)
|
||
db.update_batch_job("batch_1", bogus=1, updated_at=now)
|
||
assert db.get_batch_video("bv_1")["status"] == "COMPLETED"
|
||
|
||
db.update_batch_job("batch_1", status="COMPLETED", done=1, failed=0, progress=1.0, updated_at=now)
|
||
job = db.get_batch_job("batch_1")
|
||
assert job["status"] == "COMPLETED" and job["done"] == 1
|
||
# 完成后不再排队。
|
||
assert db.next_queued_batch_job() is None
|
||
|
||
db.delete_batch_job("batch_1")
|
||
assert db.get_batch_job("batch_1") is None
|
||
assert db.get_batch_video("bv_1") is None
|