ffx/tests/unit/test_ffmpeg_diagnostics.py

from __future__ import annotations

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))


from ffx.diagnostics import (  # noqa: E402
    FfmpegCommandRunner,
    FfmpegDiagnosticMonitor,
    FfmpegSkipFileWarning,
    getUnremediedIssues,
    iterUnremediedIssueSummaryLines,
)


class RecordingLogger:
    def __init__(self):
        self.messages: list[str] = []

    def warning(self, message, *args, **kwargs):
        if args:
            message = message % args
        self.messages.append(str(message))


class FfmpegDiagnosticsTests(unittest.TestCase):
    def test_command_runner_retries_with_genpts_after_timestamp_warning(self):
        logger = RecordingLogger()
        context = {
            "logger": logger,
            "current_source_path": "tests/assets/avi/conan_S01E754_amalgam.avi",
        }
        runner = FfmpegCommandRunner(context)
        commands = []

        def fake_execute(commandSequence, **kwargs):
            commands.append(list(commandSequence))
            stderrLineHandler = kwargs["stderrLineHandler"]
            if len(commands) == 1:
                self.assertTrue(
                    stderrLineHandler(
                        "[matroska @ 0x1] Timestamps are unset in a packet for stream 0. "
                        + "This is deprecated and will stop working in the future."
                    )
                )
                return "", "timestamp warning\n", -15

            return "done", "", 0

        with patch("ffx.diagnostics.monitor.executeProcess", side_effect=fake_execute):
            out, err, rc = runner.execute(["ffmpeg", "-y", "-i", "input.avi", "output.mkv"])

        self.assertEqual("done", out)
        self.assertEqual("", err)
        self.assertEqual(0, rc)
        self.assertEqual(
            [
                ["ffmpeg", "-y", "-i", "input.avi", "output.mkv"],
                ["ffmpeg", "-fflags", "+genpts", "-y", "-i", "input.avi", "output.mkv"],
            ],
            commands,
        )
        self.assertEqual(
            [
                "ffmpeg reported unset packet timestamps for tests/assets/avi/conan_S01E754_amalgam.avi. "
                + "Stopping early and retrying with -fflags +genpts."
            ],
            logger.messages,
        )
        self.assertEqual({}, getUnremediedIssues(context))

    def test_command_runner_skips_file_when_timestamp_warning_persists_after_genpts(self):
        logger = RecordingLogger()
        context = {
            "logger": logger,
            "current_source_path": "tests/assets/avi/conan_S01E754_amalgam.avi",
        }
        runner = FfmpegCommandRunner(context)

        def fake_execute(commandSequence, **kwargs):
            stderrLineHandler = kwargs["stderrLineHandler"]
            self.assertTrue(
                stderrLineHandler(
                    "[matroska @ 0x1] Timestamps are unset in a packet for stream 0. "
                    + "This is deprecated and will stop working in the future."
                )
            )
            return "", "timestamp warning\n", -15

        with patch("ffx.diagnostics.monitor.executeProcess", side_effect=fake_execute):
            with self.assertRaises(FfmpegSkipFileWarning):
                runner.execute(
                    ["ffmpeg", "-fflags", "+genpts", "-y", "-i", "input.avi", "output.mkv"]
                )

        self.assertEqual(
            [
                "Skipping file tests/assets/avi/conan_S01E754_amalgam.avi: ffmpeg still reported "
                + "unset packet timestamps after retry with -fflags +genpts."
            ],
            logger.messages,
        )
        self.assertEqual(
            {
                "tests/assets/avi/conan_S01E754_amalgam.avi": ["retry-with-generated-pts"]
            },
            getUnremediedIssues(context),
        )

    def test_monitor_tracks_non_harmless_corrupt_mpeg_audio_remedy_in_summary(self):
        logger = RecordingLogger()
        context = {
            "logger": logger,
            "current_source_path": "tests/assets/avi/conan_S01E763_amalgam.avi",
        }
        monitor = FfmpegDiagnosticMonitor(
            context,
            ["ffmpeg", "-y", "-i", "input.avi", "output.mkv"],
        )

        self.assertFalse(monitor.handle_stderr_line("[mp3float @ 0x1] invalid block type"))
        self.assertFalse(
            monitor.handle_stderr_line(
                "[aist#0:1/mp3 @ 0x2] [dec:mp3float @ 0x3] Error submitting packet to decoder: "
                + "Invalid data found when processing input"
            )
        )

        self.assertEqual(
            [
                "ffmpeg reported damaged MPEG audio frames while converting "
                + "tests/assets/avi/conan_S01E763_amalgam.avi. FFX will continue, but the "
                + "output audio may contain gaps or glitches."
            ],
            logger.messages,
        )
        self.assertEqual(
            {
                "tests/assets/avi/conan_S01E763_amalgam.avi": ["warn-corrupt-mpeg-audio"]
            },
            getUnremediedIssues(context),
        )
        self.assertEqual(
            ["conan_S01E763_amalgam.avi: warn-corrupt-mpeg-audio"],
            iterUnremediedIssueSummaryLines(context),
        )

    def test_monitor_tracks_unhandled_diagnostic_for_summary(self):
        context = {
            "logger": RecordingLogger(),
            "current_source_path": "tests/assets/avi/example.avi",
        }
        monitor = FfmpegDiagnosticMonitor(
            context,
            ["ffmpeg", "-y", "-i", "input.avi", "output.mkv"],
        )

        self.assertFalse(
            monitor.handle_stderr_line(
                "[avi @ 0x1] Strange warning with no automatic remedy is present"
            )
        )

        self.assertEqual(
            {
                "tests/assets/avi/example.avi": ["unhandled-warning"]
            },
            getUnremediedIssues(context),
        )
        self.assertEqual(
            ["example.avi: unhandled-warning"],
            iterUnremediedIssueSummaryLines(context),
        )
        self.assertEqual(
            [
                "ffmpeg reported a diagnostic with no automatic remedy while converting "
                + "tests/assets/avi/example.avi. FFX will continue, but review the output "
                + "file. First unhandled line: [avi @ 0x1] Strange warning with no automatic remedy is present"
            ],
            context["logger"].messages,
        )


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