kim/clean-broken-tgvideos
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add txt path import and prune command

Browse Source
This commit is contained in:
kim
2026-03-30 10:15:49 +08:00
parent 3b2d3a5df4
commit 79370ed542
30 changed files with 432 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_cli.py
+451
View File
@@ -0,0 +1,451 @@
from typer.testing import CliRunner
from tgvideo_check.cli import app
from tgvideo_check.ffmpeg_runner import FfmpegExecutionResult
from tgvideo_check.ffprobe_runner import ProbeExecutionResult
runner = CliRunner()
def test_help_shows_scan_command() -> None:
result = runner.invoke(app, ["--help"])
assert result.exit_code == 0
assert "scan" in result.stdout
def test_scan_command_processes_stage1_results_and_exports_reports(
tmp_path, monkeypatch
) -> None:
root = tmp_path / "videos"
root.mkdir()
broken = root / "broken.mp4"
passed = root / "passed.avi"
broken.write_bytes(b"broken")
passed.write_bytes(b"passed")
jsonl_path = tmp_path / "results.jsonl"
csv_path = tmp_path / "results.csv"
db_path = tmp_path / "scan_state.db"
def fake_run_ffprobe(path, **kwargs):
if path == broken:
return ProbeExecutionResult(
status="completed",
command=("ffprobe", str(path)),
exit_code=1,
stdout_text="",
stderr_text="moov atom not found",
elapsed_ms=1,
)
return ProbeExecutionResult(
status="completed",
command=("ffprobe", str(path)),
exit_code=0,
stdout_text='{"format":{"format_name":"avi","duration":"2.0"},"streams":[{"codec_type":"video"}]}',
stderr_text="",
elapsed_ms=1,
)
monkeypatch.setattr("tgvideo_check.cli.run_ffprobe", fake_run_ffprobe)
result = runner.invoke(
app,
[
"scan",
str(root),
"--db",
str(db_path),
"--export-jsonl",
str(jsonl_path),
"--export-csv",
str(csv_path),
],
)
assert result.exit_code == 0
assert "Scanned 2 file(s)" in result.stdout
assert "confirmed_broken=1" in result.stdout
assert "passed_stage1=1" in result.stdout
assert jsonl_path.exists()
assert csv_path.exists()
def test_scan_command_runs_stage2_for_pending_files(tmp_path, monkeypatch) -> None:
root = tmp_path / "videos"
root.mkdir()
risky = root / "risky.mp4"
risky.write_bytes(b"risky")
def fake_run_ffprobe(path, **kwargs):
return ProbeExecutionResult(
status="completed",
command=("ffprobe", str(path)),
exit_code=0,
stdout_text='{"format":{"format_name":"mov,mp4,m4a,3gp,3g2,mj2","duration":"20.0"},"streams":[{"codec_type":"video"}]}',
stderr_text="",
elapsed_ms=1,
)
def fake_run_ffmpeg(path, **kwargs):
return FfmpegExecutionResult(
status="completed",
command=("ffmpeg", str(path)),
exit_code=0,
stdout_text="frame=10",
stderr_text="",
elapsed_ms=2,
reason_code=None,
)
monkeypatch.setattr("tgvideo_check.cli.run_ffprobe", fake_run_ffprobe)
monkeypatch.setattr("tgvideo_check.cli.run_ffmpeg_verification", fake_run_ffmpeg)
result = runner.invoke(
app,
[
"scan",
str(root),
"--db",
str(tmp_path / "scan_state.db"),
],
)
assert result.exit_code == 0
assert "Scanned 1 file(s)" in result.stdout
assert "passed_stage2=1" in result.stdout
def test_scan_resume_preserves_reason_code_in_exported_report(tmp_path, monkeypatch) -> None:
root = tmp_path / "videos"
root.mkdir()
broken = root / "broken.mp4"
broken.write_bytes(b"broken")
db_path = tmp_path / "scan_state.db"
first_jsonl = tmp_path / "first.jsonl"
second_jsonl = tmp_path / "second.jsonl"
def fake_run_ffprobe(path, **kwargs):
return ProbeExecutionResult(
status="completed",
command=("ffprobe", str(path)),
exit_code=1,
stdout_text="",
stderr_text="moov atom not found",
elapsed_ms=1,
)
monkeypatch.setattr("tgvideo_check.cli.run_ffprobe", fake_run_ffprobe)
first_result = runner.invoke(
app,
[
"scan",
str(root),
"--db",
str(db_path),
"--export-jsonl",
str(first_jsonl),
],
)
assert first_result.exit_code == 0
assert '"reason_code":"ffprobe_exit_nonzero"' in first_jsonl.read_text(encoding="utf-8")
def fail_ffprobe(*args, **kwargs):
raise AssertionError("resume run should not call ffprobe")
monkeypatch.setattr("tgvideo_check.cli.run_ffprobe", fail_ffprobe)
second_result = runner.invoke(
app,
[
"scan",
str(root),
"--db",
str(db_path),
"--export-jsonl",
str(second_jsonl),
],
)
assert second_result.exit_code == 0
assert '"reason_code":"ffprobe_exit_nonzero"' in second_jsonl.read_text(encoding="utf-8")
def test_scan_no_resume_forces_probe_execution(tmp_path, monkeypatch) -> None:
root = tmp_path / "videos"
root.mkdir()
video = root / "video.avi"
video.write_bytes(b"video")
db_path = tmp_path / "scan_state.db"
def first_run_ffprobe(path, **kwargs):
return ProbeExecutionResult(
status="completed",
command=("ffprobe", str(path)),
exit_code=0,
stdout_text='{"format":{"format_name":"avi","duration":"2.0"},"streams":[{"codec_type":"video"}]}',
stderr_text="",
elapsed_ms=1,
)
monkeypatch.setattr("tgvideo_check.cli.run_ffprobe", first_run_ffprobe)
first_result = runner.invoke(
app,
[
"scan",
str(root),
"--db",
str(db_path),
],
)
assert first_result.exit_code == 0
probe_calls = 0
def second_run_ffprobe(path, **kwargs):
nonlocal probe_calls
probe_calls += 1
return ProbeExecutionResult(
status="completed",
command=("ffprobe", str(path)),
exit_code=0,
stdout_text='{"format":{"format_name":"avi","duration":"2.0"},"streams":[{"codec_type":"video"}]}',
stderr_text="",
elapsed_ms=1,
)
monkeypatch.setattr("tgvideo_check.cli.run_ffprobe", second_run_ffprobe)
second_result = runner.invoke(
app,
[
"scan",
str(root),
"--db",
str(db_path),
"--no-resume",
],
)
assert second_result.exit_code == 0
assert probe_calls == 1
def test_scan_can_load_directories_from_paths_file(tmp_path, monkeypatch) -> None:
root_a = tmp_path / "Dir With Space"
root_b = tmp_path / "[Special]!目录"
root_a.mkdir()
root_b.mkdir()
(root_a / "a.avi").write_bytes(b"a")
(root_b / "b.avi").write_bytes(b"b")
paths_file = tmp_path / "paths.txt"
paths_file.write_text(
f'{root_a}\n"{root_b}"\n',
encoding="utf-8",
)
probe_calls: list[str] = []
def fake_run_ffprobe(path, **kwargs):
probe_calls.append(path.name)
return ProbeExecutionResult(
status="completed",
command=("ffprobe", str(path)),
exit_code=0,
stdout_text='{"format":{"format_name":"avi","duration":"2.0"},"streams":[{"codec_type":"video"}]}',
stderr_text="",
elapsed_ms=1,
)
monkeypatch.setattr("tgvideo_check.cli.run_ffprobe", fake_run_ffprobe)
result = runner.invoke(
app,
[
"scan",
"--paths-file",
str(paths_file),
"--db",
str(tmp_path / "scan_state.db"),
],
)
assert result.exit_code == 0
assert "Scanned 2 file(s)" in result.stdout
assert sorted(probe_calls) == ["a.avi", "b.avi"]
def test_scan_merges_positional_paths_and_paths_file(tmp_path, monkeypatch) -> None:
direct_root = tmp_path / "direct"
listed_root = tmp_path / "listed"
direct_root.mkdir()
listed_root.mkdir()
(direct_root / "direct.avi").write_bytes(b"a")
(listed_root / "listed.avi").write_bytes(b"b")
paths_file = tmp_path / "paths.txt"
paths_file.write_text(f"{listed_root}\n", encoding="utf-8")
probe_calls: list[str] = []
def fake_run_ffprobe(path, **kwargs):
probe_calls.append(path.name)
return ProbeExecutionResult(
status="completed",
command=("ffprobe", str(path)),
exit_code=0,
stdout_text='{"format":{"format_name":"avi","duration":"2.0"},"streams":[{"codec_type":"video"}]}',
stderr_text="",
elapsed_ms=1,
)
monkeypatch.setattr("tgvideo_check.cli.run_ffprobe", fake_run_ffprobe)
result = runner.invoke(
app,
[
"scan",
str(direct_root),
"--paths-file",
str(paths_file),
"--db",
str(tmp_path / "scan_state.db"),
],
)
assert result.exit_code == 0
assert "Scanned 2 file(s)" in result.stdout
assert sorted(probe_calls) == ["direct.avi", "listed.avi"]
def test_prune_dry_run_reports_confirmed_broken_without_deleting(tmp_path, monkeypatch) -> None:
root = tmp_path / "videos"
root.mkdir()
broken = root / "broken.mp4"
passed = root / "passed.avi"
broken.write_bytes(b"broken")
passed.write_bytes(b"passed")
db_path = tmp_path / "scan_state.db"
def fake_run_ffprobe(path, **kwargs):
if path == broken:
return ProbeExecutionResult(
status="completed",
command=("ffprobe", str(path)),
exit_code=1,
stdout_text="",
stderr_text="moov atom not found",
elapsed_ms=1,
)
return ProbeExecutionResult(
status="completed",
command=("ffprobe", str(path)),
exit_code=0,
stdout_text='{"format":{"format_name":"avi","duration":"2.0"},"streams":[{"codec_type":"video"}]}',
stderr_text="",
elapsed_ms=1,
)
monkeypatch.setattr("tgvideo_check.cli.run_ffprobe", fake_run_ffprobe)
scan_result = runner.invoke(
app,
[
"scan",
str(root),
"--db",
str(db_path),
],
)
assert scan_result.exit_code == 0
prune_result = runner.invoke(
app,
[
"prune",
"--db",
str(db_path),
],
)
assert prune_result.exit_code == 0
assert "Dry-run" in prune_result.stdout
assert "matched=1" in prune_result.stdout
assert "deleted=0" in prune_result.stdout
assert broken.exists()
assert passed.exists()
def test_prune_no_dry_run_deletes_confirmed_broken_files(tmp_path, monkeypatch) -> None:
root = tmp_path / "videos"
root.mkdir()
broken = root / "broken.mp4"
suspicious = root / "suspicious.mp4"
broken.write_bytes(b"broken")
suspicious.write_bytes(b"suspicious")
db_path = tmp_path / "scan_state.db"
def fake_run_ffprobe(path, **kwargs):
if path == broken:
return ProbeExecutionResult(
status="completed",
command=("ffprobe", str(path)),
exit_code=1,
stdout_text="",
stderr_text="moov atom not found",
elapsed_ms=1,
)
return ProbeExecutionResult(
status="completed",
command=("ffprobe", str(path)),
exit_code=0,
stdout_text='{"format":{"format_name":"mov,mp4,m4a,3gp,3g2,mj2","duration":"20.0"},"streams":[{"codec_type":"video"}]}',
stderr_text="",
elapsed_ms=1,
)
def fake_run_ffmpeg(path, **kwargs):
return FfmpegExecutionResult(
status="completed",
command=("ffmpeg", str(path)),
exit_code=0,
stdout_text="frame=10",
stderr_text="",
elapsed_ms=2,
reason_code=None,
)
monkeypatch.setattr("tgvideo_check.cli.run_ffprobe", fake_run_ffprobe)
monkeypatch.setattr("tgvideo_check.cli.run_ffmpeg_verification", fake_run_ffmpeg)
scan_result = runner.invoke(
app,
[
"scan",
str(root),
"--db",
str(db_path),
],
)
assert scan_result.exit_code == 0
prune_result = runner.invoke(
app,
[
"prune",
"--db",
str(db_path),
"--no-dry-run",
],
)
assert prune_result.exit_code == 0
assert "deleted=1" in prune_result.stdout
assert not broken.exists()
assert suspicious.exists()
tests/test_discovery.py
+40
View File
@@ -0,0 +1,40 @@
from pathlib import Path
from tgvideo_check.discovery import discover_files
def test_discovery_filters_extensions_and_preserves_order(tmp_path: Path) -> None:
root_a = tmp_path / "A"
root_b = tmp_path / "B"
root_a.mkdir()
root_b.mkdir()
keep_1 = root_a / "001.mp4"
skip = root_a / "note.txt"
nested = root_a / "nested"
nested.mkdir()
keep_2 = nested / "002.MKV"
keep_3 = root_b / "003.webm"
for file_path in (keep_1, skip, keep_2, keep_3):
file_path.write_bytes(b"data")
discovered = discover_files((root_a, root_b), extensions=("mp4", "mkv", "webm"))
assert [item.path for item in discovered] == [keep_1, keep_2, keep_3]
assert [item.discovery_order for item in discovered] == [0, 1, 2]
def test_discovery_can_disable_recursion(tmp_path: Path) -> None:
root = tmp_path / "root"
root.mkdir()
direct = root / "001.mp4"
nested_dir = root / "nested"
nested_dir.mkdir()
nested = nested_dir / "002.mp4"
direct.write_bytes(b"a")
nested.write_bytes(b"b")
discovered = discover_files((root,), extensions=("mp4",), recursive=False)
assert [item.path for item in discovered] == [direct]
tests/test_integration_resume.py
+72
View File
@@ -0,0 +1,72 @@
import shutil
import subprocess
from pathlib import Path
import pytest
from typer.testing import CliRunner
from tgvideo_check.cli import app
runner = CliRunner()
HAS_FFMPEG = shutil.which("ffmpeg") is not None and shutil.which("ffprobe") is not None
@pytest.mark.skipif(not HAS_FFMPEG, reason="ffmpeg/ffprobe not installed")
def test_scan_reuses_previous_results_without_rerunning_probes(
tmp_path: Path, monkeypatch
) -> None:
root = tmp_path / "videos"
root.mkdir()
video_path = root / "video.mp4"
_generate_test_video(video_path, duration=2)
db_path = tmp_path / "scan_state.db"
first_result = runner.invoke(
app,
[
"scan",
str(root),
"--db",
str(db_path),
],
)
assert first_result.exit_code == 0
assert "passed_stage2=1" in first_result.stdout
def fail_ffprobe(*args, **kwargs):
raise AssertionError("run_ffprobe should not be called during resume")
def fail_ffmpeg(*args, **kwargs):
raise AssertionError("run_ffmpeg_verification should not be called during resume")
monkeypatch.setattr("tgvideo_check.cli.run_ffprobe", fail_ffprobe)
monkeypatch.setattr("tgvideo_check.cli.run_ffmpeg_verification", fail_ffmpeg)
second_result = runner.invoke(
app,
[
"scan",
str(root),
"--db",
str(db_path),
],
)
assert second_result.exit_code == 0
assert "passed_stage2=1" in second_result.stdout
def _generate_test_video(path: Path, *, duration: int) -> None:
command = [
"ffmpeg",
"-y",
"-f",
"lavfi",
"-i",
f"testsrc=size=64x64:duration={duration}:rate=25",
"-pix_fmt",
"yuv420p",
str(path),
]
subprocess.run(command, check=True, capture_output=True)
tests/test_integration_truncated_mkv.py
+68
View File
@@ -0,0 +1,68 @@
import shutil
import subprocess
from pathlib import Path
import pytest
from typer.testing import CliRunner
from tgvideo_check.cli import app
from tgvideo_check.store import ScanStore
runner = CliRunner()
HAS_FFMPEG = shutil.which("ffmpeg") is not None and shutil.which("ffprobe") is not None
@pytest.mark.skipif(not HAS_FFMPEG, reason="ffmpeg/ffprobe not installed")
def test_scan_detects_truncated_mkv_during_stage2(tmp_path: Path) -> None:
root = tmp_path / "videos"
root.mkdir()
valid_path = root / "valid.mkv"
broken_path = root / "broken.mkv"
_generate_test_video(valid_path, duration=8)
_truncate_copy(
valid_path,
broken_path,
trim_bytes=max(1024, valid_path.stat().st_size * 4 // 5),
)
db_path = tmp_path / "scan_state.db"
result = runner.invoke(
app,
[
"scan",
str(root),
"--db",
str(db_path),
],
)
assert result.exit_code == 0
store = ScanStore(db_path)
rows = store.list_files(1)
broken_row = next(row for row in rows if row.path.name == "broken.mkv")
assert broken_row.stage1_status == "pending_stage2"
assert broken_row.stage2_status == "confirmed_broken"
assert broken_row.final_status == "confirmed_broken"
def _generate_test_video(path: Path, *, duration: int) -> None:
command = [
"ffmpeg",
"-y",
"-f",
"lavfi",
"-i",
f"testsrc=size=64x64:duration={duration}:rate=25",
"-pix_fmt",
"yuv420p",
str(path),
]
subprocess.run(command, check=True, capture_output=True)
def _truncate_copy(source: Path, destination: Path, *, trim_bytes: int) -> None:
data = source.read_bytes()
destination.write_bytes(data[: max(1, len(data) - trim_bytes)])
tests/test_integration_truncated_mp4.py
+67
View File
@@ -0,0 +1,67 @@
import shutil
import subprocess
from pathlib import Path
import orjson
import pytest
from tgvideo_check.cli import app
from typer.testing import CliRunner
runner = CliRunner()
HAS_FFMPEG = shutil.which("ffmpeg") is not None and shutil.which("ffprobe") is not None
@pytest.mark.skipif(not HAS_FFMPEG, reason="ffmpeg/ffprobe not installed")
def test_scan_detects_truncated_mp4_during_stage1(tmp_path: Path) -> None:
root = tmp_path / "videos"
root.mkdir()
valid_path = root / "valid.mp4"
broken_path = root / "broken.mp4"
_generate_test_video(valid_path, container="mp4", duration=2)
_truncate_copy(valid_path, broken_path, trim_bytes=512)
jsonl_path = tmp_path / "results.jsonl"
result = runner.invoke(
app,
[
"scan",
str(root),
"--db",
str(tmp_path / "scan_state.db"),
"--export-jsonl",
str(jsonl_path),
],
)
assert result.exit_code == 0
rows = _read_jsonl(jsonl_path)
broken_row = next(row for row in rows if row["path"].endswith("broken.mp4"))
assert broken_row["final_status"] == "confirmed_broken"
assert broken_row["reason_code"] in {"ffprobe_exit_nonzero", "hard_error_keyword"}
def _generate_test_video(path: Path, *, container: str, duration: int) -> None:
command = [
"ffmpeg",
"-y",
"-f",
"lavfi",
"-i",
f"testsrc=size=64x64:duration={duration}:rate=25",
"-pix_fmt",
"yuv420p",
str(path),
]
subprocess.run(command, check=True, capture_output=True)
def _truncate_copy(source: Path, destination: Path, *, trim_bytes: int) -> None:
data = source.read_bytes()
destination.write_bytes(data[: max(1, len(data) - trim_bytes)])
def _read_jsonl(path: Path) -> list[dict[str, object]]:
return [orjson.loads(line) for line in path.read_text(encoding="utf-8").splitlines()]
tests/test_perf_smoke.py
+67
View File
@@ -0,0 +1,67 @@
import os
import shutil
import subprocess
import time
from pathlib import Path
import pytest
from tgvideo_check.cli import app
from typer.testing import CliRunner
runner = CliRunner()
HAS_FFMPEG = shutil.which("ffmpeg") is not None and shutil.which("ffprobe") is not None
RUN_PERF_SMOKE = os.environ.get("TGVIDEO_RUN_PERF_SMOKE") == "1"
@pytest.mark.skipif(
not HAS_FFMPEG or not RUN_PERF_SMOKE,
reason="set TGVIDEO_RUN_PERF_SMOKE=1 and install ffmpeg/ffprobe to run",
)
def test_perf_smoke_collects_baseline_for_stage1_worker_counts(tmp_path: Path) -> None:
root = tmp_path / "videos"
root.mkdir()
for index in range(6):
_generate_test_video(root / f"video-{index}.mp4", duration=2)
single_worker_seconds = _run_scan(root, tmp_path / "scan-1.db", stage1_workers=1)
multi_worker_seconds = _run_scan(root, tmp_path / "scan-2.db", stage1_workers=2)
assert single_worker_seconds > 0
assert multi_worker_seconds > 0
def _run_scan(root: Path, db_path: Path, *, stage1_workers: int) -> float:
start = time.perf_counter()
result = runner.invoke(
app,
[
"scan",
str(root),
"--db",
str(db_path),
"--stage1-workers",
str(stage1_workers),
"--stage2-workers",
"1",
],
)
elapsed = time.perf_counter() - start
assert result.exit_code == 0
return elapsed
def _generate_test_video(path: Path, *, duration: int) -> None:
command = [
"ffmpeg",
"-y",
"-f",
"lavfi",
"-i",
f"testsrc=size=64x64:duration={duration}:rate=25",
"-pix_fmt",
"yuv420p",
str(path),
]
subprocess.run(command, check=True, capture_output=True)
tests/test_profiles.py
+29
View File
@@ -0,0 +1,29 @@
from tgvideo_check.config import ScanConfig
from tgvideo_check.profiles import get_profile, resolve_config
def test_tg_hdd_profile_uses_conservative_defaults() -> None:
profile = get_profile("tg-hdd")
assert profile.name == "tg-hdd"
assert profile.stage1_workers == 1
assert profile.stage2_workers == 1
assert profile.sample_duration == 3
assert profile.high_risk_extensions == ("mp4", "mov", "m4v", "mkv", "webm")
def test_resolve_config_applies_profile_overrides() -> None:
config = resolve_config(
ScanConfig(
roots=(),
profile="tg-hdd",
stage1_workers=3,
stage2_workers=None,
extensions=("MP4", ".MKV"),
)
)
assert config.stage1_workers == 3
assert config.stage2_workers == 1
assert config.extensions == ("mp4", "mkv")
tests/test_reporting.py
+63
View File
@@ -0,0 +1,63 @@
from pathlib import Path
from tgvideo_check.reporting import (
ReportRow,
collect_redownload_candidates,
export_csv_summary,
export_jsonl,
summarize_reason_counts,
)
def sample_rows() -> list[ReportRow]:
return [
ReportRow(
path=Path("a.mp4"),
final_status="confirmed_broken",
reason_code="ffprobe_exit_nonzero",
size=100,
),
ReportRow(
path=Path("b.mkv"),
final_status="suspicious",
reason_code="timeout",
size=200,
),
ReportRow(
path=Path("c.avi"),
final_status="passed_stage1",
reason_code="passed_stage1",
size=300,
),
]
def test_summarize_reason_counts_groups_rows_by_reason() -> None:
summary = summarize_reason_counts(sample_rows())
assert summary == {
"ffprobe_exit_nonzero": 1,
"timeout": 1,
"passed_stage1": 1,
}
def test_collect_redownload_candidates_returns_broken_and_suspicious() -> None:
candidates = collect_redownload_candidates(sample_rows())
assert candidates == [Path("a.mp4"), Path("b.mkv")]
def test_export_jsonl_and_csv_write_expected_content(tmp_path: Path) -> None:
jsonl_path = tmp_path / "results.jsonl"
csv_path = tmp_path / "results.csv"
export_jsonl(sample_rows(), jsonl_path)
export_csv_summary(sample_rows(), csv_path)
jsonl_text = jsonl_path.read_text(encoding="utf-8")
csv_text = csv_path.read_text(encoding="utf-8")
assert '"path":"a.mp4"' in jsonl_text
assert "final_status,reason_code,path,size" in csv_text
assert "confirmed_broken,ffprobe_exit_nonzero,a.mp4,100" in csv_text
tests/test_resume_logic.py
+100
View File
@@ -0,0 +1,100 @@
import sqlite3
from pathlib import Path
from tgvideo_check.models import FileCandidate
from tgvideo_check.store import ScanStore
def make_candidate(path: Path, *, order: int, size: int = 10, mtime_ns: int = 100) -> FileCandidate:
path.write_bytes(b"x" * size)
return FileCandidate(
path=path,
size=size,
mtime_ns=mtime_ns,
extension=path.suffix.lower().lstrip("."),
discovery_order=order,
)
def test_initialize_creates_expected_tables(tmp_path: Path) -> None:
store = ScanStore(tmp_path / "scan_state.db")
store.initialize()
with sqlite3.connect(store.db_path) as connection:
table_names = {
row[0]
for row in connection.execute(
"SELECT name FROM sqlite_master WHERE type = 'table'"
)
}
assert {"scan_runs", "files", "stage1_results", "stage2_results"} <= table_names
def test_upsert_files_persists_candidates_in_discovery_order(tmp_path: Path) -> None:
store = ScanStore(tmp_path / "scan_state.db")
store.initialize()
run_id = store.create_run(profile="tg-hdd", root_paths=(tmp_path,))
first = make_candidate(tmp_path / "001.mp4", order=0)
second = make_candidate(tmp_path / "002.mkv", order=1)
store.upsert_files(run_id, [first, second])
rows = store.list_files(run_id)
assert [row.path for row in rows] == [first.path, second.path]
assert [row.discovery_order for row in rows] == [0, 1]
assert all(row.stage1_status == "pending_stage1" for row in rows)
def test_upsert_files_reuses_previous_status_when_identity_matches(tmp_path: Path) -> None:
store = ScanStore(tmp_path / "scan_state.db")
store.initialize()
first_run = store.create_run(profile="tg-hdd", root_paths=(tmp_path,))
file_path = tmp_path / "clip.mp4"
candidate = make_candidate(file_path, order=0, size=12, mtime_ns=999)
[stored] = store.upsert_files(first_run, [candidate])
store.record_stage1_status(
stored.file_id,
stage1_status="passed_stage1",
final_status="passed_stage1",
)
second_run = store.create_run(profile="tg-hdd", root_paths=(tmp_path,))
[resumed] = store.upsert_files(second_run, [candidate])
assert resumed.stage1_status == "passed_stage1"
assert resumed.final_status == "passed_stage1"
assert resumed.resumed_from_run_id == first_run
def test_upsert_files_does_not_reuse_changed_file(tmp_path: Path) -> None:
store = ScanStore(tmp_path / "scan_state.db")
store.initialize()
first_run = store.create_run(profile="tg-hdd", root_paths=(tmp_path,))
file_path = tmp_path / "clip.mp4"
original = make_candidate(file_path, order=0, size=12, mtime_ns=999)
[stored] = store.upsert_files(first_run, [original])
store.record_stage1_status(
stored.file_id,
stage1_status="confirmed_broken",
final_status="confirmed_broken",
)
second_run = store.create_run(profile="tg-hdd", root_paths=(tmp_path,))
changed = FileCandidate(
path=file_path,
size=13,
mtime_ns=1000,
extension="mp4",
discovery_order=0,
)
[fresh] = store.upsert_files(second_run, [changed])
assert fresh.stage1_status == "pending_stage1"
assert fresh.final_status is None
assert fresh.resumed_from_run_id is None
tests/test_scheduler_hdd.py
+146
View File
@@ -0,0 +1,146 @@
from pathlib import Path
import threading
import time
from tgvideo_check.models import StoredFile
from tgvideo_check.scheduler import (
build_stage_plan,
execute_stage_batch,
select_pending_stage1_files,
select_pending_stage2_files,
)
def make_stored_file(
*,
file_id: int,
discovery_order: int,
stage1_status: str,
stage2_status: str | None = None,
) -> StoredFile:
return StoredFile(
file_id=file_id,
run_id=1,
path=Path(f"video-{file_id}.mp4"),
size=100,
mtime_ns=1000,
extension="mp4",
discovery_order=discovery_order,
stage1_status=stage1_status,
stage2_status=stage2_status,
final_status=None,
resumed_from_run_id=None,
)
def test_select_pending_stage1_files_only_keeps_stage1_work() -> None:
files = [
make_stored_file(file_id=1, discovery_order=2, stage1_status="pending_stage1"),
make_stored_file(file_id=2, discovery_order=1, stage1_status="passed_stage1"),
make_stored_file(file_id=3, discovery_order=0, stage1_status="pending_stage1"),
]
selected = select_pending_stage1_files(files)
assert [item.file_id for item in selected] == [3, 1]
def test_select_pending_stage2_files_skips_completed_or_irrelevant_items() -> None:
files = [
make_stored_file(
file_id=1,
discovery_order=2,
stage1_status="pending_stage2",
stage2_status=None,
),
make_stored_file(
file_id=2,
discovery_order=1,
stage1_status="pending_stage2",
stage2_status="passed_stage2",
),
make_stored_file(
file_id=3,
discovery_order=0,
stage1_status="passed_stage1",
stage2_status=None,
),
]
selected = select_pending_stage2_files(files)
assert [item.file_id for item in selected] == [1]
def test_build_stage_plan_keeps_stage1_and_stage2_separate() -> None:
stage1_files = [
make_stored_file(file_id=1, discovery_order=0, stage1_status="pending_stage1"),
make_stored_file(file_id=2, discovery_order=1, stage1_status="pending_stage1"),
make_stored_file(file_id=3, discovery_order=2, stage1_status="pending_stage1"),
]
stage2_files = [
make_stored_file(file_id=4, discovery_order=3, stage1_status="pending_stage2"),
make_stored_file(file_id=5, discovery_order=4, stage1_status="pending_stage2"),
]
plan = build_stage_plan(
stage1_files=stage1_files,
stage2_files=stage2_files,
stage1_workers=1,
stage2_workers=1,
queue_limit=2,
)
assert [batch.stage_name for batch in plan] == [
"stage1",
"stage1",
"stage2",
]
assert [len(batch.files) for batch in plan] == [2, 1, 2]
assert [[item.file_id for item in batch.files] for batch in plan] == [
[1, 2],
[3],
[4, 5],
]
def test_execute_stage_batch_preserves_input_order() -> None:
batch = build_stage_plan(
stage1_files=[
make_stored_file(file_id=1, discovery_order=0, stage1_status="pending_stage1"),
make_stored_file(file_id=2, discovery_order=1, stage1_status="pending_stage1"),
],
stage2_files=[],
stage1_workers=2,
stage2_workers=1,
queue_limit=64,
)[0]
results = execute_stage_batch(batch, lambda item: item.file_id * 10)
assert results == (10, 20)
def test_execute_stage_batch_uses_multiple_threads_when_workers_allow() -> None:
batch = build_stage_plan(
stage1_files=[
make_stored_file(file_id=1, discovery_order=0, stage1_status="pending_stage1"),
make_stored_file(file_id=2, discovery_order=1, stage1_status="pending_stage1"),
make_stored_file(file_id=3, discovery_order=2, stage1_status="pending_stage1"),
],
stage2_files=[],
stage1_workers=3,
stage2_workers=1,
queue_limit=64,
)[0]
thread_ids: set[int] = set()
def worker(item: StoredFile) -> int:
thread_ids.add(threading.get_ident())
time.sleep(0.02)
return item.file_id
execute_stage_batch(batch, worker)
assert len(thread_ids) >= 2
tests/test_stage1_classification.py
+78
View File
@@ -0,0 +1,78 @@
from tgvideo_check.classifiers import classify_stage1_result
def test_stage1_marks_nonzero_exit_as_confirmed_broken() -> None:
result = classify_stage1_result(
exit_code=1,
stderr_text="moov atom not found",
probe_payload=None,
extension="mp4",
high_risk_extensions=("mp4", "mkv"),
)
assert result.status == "confirmed_broken"
assert result.reason_code == "ffprobe_exit_nonzero"
def test_stage1_marks_missing_video_stream_as_confirmed_broken() -> None:
result = classify_stage1_result(
exit_code=0,
stderr_text="",
probe_payload={
"format": {"format_name": "mov,mp4,m4a,3gp,3g2,mj2", "duration": "12.3"},
"streams": [{"codec_type": "audio"}],
},
extension="mp4",
high_risk_extensions=(),
)
assert result.status == "confirmed_broken"
assert result.reason_code == "missing_video_stream"
def test_stage1_marks_invalid_duration_as_pending_stage2() -> None:
result = classify_stage1_result(
exit_code=0,
stderr_text="",
probe_payload={
"format": {"format_name": "matroska,webm", "duration": "N/A"},
"streams": [{"codec_type": "video"}],
},
extension="webm",
high_risk_extensions=(),
)
assert result.status == "pending_stage2"
assert result.reason_code == "invalid_duration"
def test_stage1_marks_profile_high_risk_extension_as_pending_stage2() -> None:
result = classify_stage1_result(
exit_code=0,
stderr_text="",
probe_payload={
"format": {"format_name": "mov,mp4,m4a,3gp,3g2,mj2", "duration": "21.0"},
"streams": [{"codec_type": "video"}],
},
extension="mp4",
high_risk_extensions=("mp4", "mkv"),
)
assert result.status == "pending_stage2"
assert result.reason_code == "high_risk_extension"
def test_stage1_marks_clean_non_risky_file_as_passed() -> None:
result = classify_stage1_result(
exit_code=0,
stderr_text="",
probe_payload={
"format": {"format_name": "avi", "duration": "21.0"},
"streams": [{"codec_type": "video"}],
},
extension="avi",
high_risk_extensions=("mp4", "mkv"),
)
assert result.status == "passed_stage1"
assert result.reason_code == "passed_stage1"
tests/test_stage1_runner.py
+65
View File
@@ -0,0 +1,65 @@
import sys
from pathlib import Path
from tgvideo_check.ffprobe_runner import build_ffprobe_command, run_ffprobe
def test_build_ffprobe_command_requests_minimal_fields(tmp_path: Path) -> None:
video_path = tmp_path / "clip.mp4"
command = build_ffprobe_command(video_path)
assert command[:3] == ["ffprobe", "-v", "error"]
assert "format=format_name,duration,size:stream=index,codec_type,codec_name,width,height" in command
assert command[-1] == str(video_path)
def test_run_ffprobe_captures_stdout_stderr_and_exit_code(tmp_path: Path) -> None:
script = tmp_path / "fake_ffprobe.py"
script.write_text(
"import sys\n"
"print('{\"format\": {\"duration\": \"1.0\"}}')\n"
"print('soft warning', file=sys.stderr)\n",
encoding="utf-8",
)
result = run_ffprobe(
tmp_path / "clip.mp4",
executable=sys.executable,
launcher_args=(str(script),),
)
assert result.status == "completed"
assert result.exit_code == 0
assert "\"duration\": \"1.0\"" in result.stdout_text
assert "soft warning" in result.stderr_text
assert result.elapsed_ms >= 0
def test_run_ffprobe_reports_missing_command(tmp_path: Path) -> None:
result = run_ffprobe(
tmp_path / "clip.mp4",
executable="ffprobe-command-that-does-not-exist",
)
assert result.status == "missing_command"
assert result.exit_code is None
def test_run_ffprobe_reports_timeout(tmp_path: Path) -> None:
script = tmp_path / "slow_ffprobe.py"
script.write_text(
"import time\n"
"time.sleep(0.3)\n",
encoding="utf-8",
)
result = run_ffprobe(
tmp_path / "clip.mp4",
executable=sys.executable,
launcher_args=(str(script),),
timeout_seconds=0.05,
)
assert result.status == "timeout"
assert result.exit_code is None
tests/test_stage2_classification.py
+95
View File
@@ -0,0 +1,95 @@
import sys
from pathlib import Path
from tgvideo_check.ffmpeg_runner import (
build_ffmpeg_command,
classify_ffmpeg_error,
generate_sample_offsets,
run_ffmpeg_verification,
)
def test_generate_sample_offsets_for_short_clip() -> None:
assert generate_sample_offsets(30.0) == (3.0, 25.5)
def test_generate_sample_offsets_for_medium_clip() -> None:
assert generate_sample_offsets(300.0) == (30.0, 150.0, 270.0)
def test_build_ffmpeg_command_for_sample_window(tmp_path: Path) -> None:
video_path = tmp_path / "clip.mp4"
command = build_ffmpeg_command(
video_path,
offset_seconds=30.0,
sample_duration=3,
)
assert command[:4] == ["ffmpeg", "-v", "error", "-xerror"]
assert "-ss" in command
assert "-t" in command
assert "-map" in command
assert command[-1] == "-"
def test_build_ffmpeg_command_for_full_decode_mode(tmp_path: Path) -> None:
video_path = tmp_path / "clip.mp4"
command = build_ffmpeg_command(
video_path,
full_decode=True,
)
assert "-ss" not in command
assert "-t" not in command
assert command[-1] == "-"
def test_classify_ffmpeg_error_detects_truncation() -> None:
assert (
classify_ffmpeg_error("Truncating packet of size 4096 to 1024")
== "truncating_packet"
)
def test_run_ffmpeg_verification_captures_process_result(tmp_path: Path) -> None:
script = tmp_path / "fake_ffmpeg.py"
script.write_text(
"import sys\n"
"print('frame=12 fps=0.0')\n"
"print('decode warning', file=sys.stderr)\n",
encoding="utf-8",
)
result = run_ffmpeg_verification(
tmp_path / "clip.mp4",
offset_seconds=3.0,
executable=sys.executable,
launcher_args=(str(script),),
)
assert result.status == "completed"
assert result.exit_code == 0
assert "frame=12" in result.stdout_text
assert "decode warning" in result.stderr_text
def test_run_ffmpeg_verification_reports_timeout(tmp_path: Path) -> None:
script = tmp_path / "slow_ffmpeg.py"
script.write_text(
"import time\n"
"time.sleep(0.3)\n",
encoding="utf-8",
)
result = run_ffmpeg_verification(
tmp_path / "clip.mp4",
offset_seconds=3.0,
executable=sys.executable,
launcher_args=(str(script),),
timeout_seconds=0.05,
)
assert result.status == "timeout"
assert result.exit_code is None