francwa
c303efea48
Add probe(video) -> MediaInfo | None to the MediaProber Protocol and implement it on FfprobeMediaProber. The standalone alfred/infrastructure/filesystem/ffprobe.py module is removed; all callers (analyze_release / probe_media tools, testing scripts) now go through the adapter. Tests for the probe path moved to tests/infrastructure/test_ffprobe_prober.py (patching subprocess.run at the adapter module level). Unblocks the upcoming inspect_release orchestrator, which needs the port — not a free function — to compose parse + main-video selection + probe in one shot.
156 lines
5.1 KiB
Python
156 lines
5.1 KiB
Python
"""Tests for :class:`FfprobeMediaProber`.
|
|
|
|
Covers the full-probe path (``probe()`` returning a ``MediaInfo``) by
|
|
patching ``subprocess.run`` at the adapter module level. The
|
|
subtitle-streams path is exercised by the subtitle domain tests via
|
|
the same adapter.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import subprocess
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
from alfred.infrastructure.probe import FfprobeMediaProber
|
|
|
|
_PROBER = FfprobeMediaProber()
|
|
_PATCH_TARGET = "alfred.infrastructure.probe.ffprobe_prober.subprocess.run"
|
|
|
|
|
|
def _ffprobe_result(returncode=0, stdout="{}", stderr="") -> MagicMock:
|
|
return MagicMock(returncode=returncode, stdout=stdout, stderr=stderr)
|
|
|
|
|
|
class TestProbe:
|
|
def test_timeout_returns_none(self, tmp_path):
|
|
f = tmp_path / "x.mkv"
|
|
f.write_bytes(b"")
|
|
with patch(
|
|
_PATCH_TARGET,
|
|
side_effect=subprocess.TimeoutExpired(cmd="ffprobe", timeout=30),
|
|
):
|
|
assert _PROBER.probe(f) is None
|
|
|
|
def test_nonzero_returncode_returns_none(self, tmp_path):
|
|
f = tmp_path / "x.mkv"
|
|
f.write_bytes(b"")
|
|
with patch(
|
|
_PATCH_TARGET,
|
|
return_value=_ffprobe_result(returncode=1, stderr="not a media file"),
|
|
):
|
|
assert _PROBER.probe(f) is None
|
|
|
|
def test_invalid_json_returns_none(self, tmp_path):
|
|
f = tmp_path / "x.mkv"
|
|
f.write_bytes(b"")
|
|
with patch(
|
|
_PATCH_TARGET,
|
|
return_value=_ffprobe_result(stdout="not json {"),
|
|
):
|
|
assert _PROBER.probe(f) is None
|
|
|
|
def test_parses_format_duration_and_bitrate(self, tmp_path):
|
|
f = tmp_path / "x.mkv"
|
|
f.write_bytes(b"")
|
|
payload = {
|
|
"format": {"duration": "1234.5", "bit_rate": "5000000"},
|
|
"streams": [],
|
|
}
|
|
with patch(
|
|
_PATCH_TARGET,
|
|
return_value=_ffprobe_result(stdout=json.dumps(payload)),
|
|
):
|
|
info = _PROBER.probe(f)
|
|
assert info is not None
|
|
assert info.duration_seconds == 1234.5
|
|
assert info.bitrate_kbps == 5000 # bit_rate // 1000
|
|
|
|
def test_invalid_numeric_format_fields_skipped(self, tmp_path):
|
|
f = tmp_path / "x.mkv"
|
|
f.write_bytes(b"")
|
|
payload = {
|
|
"format": {"duration": "garbage", "bit_rate": "also-bad"},
|
|
"streams": [],
|
|
}
|
|
with patch(
|
|
_PATCH_TARGET,
|
|
return_value=_ffprobe_result(stdout=json.dumps(payload)),
|
|
):
|
|
info = _PROBER.probe(f)
|
|
assert info is not None
|
|
assert info.duration_seconds is None
|
|
assert info.bitrate_kbps is None
|
|
|
|
def test_parses_streams(self, tmp_path):
|
|
f = tmp_path / "x.mkv"
|
|
f.write_bytes(b"")
|
|
payload = {
|
|
"format": {},
|
|
"streams": [
|
|
{
|
|
"index": 0,
|
|
"codec_type": "video",
|
|
"codec_name": "h264",
|
|
"width": 1920,
|
|
"height": 1080,
|
|
},
|
|
{
|
|
"index": 1,
|
|
"codec_type": "audio",
|
|
"codec_name": "ac3",
|
|
"channels": 6,
|
|
"channel_layout": "5.1",
|
|
"tags": {"language": "eng"},
|
|
"disposition": {"default": 1},
|
|
},
|
|
{
|
|
"index": 2,
|
|
"codec_type": "audio",
|
|
"codec_name": "aac",
|
|
"channels": 2,
|
|
"tags": {"language": "fra"},
|
|
},
|
|
{
|
|
"index": 3,
|
|
"codec_type": "subtitle",
|
|
"codec_name": "subrip",
|
|
"tags": {"language": "fra"},
|
|
"disposition": {"forced": 1},
|
|
},
|
|
],
|
|
}
|
|
with patch(
|
|
_PATCH_TARGET,
|
|
return_value=_ffprobe_result(stdout=json.dumps(payload)),
|
|
):
|
|
info = _PROBER.probe(f)
|
|
assert info.video_codec == "h264"
|
|
assert info.width == 1920 and info.height == 1080
|
|
assert len(info.audio_tracks) == 2
|
|
eng = info.audio_tracks[0]
|
|
assert eng.language == "eng"
|
|
assert eng.is_default is True
|
|
assert info.audio_tracks[1].is_default is False
|
|
assert len(info.subtitle_tracks) == 1
|
|
assert info.subtitle_tracks[0].is_forced is True
|
|
|
|
def test_first_video_stream_wins(self, tmp_path):
|
|
# The implementation only fills video_codec on the FIRST video stream.
|
|
f = tmp_path / "x.mkv"
|
|
f.write_bytes(b"")
|
|
payload = {
|
|
"format": {},
|
|
"streams": [
|
|
{"codec_type": "video", "codec_name": "h264", "width": 1920},
|
|
{"codec_type": "video", "codec_name": "hevc", "width": 3840},
|
|
],
|
|
}
|
|
with patch(
|
|
_PATCH_TARGET,
|
|
return_value=_ffprobe_result(stdout=json.dumps(payload)),
|
|
):
|
|
info = _PROBER.probe(f)
|
|
assert info.video_codec == "h264"
|
|
assert info.width == 1920
|