Optimizes ffprobe usage

2026-04-11 16:00:01 +02:00
parent c384d54c12
commit 0939a0c6c2
5 changed files with 161 additions and 59 deletions
--- a/tests/unit/test_file_properties_probe.py
+++ b/tests/unit/test_file_properties_probe.py
@@ -0,0 +1,111 @@
+from __future__ import annotations
+
+import json
+import logging
+from pathlib import Path
+import sys
+import unittest
+from unittest.mock import patch
+
+
+SRC_ROOT = Path(__file__).resolve().parents[2] / "src"
+
+if str(SRC_ROOT) not in sys.path:
+    sys.path.insert(0, str(SRC_ROOT))
+
+
+class StaticConfig:
+    def getData(self):
+        return {}
+
+
+class DummyPatternController:
+    def __init__(self, context):
+        self.context = context
+
+    def matchFilename(self, filename):
+        return {}
+
+
+def make_logger(name: str) -> logging.Logger:
+    logger = logging.getLogger(name)
+    logger.handlers = []
+    logger.setLevel(logging.DEBUG)
+    logger.propagate = False
+    logger.addHandler(logging.NullHandler())
+    return logger
+
+
+class FilePropertiesProbeTests(unittest.TestCase):
+    def import_module(self):
+        try:
+            import ffx.file_properties as file_properties_module
+        except ModuleNotFoundError as ex:
+            if ex.name == "sqlalchemy":
+                self.skipTest("sqlalchemy is not installed in this environment")
+            raise
+        return file_properties_module
+
+    def make_context(self):
+        return {
+            "logger": make_logger("ffx-test-file-properties-probe"),
+            "config": StaticConfig(),
+            "database": {"session": object()},
+            "use_pattern": False,
+        }
+
+    def sample_probe_data(self):
+        return {
+            "format": {
+                "filename": "/tmp/example_s01e01.mkv",
+                "nb_streams": 2,
+                "format_name": "matroska,webm",
+            },
+            "streams": [
+                {
+                    "index": 0,
+                    "codec_name": "h264",
+                    "codec_type": "video",
+                    "disposition": {"default": 1},
+                    "tags": {},
+                },
+                {
+                    "index": 1,
+                    "codec_name": "aac",
+                    "codec_type": "audio",
+                    "channel_layout": "stereo",
+                    "channels": 2,
+                    "disposition": {"default": 0},
+                    "tags": {"language": "eng"},
+                },
+            ],
+        }
+
+    def test_format_and_stream_accessors_share_one_combined_probe(self):
+        file_properties_module = self.import_module()
+        probe_output = self.sample_probe_data()
+
+        with (
+            patch.object(file_properties_module, "PatternController", DummyPatternController),
+            patch.object(
+                file_properties_module,
+                "executeProcess",
+                return_value=(json.dumps(probe_output), "", 0),
+            ) as mocked_execute,
+        ):
+            file_properties = file_properties_module.FileProperties(
+                self.make_context(),
+                "/tmp/example_s01e01.mkv",
+            )
+
+            self.assertEqual(probe_output["format"], file_properties.getFormatData())
+            self.assertEqual(probe_output["streams"], file_properties.getStreamData())
+
+        mocked_execute.assert_called_once_with(
+            file_properties_module.FileProperties.FFPROBE_COMMAND_TOKENS
+            + ["/tmp/example_s01e01.mkv"]
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()