ffx/tests/unit/test_cli_convert_diagnostics.py

from __future__ import annotations

import os
from pathlib import Path
import sys
import tempfile
import unittest
from unittest.mock import patch

from click.testing import CliRunner


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 import cli  # noqa: E402
from ffx.diagnostics import FfmpegSkipFileWarning, recordUnremediedIssue  # noqa: E402
from ffx.logging_utils import get_ffx_logger  # noqa: E402


class _FakeMediaDescriptor:
    def getVideoTracks(self):
        return []

    def getAudioTracks(self):
        return []

    def getSubtitleTracks(self):
        return []

    def getAttachmentTracks(self):
        return []

    def applyOverrides(self, overrides):
        return None


class _FakeFileProperties:
    def __init__(self, context, source_path):
        self.source_path = source_path

    def getShowId(self):
        return -1

    def getSeason(self):
        return -1

    def getEpisode(self):
        return -1

    def getMediaDescriptor(self):
        return _FakeMediaDescriptor()

    def getPattern(self):
        return None


class _FakeShiftedSeasonController:
    def __init__(self, context):
        self.context = context

    def shiftSeason(self, show_id, season, episode, patternId=None):
        return season, episode


class _FakeShowController:
    def __init__(self, context):
        self.context = context

    def getShowDescriptor(self, show_id):
        return None


class _FakeFfxController:
    calls: list[str] = []
    mode = "skip_first"

    def __init__(self, context, *args, **kwargs):
        self.context = context

    def runJob(self, sourcePath, *args, **kwargs):
        self.calls.append(sourcePath)
        if self.mode == "clean":
            return

        if self.mode == "warn_unhandled" and sourcePath.endswith("episode1.avi"):
            recordUnremediedIssue(
                self.context,
                sourcePath,
                "unhandled-warning",
            )
            return

        if self.mode == "skip_first" and sourcePath.endswith("episode1.avi"):
            message = (
                f"Skipping file {sourcePath}: ffmpeg still reported unset packet "
                + "timestamps after retry with -fflags +genpts."
            )
            recordUnremediedIssue(
                self.context,
                sourcePath,
                "retry-with-generated-pts",
            )
            self.context["logger"].warning(message)
            raise FfmpegSkipFileWarning(message)


class ConvertDiagnosticCliTests(unittest.TestCase):
    def setUp(self):
        logger = get_ffx_logger()
        for handler in list(logger.handlers):
            logger.removeHandler(handler)
            try:
                handler.close()
            except Exception:
                pass

        self.tempdir = tempfile.TemporaryDirectory()
        self.home_dir = Path(self.tempdir.name) / "home"
        self.home_dir.mkdir()
        self.database_path = Path(self.tempdir.name) / "test.db"
        self.source_dir = Path(self.tempdir.name) / "source"
        self.source_dir.mkdir()
        self.source_one = self.source_dir / "episode1.avi"
        self.source_two = self.source_dir / "episode2.avi"
        self.source_one.write_bytes(b"one")
        self.source_two.write_bytes(b"two")
        _FakeFfxController.calls = []
        _FakeFfxController.mode = "skip_first"

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

    def test_convert_continues_after_skipping_one_file_due_to_ffmpeg_diagnostic(self):
        runner = CliRunner()

        with (
            patch("ffx.file_properties.FileProperties", _FakeFileProperties),
            patch("ffx.ffx_controller.FfxController", _FakeFfxController),
            patch(
                "ffx.shifted_season_controller.ShiftedSeasonController",
                _FakeShiftedSeasonController,
            ),
            patch("ffx.show_controller.ShowController", _FakeShowController),
        ):
            result = runner.invoke(
                cli.ffx,
                [
                    "--database-file",
                    str(self.database_path),
                    "convert",
                    "--no-tmdb",
                    "--no-pattern",
                    str(self.source_one),
                    str(self.source_two),
                ],
                env={**os.environ, "HOME": str(self.home_dir)},
            )

        self.assertEqual(0, result.exit_code, result.output)
        self.assertEqual(
            [str(self.source_one), str(self.source_two)],
            _FakeFfxController.calls,
        )
        self.assertIn("Skipping file", result.output)
        self.assertIn("-fflags +genpts", result.output)
        self.assertIn("Files with ffmpeg findings that require review:", result.output)
        self.assertIn(
            "episode1.avi: retry-with-generated-pts",
            result.output,
        )

    def test_convert_prints_clean_summary_when_no_unremedied_issues_were_seen(self):
        runner = CliRunner()
        _FakeFfxController.mode = "clean"

        with (
            patch("ffx.file_properties.FileProperties", _FakeFileProperties),
            patch("ffx.ffx_controller.FfxController", _FakeFfxController),
            patch(
                "ffx.shifted_season_controller.ShiftedSeasonController",
                _FakeShiftedSeasonController,
            ),
            patch("ffx.show_controller.ShowController", _FakeShowController),
        ):
            result = runner.invoke(
                cli.ffx,
                [
                    "--database-file",
                    str(self.database_path),
                    "convert",
                    "--no-tmdb",
                    "--no-pattern",
                    str(self.source_one),
                    str(self.source_two),
                ],
                env={**os.environ, "HOME": str(self.home_dir)},
            )

        self.assertEqual(0, result.exit_code, result.output)
        self.assertIn(
            "All files converted with no issues.",
            result.output,
        )


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