Adds diagnostics/remedy system

tests/unit/test_cli_convert_diagnostics.py

@@ -0,0 +1,211 @@
+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 ffmpeg findings requiring review.",
+            result.output,
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()