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 == {}

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", {}))