ffx/tests/integration/test_cli_unmux.py

from __future__ import annotations

import json
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import unittest

from tests.support.ffx_bundle import (
    SourceTrackSpec,
    build_controller_context,
    create_source_fixture,
    dispose_controller_context,
)

from ffx.pattern_controller import PatternController
from ffx.show_controller import ShowController
from ffx.show_descriptor import ShowDescriptor
from ffx.track_codec import TrackCodec
from ffx.track_descriptor import TrackDescriptor
from ffx.track_type import TrackType

try:
    import pytest
except ImportError:  # pragma: no cover - unittest-only environments
    pytest = None

if pytest is not None:
    pytestmark = [pytest.mark.integration]


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


def run_ffx_unmux(workdir: Path, home_dir: Path, database_path: Path, *args: str) -> subprocess.CompletedProcess[str]:
    env = os.environ.copy()
    env["HOME"] = str(home_dir)
    existing_pythonpath = env.get("PYTHONPATH", "")
    env["PYTHONPATH"] = str(SRC_ROOT) if not existing_pythonpath else f"{SRC_ROOT}{os.pathsep}{existing_pythonpath}"

    command = [
        sys.executable,
        "-m",
        "ffx",
        "--database-file",
        str(database_path),
        "unmux",
        *args,
    ]
    return subprocess.run(command, cwd=workdir, env=env, capture_output=True, text=True)


class UnmuxCliTests(unittest.TestCase):
    def setUp(self):
        self.tempdir = tempfile.TemporaryDirectory()
        self.workdir = Path(self.tempdir.name)
        self.home_dir = self.workdir / "home"
        self.home_dir.mkdir()
        self.database_path = self.workdir / "test.db"

    def tearDown(self):
        self.tempdir.cleanup()

    def write_config(self, data: dict) -> None:
        config_dir = self.home_dir / ".local" / "etc"
        config_dir.mkdir(parents=True, exist_ok=True)
        (config_dir / "ffx.json").write_text(json.dumps(data), encoding="utf-8")

    def assertCompleted(self, completed):
        if completed.returncode != 0:
            self.fail(
                "FFX unmux failed\n"
                f"STDOUT:\n{completed.stdout}\n"
                f"STDERR:\n{completed.stderr}"
            )

    def seed_matching_show(self, pattern_expression: str, *, indicator_season_digits: int, indicator_episode_digits: int) -> None:
        context = build_controller_context(self.database_path)
        try:
            ShowController(context).updateShow(
                ShowDescriptor(
                    id=1,
                    name="Unmux Test Show",
                    year=2000,
                    indicator_season_digits=indicator_season_digits,
                    indicator_episode_digits=indicator_episode_digits,
                )
            )
            PatternController(context).savePatternSchema(
                {
                    "show_id": 1,
                    "pattern": pattern_expression,
                    "quality": 0,
                    "notes": "",
                },
                trackDescriptors=[
                    TrackDescriptor(
                        index=0,
                        source_index=0,
                        track_type=TrackType.VIDEO,
                        codec_name=TrackCodec.H264,
                        tags={},
                        disposition_set=set(),
                    )
                ],
            )
        finally:
            dispose_controller_context(context)

    def test_subtitles_only_without_output_directory_uses_configured_base_plus_label(self):
        self.write_config(
            {
                "subtitlesDirectory": "~/.local/var/sync/subtitles",
            }
        )
        source_filename = "unmux_s01e01.mkv"
        source_path = create_source_fixture(
            self.workdir,
            source_filename,
            [
                SourceTrackSpec(TrackType.VIDEO, identity="video-0"),
                SourceTrackSpec(
                    TrackType.SUBTITLE,
                    identity="subtitle-1",
                    language="eng",
                    subtitle_lines=("subtitle payload",),
                ),
            ],
        )

        completed = run_ffx_unmux(
            self.workdir,
            self.home_dir,
            self.database_path,
            "--subtitles-only",
            "--label",
            "dball",
            str(source_path),
        )
        self.assertCompleted(completed)

        expected_directory = self.home_dir / ".local" / "var" / "sync" / "subtitles" / "dball"
        self.assertTrue(expected_directory.is_dir(), expected_directory)

    def test_unmux_uses_configured_indicator_digits_in_output_filenames(self):
        self.write_config(
            {
                "defaultIndicatorSeasonDigits": 3,
                "defaultIndicatorEpisodeDigits": 4,
            }
        )
        source_filename = "unmux_s01e01.mkv"
        output_directory = self.workdir / "unmux-output"
        output_directory.mkdir()
        source_path = create_source_fixture(
            self.workdir,
            source_filename,
            [
                SourceTrackSpec(TrackType.VIDEO, identity="video-0"),
            ],
        )

        completed = run_ffx_unmux(
            self.workdir,
            self.home_dir,
            self.database_path,
            "--label",
            "dball",
            "--output-directory",
            str(output_directory),
            str(source_path),
        )
        self.assertCompleted(completed)

        output_filenames = sorted(path.name for path in output_directory.iterdir())
        self.assertEqual(1, len(output_filenames), output_filenames)
        self.assertTrue(
            output_filenames[0].startswith("dball_S001E0001_"),
            output_filenames,
        )

    def test_unmux_prefers_matched_show_indicator_digits_over_config_defaults(self):
        self.write_config(
            {
                "defaultIndicatorSeasonDigits": 4,
                "defaultIndicatorEpisodeDigits": 4,
            }
        )
        self.seed_matching_show(
            r"^unmux_([sS][0-9]+[eE][0-9]+)\.mkv$",
            indicator_season_digits=1,
            indicator_episode_digits=3,
        )
        source_filename = "unmux_s01e01.mkv"
        output_directory = self.workdir / "unmux-output"
        output_directory.mkdir()
        source_path = create_source_fixture(
            self.workdir,
            source_filename,
            [
                SourceTrackSpec(TrackType.VIDEO, identity="video-0"),
            ],
        )

        completed = run_ffx_unmux(
            self.workdir,
            self.home_dir,
            self.database_path,
            "--label",
            "dball",
            "--output-directory",
            str(output_directory),
            str(source_path),
        )
        self.assertCompleted(completed)

        output_filenames = sorted(path.name for path in output_directory.iterdir())
        self.assertEqual(1, len(output_filenames), output_filenames)
        self.assertTrue(
            output_filenames[0].startswith("dball_S1E001_"),
            output_filenames,
        )


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