ffx/tests/unit/test_ffx_controller.py

from __future__ import annotations

from pathlib import Path
import sys
import unittest
from unittest.mock import patch
from types import SimpleNamespace


SRC_ROOT = Path(__file__).resolve().parents[2] / "src"

if str(SRC_ROOT) not in sys.path:
    sys.path.insert(0, str(SRC_ROOT))


from ffx.ffx_controller import FfxController  # noqa: E402
from ffx.logging_utils import get_ffx_logger  # noqa: E402
from ffx.media_descriptor import MediaDescriptor  # noqa: E402
from ffx.show_descriptor import ShowDescriptor  # noqa: E402
from ffx.track_codec import TrackCodec  # noqa: E402
from ffx.track_descriptor import TrackDescriptor  # noqa: E402
from ffx.track_type import TrackType  # noqa: E402
from ffx.video_encoder import VideoEncoder  # noqa: E402


class StaticConfig:
    def __init__(self, data: dict | None = None):
        self._data = data or {}

    def getData(self):
        return self._data


class FfxControllerTests(unittest.TestCase):
    def make_context(self, video_encoder: VideoEncoder) -> dict:
        return {
            "logger": get_ffx_logger(),
            "config": StaticConfig(),
            "video_encoder": video_encoder,
            "dry_run": False,
            "perform_cut": False,
            "bitrates": {
                "stereo": "112k",
                "ac3": "256k",
                "dts": "320k",
            },
        }

    def make_media_descriptors(self) -> tuple[MediaDescriptor, MediaDescriptor]:
        descriptor = MediaDescriptor(
            track_descriptors=[
                TrackDescriptor(
                    index=0,
                    source_index=0,
                    sub_index=0,
                    track_type=TrackType.VIDEO,
                    codec_name=TrackCodec.H264,
                )
            ]
        )
        source_descriptor = MediaDescriptor(
            track_descriptors=[
                TrackDescriptor(
                    index=0,
                    source_index=0,
                    sub_index=0,
                    track_type=TrackType.VIDEO,
                    codec_name=TrackCodec.H264,
                )
            ]
        )
        return descriptor, source_descriptor

    def test_vp9_run_job_emits_file_level_encoding_quality_metadata(self):
        context = self.make_context(VideoEncoder.VP9)
        target_descriptor, source_descriptor = self.make_media_descriptors()
        controller = FfxController(context, target_descriptor, source_descriptor)
        commands = []

        with (
            patch.object(
                controller,
                "executeCommandSequence",
                side_effect=lambda command: commands.append(command) or ("", "", 0),
            ),
            patch("ffx.ffx_controller.os.path.exists", return_value=False),
        ):
            controller.runJob(
                "input.mkv",
                "output.webm",
                targetFormat="webm",
                chainIteration=[
                    {
                        "identifier": "quality",
                        "parameters": {"quality": 27},
                    }
                ],
            )

        self.assertEqual(2, len(commands))
        self.assertIn("-metadata:g", commands[1])
        self.assertIn("ENCODING_QUALITY=27", commands[1])
        self.assertFalse(
            any(token.startswith("ENCODING_PRESET=") for token in commands[1])
        )

    def test_av1_run_job_emits_file_level_quality_and_preset_metadata(self):
        context = self.make_context(VideoEncoder.AV1)
        target_descriptor, source_descriptor = self.make_media_descriptors()
        controller = FfxController(context, target_descriptor, source_descriptor)
        commands = []

        with patch.object(
            controller,
            "executeCommandSequence",
            side_effect=lambda command: commands.append(command) or ("", "", 0),
        ):
            controller.runJob(
                "input.mkv",
                "output.webm",
                targetFormat="webm",
                chainIteration=[
                    {
                        "identifier": "quality",
                        "parameters": {"quality": 29},
                    },
                    {
                        "identifier": "preset",
                        "parameters": {"preset": 7},
                    },
                ],
            )

        self.assertEqual(1, len(commands))
        self.assertIn("-metadata:g", commands[0])
        self.assertIn("ENCODING_QUALITY=29", commands[0])
        self.assertIn("ENCODING_PRESET=7", commands[0])

    def test_run_job_uses_show_quality_when_pattern_quality_is_unset(self):
        context = self.make_context(VideoEncoder.H264)
        target_descriptor, source_descriptor = self.make_media_descriptors()
        controller = FfxController(context, target_descriptor, source_descriptor)
        commands = []
        show_descriptor = ShowDescriptor(id=1, name="Show", year=2024, quality=23)
        pattern = SimpleNamespace(quality=0)

        with (
            patch.object(
                controller,
                "executeCommandSequence",
                side_effect=lambda command: commands.append(command) or ("", "", 0),
            ),
            patch.object(context["logger"], "info") as mocked_info,
        ):
            controller.runJob(
                "input.mkv",
                "output.mkv",
                chainIteration=[],
                currentPattern=pattern,
                currentShowDescriptor=show_descriptor,
            )

        self.assertEqual(1, len(commands))
        self.assertIn("ENCODING_QUALITY=23", commands[0])
        mocked_info.assert_any_call("Setting quality 23 from show")

    def test_run_job_prefers_pattern_quality_over_show_quality(self):
        context = self.make_context(VideoEncoder.H264)
        target_descriptor, source_descriptor = self.make_media_descriptors()
        controller = FfxController(context, target_descriptor, source_descriptor)
        commands = []
        show_descriptor = ShowDescriptor(id=1, name="Show", year=2024, quality=23)
        pattern = SimpleNamespace(quality=19)

        with (
            patch.object(
                controller,
                "executeCommandSequence",
                side_effect=lambda command: commands.append(command) or ("", "", 0),
            ),
            patch.object(context["logger"], "info") as mocked_info,
        ):
            controller.runJob(
                "input.mkv",
                "output.mkv",
                chainIteration=[],
                currentPattern=pattern,
                currentShowDescriptor=show_descriptor,
            )

        self.assertEqual(1, len(commands))
        self.assertIn("ENCODING_QUALITY=19", commands[0])
        mocked_info.assert_any_call("Setting quality 19 from pattern")


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