241 lines
8.4 KiB
Python
241 lines
8.4 KiB
Python
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.helper import LogLevel # 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 {}
|
|
|
|
|
|
class NotificationCollector:
|
|
def __init__(self) -> None:
|
|
self.messages: list[str] = []
|
|
self.levels: list[LogLevel | None] = []
|
|
|
|
def __call__(self, message: str, level: LogLevel | None = None) -> None:
|
|
self.messages.append(message)
|
|
self.levels.append(level)
|
|
|
|
|
|
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 = NotificationCollector()
|
|
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,
|
|
loggingHandler = notifications,
|
|
)
|
|
|
|
mocked_execute.assert_not_called()
|
|
mocked_replace.assert_not_called()
|
|
self.assertEqual(["ffmpeg dry-run prepared."], notifications.messages)
|
|
self.assertEqual([None], notifications.levels)
|
|
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 = NotificationCollector()
|
|
|
|
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,
|
|
loggingHandler = notifications,
|
|
)
|
|
|
|
self.assertEqual(1, len(notifications.messages))
|
|
self.assertTrue(notifications.messages[0].startswith("ffmpeg: ffmpeg "))
|
|
self.assertEqual([LogLevel.DEBUG], notifications.levels)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|