109 lines
4.0 KiB
Python
109 lines
4.0 KiB
Python
"""tools/build_trees.py 单元测试。
|
|
|
|
覆盖纯函数（完整性校验、待建清单发现、SRT 查找）与编排集成（Task 3 追加）。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
|
|
if str(PROJECT_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(PROJECT_ROOT))
|
|
|
|
from app.tree.index import IndexMeta, L1Card, L1Node, TreeIndex # noqa: E402
|
|
from tools.build_trees import ( # noqa: E402
|
|
_discover_pending,
|
|
_find_srt_entries,
|
|
_tree_is_complete,
|
|
)
|
|
|
|
|
|
def _write_valid_tree(tree_path: Path) -> None:
|
|
"""写入一棵最小合法树。"""
|
|
l1 = L1Node(
|
|
id="vid_L1_000",
|
|
card=L1Card("场景", "室内", ["实体"], ["动作"], ["关键词"], [], "线性"),
|
|
time_range=(0.0, 10.0),
|
|
children=[],
|
|
)
|
|
index = TreeIndex(metadata=IndexMeta("/v.mp4", "video"), roots=[l1])
|
|
tree_path.parent.mkdir(parents=True, exist_ok=True)
|
|
index.save_json(str(tree_path))
|
|
|
|
|
|
class TestTreeIsComplete:
|
|
"""_tree_is_complete 测试。"""
|
|
|
|
def test_valid_tree(self, tmp_path: Path) -> None:
|
|
"""合法 tree.json 判定完整。"""
|
|
tree_path = tmp_path / "vid" / "tree.json"
|
|
_write_valid_tree(tree_path)
|
|
assert _tree_is_complete(tree_path) is True
|
|
|
|
def test_missing_file(self, tmp_path: Path) -> None:
|
|
"""文件不存在判定不完整。"""
|
|
assert _tree_is_complete(tmp_path / "nope" / "tree.json") is False
|
|
|
|
def test_corrupt_json(self, tmp_path: Path) -> None:
|
|
"""损坏 JSON 判定不完整（不抛异常）。"""
|
|
p = tmp_path / "vid" / "tree.json"
|
|
p.parent.mkdir(parents=True)
|
|
p.write_text("{broken", encoding="utf-8")
|
|
assert _tree_is_complete(p) is False
|
|
|
|
|
|
class TestDiscoverPending:
|
|
"""_discover_pending 测试。"""
|
|
|
|
def _touch_videos(self, videos_dir: Path, names: list[str]) -> None:
|
|
videos_dir.mkdir(parents=True, exist_ok=True)
|
|
for n in names:
|
|
(videos_dir / n).write_bytes(b"")
|
|
|
|
def test_all_pending_when_fresh(self, tmp_path: Path) -> None:
|
|
"""无进度无产物时全部待建，按名排序。"""
|
|
videos = tmp_path / "videos"
|
|
self._touch_videos(videos, ["b.mp4", "a.mkv", "c.txt"])
|
|
pending = _discover_pending(videos, tmp_path / "out", set())
|
|
assert [p.name for p in pending] == ["a.mkv", "b.mp4"] # 非视频扩展名被忽略
|
|
|
|
def test_skips_finished_and_complete(self, tmp_path: Path) -> None:
|
|
"""progress 已记录或 tree.json 完整的视频被跳过。"""
|
|
videos = tmp_path / "videos"
|
|
out = tmp_path / "out"
|
|
self._touch_videos(videos, ["a.mp4", "b.mp4", "c.mp4"])
|
|
_write_valid_tree(out / "b" / "tree.json") # b 已有完整树
|
|
pending = _discover_pending(videos, out, {"a"}) # a 在 progress 中
|
|
assert [p.name for p in pending] == ["c.mp4"]
|
|
|
|
def test_incomplete_tree_not_skipped(self, tmp_path: Path) -> None:
|
|
"""tree.json 损坏的视频仍待建（重建覆盖）。"""
|
|
videos = tmp_path / "videos"
|
|
out = tmp_path / "out"
|
|
self._touch_videos(videos, ["a.mp4"])
|
|
(out / "a").mkdir(parents=True)
|
|
(out / "a" / "tree.json").write_text("{broken", encoding="utf-8")
|
|
pending = _discover_pending(videos, out, set())
|
|
assert [p.name for p in pending] == ["a.mp4"]
|
|
|
|
|
|
class TestFindSrtEntries:
|
|
"""_find_srt_entries 测试。"""
|
|
|
|
def test_found(self, tmp_path: Path) -> None:
|
|
"""同名 .srt 存在时解析返回条目。"""
|
|
srt = tmp_path / "vid.srt"
|
|
srt.write_text(
|
|
"1\n00:00:01,000 --> 00:00:03,000\nhello world\n\n",
|
|
encoding="utf-8",
|
|
)
|
|
entries = _find_srt_entries(tmp_path / "vid.mp4", tmp_path)
|
|
assert entries is not None
|
|
assert len(entries) == 1
|
|
|
|
def test_missing_returns_none(self, tmp_path: Path) -> None:
|
|
"""无同名 .srt 返回 None。"""
|
|
assert _find_srt_entries(tmp_path / "vid.mp4", tmp_path) is None
|