160 lines
6.0 KiB
Python
160 lines
6.0 KiB
Python
"""Tests for video integrity checking (src.core.integrity).
|
|
|
|
The FFmpeg subprocess call is monkeypatched so the tests don't need ffmpeg
|
|
installed and don't touch real video data.
|
|
"""
|
|
import subprocess
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from src.core import integrity
|
|
from src.core.integrity import IntegrityResult, check_video_integrity
|
|
|
|
|
|
@pytest.fixture
|
|
def video_file(tmp_path) -> Path:
|
|
f = tmp_path / "movie.mkv"
|
|
f.write_bytes(b"not really a video, but a real file")
|
|
return f
|
|
|
|
|
|
def _fake_run(returncode=0, stdout="", stderr=""):
|
|
def run(cmd, **kwargs):
|
|
return SimpleNamespace(returncode=returncode, stdout=stdout, stderr=stderr)
|
|
return run
|
|
|
|
|
|
def test_deep_clean_file_is_ok(monkeypatch, video_file):
|
|
monkeypatch.setattr(integrity.shutil, "which", lambda _: "/usr/bin/ffmpeg")
|
|
monkeypatch.setattr(integrity.subprocess, "run", _fake_run(0, "", ""))
|
|
result = check_video_integrity(video_file, deep=True)
|
|
assert result.ok and result.checked and result.error == ""
|
|
|
|
|
|
def test_deep_reports_decode_errors(monkeypatch, video_file):
|
|
monkeypatch.setattr(integrity.shutil, "which", lambda _: "/usr/bin/ffmpeg")
|
|
monkeypatch.setattr(
|
|
integrity.subprocess, "run",
|
|
_fake_run(1, "", "[matroska @ 0x..] Read error\nmore noise"))
|
|
result = check_video_integrity(video_file, deep=True)
|
|
assert not result.ok and result.checked
|
|
assert result.error == "[matroska @ 0x..] Read error" # first line only
|
|
|
|
|
|
def test_deep_stderr_without_nonzero_rc_still_fails(monkeypatch, video_file):
|
|
"""ffmpeg can print decode errors while returning rc 0 — must still fail."""
|
|
monkeypatch.setattr(integrity.shutil, "which", lambda _: "/usr/bin/ffmpeg")
|
|
monkeypatch.setattr(
|
|
integrity.subprocess, "run", _fake_run(0, "", "corrupt packet"))
|
|
result = check_video_integrity(video_file, deep=True)
|
|
assert not result.ok
|
|
|
|
|
|
def test_quick_ok_when_video_stream_present(monkeypatch, video_file):
|
|
monkeypatch.setattr(integrity.shutil, "which", lambda _: "/usr/bin/ffprobe")
|
|
monkeypatch.setattr(integrity.subprocess, "run", _fake_run(0, "video\n", ""))
|
|
result = check_video_integrity(video_file, deep=False)
|
|
assert result.ok
|
|
|
|
|
|
def test_quick_fails_without_video_stream(monkeypatch, video_file):
|
|
monkeypatch.setattr(integrity.shutil, "which", lambda _: "/usr/bin/ffprobe")
|
|
monkeypatch.setattr(integrity.subprocess, "run", _fake_run(0, "", ""))
|
|
result = check_video_integrity(video_file, deep=False)
|
|
assert not result.ok and "video stream" in result.error.lower()
|
|
|
|
|
|
def test_missing_file_is_unchecked_not_corrupt(monkeypatch, tmp_path):
|
|
monkeypatch.setattr(integrity.shutil, "which", lambda _: "/usr/bin/ffmpeg")
|
|
result = check_video_integrity(tmp_path / "nope.mkv", deep=True)
|
|
assert not result.ok and not result.checked and "neexistuje" in result.error
|
|
|
|
|
|
def test_missing_tool_is_unchecked(monkeypatch, video_file):
|
|
monkeypatch.setattr(integrity.shutil, "which", lambda _: None)
|
|
result = check_video_integrity(video_file, deep=True)
|
|
assert not result.ok and not result.checked
|
|
assert "ffmpeg" in result.error
|
|
|
|
|
|
def test_timeout_reports_failure(monkeypatch, video_file):
|
|
monkeypatch.setattr(integrity.shutil, "which", lambda _: "/usr/bin/ffmpeg")
|
|
|
|
def raise_timeout(cmd, **kwargs):
|
|
raise subprocess.TimeoutExpired(cmd, 1)
|
|
|
|
monkeypatch.setattr(integrity.subprocess, "run", raise_timeout)
|
|
result = check_video_integrity(video_file, deep=True, timeout=1)
|
|
assert not result.ok and result.checked and "timeout" in result.error.lower()
|
|
|
|
|
|
def test_ffmpeg_available_reflects_which(monkeypatch):
|
|
monkeypatch.setattr(integrity.shutil, "which", lambda _: "/usr/bin/ffmpeg")
|
|
assert integrity.ffmpeg_available(deep=True)
|
|
monkeypatch.setattr(integrity.shutil, "which", lambda _: None)
|
|
assert not integrity.ffmpeg_available(deep=True)
|
|
|
|
|
|
def test_scan_collects_problems_and_progress(monkeypatch, tmp_path):
|
|
"""FileManager.scan_video_integrity aggregates per-file results."""
|
|
from src.core.tag_manager import TagManager
|
|
from src.core.file_manager import FileManager
|
|
from src.core.file import File
|
|
|
|
good = tmp_path / "good.mkv"
|
|
bad = tmp_path / "bad.mkv"
|
|
good.write_bytes(b"x")
|
|
bad.write_bytes(b"y")
|
|
|
|
tm = TagManager()
|
|
fm = FileManager(tm)
|
|
files = [File(good, tm), File(bad, tm)]
|
|
|
|
def fake_check(path, deep=True):
|
|
ok = Path(path).name == "good.mkv"
|
|
return IntegrityResult(Path(path), ok=ok, error="" if ok else "broken")
|
|
|
|
monkeypatch.setattr("src.core.file_manager.ffmpeg_available", lambda deep: True)
|
|
monkeypatch.setattr("src.core.file_manager.check_video_integrity", fake_check)
|
|
|
|
seen: list[int] = []
|
|
result = fm.scan_video_integrity(
|
|
files, deep=True, on_progress=lambda d, t, f: seen.append(d))
|
|
|
|
assert result["ok"] and result["checked"] == 2
|
|
assert [r.path.name for r in result["problems"]] == ["bad.mkv"]
|
|
assert seen == [1, 2]
|
|
|
|
|
|
def test_scan_stops_on_cancel(monkeypatch, tmp_path):
|
|
from src.core.tag_manager import TagManager
|
|
from src.core.file_manager import FileManager
|
|
from src.core.file import File
|
|
|
|
files = []
|
|
for i in range(3):
|
|
p = tmp_path / f"m{i}.mkv"
|
|
p.write_bytes(b"x")
|
|
files.append(File(p, TagManager()))
|
|
|
|
monkeypatch.setattr("src.core.file_manager.ffmpeg_available", lambda deep: True)
|
|
monkeypatch.setattr(
|
|
"src.core.file_manager.check_video_integrity",
|
|
lambda path, deep=True: IntegrityResult(Path(path), ok=True))
|
|
|
|
fm = FileManager(TagManager())
|
|
result = fm.scan_video_integrity(files, should_cancel=lambda: True)
|
|
assert result["cancelled"] and result["checked"] == 0
|
|
|
|
|
|
def test_scan_reports_missing_ffmpeg(monkeypatch, tmp_path):
|
|
from src.core.tag_manager import TagManager
|
|
from src.core.file_manager import FileManager
|
|
|
|
monkeypatch.setattr("src.core.file_manager.ffmpeg_available", lambda deep: False)
|
|
fm = FileManager(TagManager())
|
|
result = fm.scan_video_integrity([], deep=True)
|
|
assert not result["ok"] and "ffmpeg" in result["error"]
|