from __future__ import annotations from pathlib import Path import os import sys import tempfile 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.logging_utils import get_ffx_logger # noqa: E402 from ffx.media_descriptor import MediaDescriptor # noqa: E402 from ffx.metadata_editor import ( # noqa: E402 apply_metadata_edits, build_metadata_edit_command, build_metadata_edit_context, create_temporary_output_path, ) 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 getData(self): return {} def make_context(*, dry_run: bool = False) -> dict: return { "logger": get_ffx_logger(), "config": StaticConfig(), "dry_run": dry_run, "apply_metadata_cleanup": True, "apply_metadata_normalization": True, } def make_descriptor() -> MediaDescriptor: return MediaDescriptor( track_descriptors=[ TrackDescriptor( index=0, source_index=0, sub_index=0, track_type=TrackType.VIDEO, codec_name=TrackCodec.H264, tags={"title": "Main"}, ) ], tags={"TITLE": "Demo"}, ) class MetadataEditorTests(unittest.TestCase): def test_build_metadata_edit_context_forces_copy_without_signature(self): context = build_metadata_edit_context(make_context()) self.assertEqual(VideoEncoder.COPY, context["video_encoder"]) self.assertFalse(context["perform_cut"]) self.assertTrue(context["no_signature"]) self.assertEqual({}, context["encoding_metadata_tags"]) self.assertTrue(context["apply_metadata_cleanup"]) self.assertTrue(context["apply_metadata_normalization"]) def test_create_temporary_output_path_uses_same_directory_and_extension(self): with tempfile.TemporaryDirectory() as tmpdir: source_path = os.path.join(tmpdir, "episode.mkv") temporary_path = create_temporary_output_path(source_path) self.assertEqual(".mkv", Path(temporary_path).suffix) self.assertEqual(Path(source_path).parent, Path(temporary_path).parent) def test_build_metadata_edit_command_maps_all_streams_and_uses_single_copy_codec(self): context = build_metadata_edit_context(make_context()) baseline_descriptor = make_descriptor() draft_descriptor = baseline_descriptor.clone(context=context) command = build_metadata_edit_command( context, "/tmp/example.mkv", "/tmp/.edit.mkv", baseline_descriptor, draft_descriptor, ) self.assertEqual(1, command.count("-map")) self.assertEqual(1, command.count("-c")) self.assertNotIn("-c:v:0", command) self.assertNotIn("-c:a:0", command) self.assertNotIn("-c:s:0", command) self.assertEqual( ["-map", "0", "-c", "copy"], command[command.index("-map"):command.index("-c") + 2], ) def test_apply_metadata_edits_rewrites_via_temporary_file_then_replaces_source(self): context = make_context() baseline_descriptor = make_descriptor() draft_descriptor = baseline_descriptor.clone(context=context) source_path = "/tmp/example.mkv" expected_command = build_metadata_edit_command( build_metadata_edit_context(context), source_path, "/tmp/.edit.mkv", baseline_descriptor, draft_descriptor, ) with ( patch("ffx.metadata_editor.create_temporary_output_path", return_value="/tmp/.edit.mkv"), patch("ffx.metadata_editor.executeProcess", return_value=("", "", 0)) as mocked_execute, patch("ffx.metadata_editor.os.replace") as mocked_replace, ): result = apply_metadata_edits( context, source_path, baseline_descriptor, draft_descriptor, ) mocked_execute.assert_called_once_with(expected_command, context=build_metadata_edit_context(context)) mocked_replace.assert_called_once_with("/tmp/.edit.mkv", source_path) self.assertEqual( { "applied": True, "dry_run": False, "target_path": source_path, "command_sequence": expected_command, }, { "applied": result["applied"], "dry_run": result["dry_run"], "target_path": result["target_path"], "command_sequence": result["command_sequence"], }, ) self.assertIn("timings", result) self.assertIn("ffmpeg_seconds", result["timings"]) self.assertIn("replace_seconds", result["timings"]) self.assertIn("write_seconds", result["timings"]) def test_apply_metadata_edits_dry_run_skips_replace_and_cleans_temp_path(self): context = make_context(dry_run=True) baseline_descriptor = make_descriptor() draft_descriptor = baseline_descriptor.clone(context=context) notifications = [] expected_command = build_metadata_edit_command( build_metadata_edit_context(context), "/tmp/example.mkv", "/tmp/.edit.mkv", baseline_descriptor, draft_descriptor, ) with ( patch("ffx.metadata_editor.create_temporary_output_path", return_value="/tmp/.edit.mkv"), patch("ffx.metadata_editor.executeProcess") as mocked_execute, patch("ffx.metadata_editor.os.replace") as mocked_replace, ): result = apply_metadata_edits( context, "/tmp/example.mkv", baseline_descriptor, draft_descriptor, notify=notifications.append, ) mocked_execute.assert_not_called() mocked_replace.assert_not_called() self.assertEqual(["ffmpeg dry-run prepared."], notifications) self.assertEqual( { "applied": False, "dry_run": True, "target_path": "/tmp/.edit.mkv", "command_sequence": expected_command, }, { "applied": result["applied"], "dry_run": result["dry_run"], "target_path": result["target_path"], "command_sequence": result["command_sequence"], }, ) self.assertEqual( { "ffmpeg_seconds": 0.0, "replace_seconds": 0.0, "write_seconds": 0.0, }, result["timings"], ) def test_apply_metadata_edits_notifies_with_command_when_verbose(self): context = make_context() context["verbosity"] = 1 baseline_descriptor = make_descriptor() draft_descriptor = baseline_descriptor.clone(context=context) notifications = [] with ( patch("ffx.metadata_editor.create_temporary_output_path", return_value="/tmp/.edit.mkv"), patch("ffx.metadata_editor.executeProcess", return_value=("", "", 0)), patch("ffx.metadata_editor.os.replace"), ): apply_metadata_edits( context, "/tmp/example.mkv", baseline_descriptor, draft_descriptor, notify=notifications.append, ) self.assertEqual(1, len(notifications)) self.assertTrue(notifications[0].startswith("ffmpeg: ffmpeg ")) if __name__ == "__main__": unittest.main()