ffx/tests/unit/test_file_properties_probe.py

from __future__ import annotations

import json
import logging
from pathlib import Path
import sys
from types import SimpleNamespace
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"]
        )

    def test_use_pattern_false_skips_pattern_controller_construction(self):
        file_properties_module = self.import_module()

        with patch.object(
            file_properties_module,
            "PatternController",
            side_effect=AssertionError("PatternController should not be created"),
        ):
            file_properties = file_properties_module.FileProperties(
                self.make_context(),
                "/tmp/example_s01e01.mkv",
            )

        self.assertEqual(-1, file_properties.getShowId())
        self.assertIsNone(file_properties.getPattern())

    def test_cropdetect_uses_configured_window_and_caches_results(self):
        file_properties_module = self.import_module()
        file_properties_module.FileProperties._clear_cropdetect_cache()

        cropdetect_stderr = "\n".join(
            [
                "[Parsed_cropdetect_0] crop=1440:1080:240:0",
                "[Parsed_cropdetect_0] crop=1440:1080:240:0",
                "[Parsed_cropdetect_0] crop=1438:1080:242:0",
            ]
        )
        context = self.make_context()
        context["cropdetect"] = {"seek_seconds": 15, "duration_seconds": 45}

        with (
            patch.object(
                file_properties_module.os,
                "stat",
                return_value=SimpleNamespace(st_mtime_ns=1234, st_size=5678),
            ),
            patch.object(file_properties_module, "PatternController", DummyPatternController),
            patch.object(
                file_properties_module,
                "executeProcess",
                return_value=("", cropdetect_stderr, 0),
            ) as mocked_execute,
        ):
            file_properties = file_properties_module.FileProperties(
                context,
                "/tmp/example_s01e01.mkv",
            )

            first = file_properties.findCropArguments()
            second = file_properties.findCropArguments()

        self.assertEqual(first, second)
        self.assertEqual(
            {
                "output_width": "1440",
                "output_height": "1080",
                "x_offset": "240",
                "y_offset": "0",
            },
            first,
        )
        mocked_execute.assert_called_once_with(
            list(file_properties_module.FFMPEG_COMMAND_TOKENS)
            + [
                "-ss",
                "15",
                "-i",
                "/tmp/example_s01e01.mkv",
                "-t",
                "45",
                "-vf",
                "cropdetect",
            ]
            + list(file_properties_module.FFMPEG_NULL_OUTPUT_TOKENS),
            context=context,
        )

        file_properties_module.FileProperties._clear_cropdetect_cache()


if __name__ == "__main__":
    unittest.main()