diff --git a/tests/test_probe.py b/tests/test_probe.py new file mode 100644 index 0000000..fcc1318 --- /dev/null +++ b/tests/test_probe.py @@ -0,0 +1,65 @@ +import json + +from tools.probe import Probe, probe_file + + +def _runner_for(payload): + def fake_runner(args): # mimics subprocess.run(...).stdout + return json.dumps(payload) + + return fake_runner + + +def test_probe_parses_streams_and_format(): + runner = _runner_for( + { + "streams": [ + { + "codec_type": "video", + "width": 1920, + "height": 1080, + "disposition": {"attached_pic": 0}, + }, + {"codec_type": "audio"}, + ], + "format": {"duration": "12.5"}, + } + ) + p = probe_file("x.mp4", runner=runner) + assert isinstance(p, Probe) + assert any(s["codec_type"] == "video" for s in p.streams) + assert p.format["duration"] == "12.5" + + +def test_probe_audio_only(): + runner = _runner_for( + {"streams": [{"codec_type": "audio"}], "format": {"duration": "30.0"}} + ) + p = probe_file("x.mp3", runner=runner) + assert [s["codec_type"] for s in p.streams] == ["audio"] + + +def test_probe_audio_with_cover_art(): + runner = _runner_for( + { + "streams": [ + {"codec_type": "audio"}, + { + "codec_type": "video", + "width": 600, + "height": 600, + "disposition": {"attached_pic": 1}, + }, + ], + "format": {"duration": "200.0"}, + } + ) + p = probe_file("x.mp3", runner=runner) + cover = [s for s in p.streams if s.get("disposition", {}).get("attached_pic")] + assert len(cover) == 1 + + +def test_probe_handles_missing_keys(): + runner = _runner_for({}) + p = probe_file("x.mp4", runner=runner) + assert p.streams == [] and p.format == {} diff --git a/tools/probe.py b/tools/probe.py new file mode 100644 index 0000000..a972d9d --- /dev/null +++ b/tools/probe.py @@ -0,0 +1,34 @@ +"""ffprobe wrapper -> parsed streams/format. Subprocess runner is injectable.""" + +from __future__ import annotations + +import json +import subprocess +from dataclasses import dataclass + + +@dataclass +class Probe: + streams: list + format: dict + + +def _default_runner(args) -> str: + return subprocess.run(args, capture_output=True, text=True, check=True).stdout + + +def probe_file(path, *, runner=_default_runner) -> Probe: + out = runner( + [ + "ffprobe", + "-v", + "quiet", + "-print_format", + "json", + "-show_format", + "-show_streams", + str(path), + ] + ) + data = json.loads(out) + return Probe(streams=data.get("streams", []), format=data.get("format", {}))