from __future__ import annotations import json from pathlib import Path import tempfile import unittest from tests.support.ffx_bundle import ( PatternTrackSpec, SourceTrackSpec, create_source_fixture, expected_output_path, extract_first_subtitle_text, ffprobe_json, get_tag, prepare_pattern_database, run_ffx_convert, write_vtt, ) from ffx.track_type import TrackType try: import pytest except ImportError: # pragma: no cover - unittest-only environments pytest = None if pytest is not None: pytestmark = [pytest.mark.integration, pytest.mark.subtrack_mapping] class SubtrackMappingBundleTests(unittest.TestCase): def setUp(self): self.tempdir = tempfile.TemporaryDirectory() self.workdir = Path(self.tempdir.name) self.home_dir = self.workdir / "home" self.home_dir.mkdir() self.database_path = self.workdir / "test.db" def tearDown(self): self.tempdir.cleanup() def write_config(self, data: dict) -> None: config_dir = self.home_dir / ".local" / "etc" config_dir.mkdir(parents=True, exist_ok=True) (config_dir / "ffx.json").write_text(json.dumps(data), encoding="utf-8") def assertCompleted(self, completed): if completed.returncode != 0: self.fail( "FFX convert failed\n" f"STDOUT:\n{completed.stdout}\n" f"STDERR:\n{completed.stderr}" ) def test_pattern_reorders_and_omits_tracks_preserving_metadata_and_group_order(self): source_filename = "reorder_s01e01.mkv" source_path = create_source_fixture( self.workdir, source_filename, [ SourceTrackSpec(TrackType.VIDEO, identity="video-0", title="Video Zero"), SourceTrackSpec( TrackType.SUBTITLE, identity="subtitle-1", language="eng", title="First Subtitle", subtitle_lines=("first embedded subtitle",), ), SourceTrackSpec( TrackType.AUDIO, identity="audio-2", language="deu", title="German Audio", ), SourceTrackSpec( TrackType.SUBTITLE, identity="subtitle-3", language="fra", title="Second Subtitle", subtitle_lines=("second embedded subtitle",), ), SourceTrackSpec(TrackType.ATTACHMENT, attachment_name="ordered.ttf"), ], ) prepare_pattern_database( self.database_path, r"^reorder_(s[0-9]+e[0-9]+)\.mkv$", [ PatternTrackSpec( index=0, source_index=0, track_type=TrackType.VIDEO, tags={"THIS_IS": "video-0", "title": "Video Zero"}, ), PatternTrackSpec( index=1, source_index=2, track_type=TrackType.AUDIO, tags={"THIS_IS": "audio-2", "language": "deu", "title": "German Audio"}, ), PatternTrackSpec( index=2, source_index=1, track_type=TrackType.SUBTITLE, tags={"THIS_IS": "subtitle-1", "language": "eng", "title": "First Subtitle"}, ), ], ) completed = run_ffx_convert( self.workdir, self.home_dir, self.database_path, "--video-encoder", "copy", "--no-tmdb", "--no-prompt", "--no-signature", str(source_path), ) self.assertCompleted(completed) output_path = expected_output_path(self.workdir, source_filename) self.assertTrue(output_path.is_file(), output_path) streams = ffprobe_json(output_path)["streams"] self.assertEqual( [stream["codec_type"] for stream in streams], ["video", "audio", "subtitle", "attachment"], ) self.assertEqual( [get_tag(streams[index], "THIS_IS") for index in range(3)], ["video-0", "audio-2", "subtitle-1"], ) self.assertNotIn( "subtitle-3", [get_tag(stream, "THIS_IS") for stream in streams if stream["codec_type"] != "attachment"], ) self.assertEqual(streams[-1]["codec_name"], "ttf") extracted_subtitle = extract_first_subtitle_text(self.workdir, output_path) self.assertIn("first embedded subtitle", extracted_subtitle) self.assertNotIn("second embedded subtitle", extracted_subtitle) def test_cli_rearrange_streams_reorders_tracks_without_database_pattern(self): source_filename = "cli_s01e01.mkv" source_path = create_source_fixture( self.workdir, source_filename, [ SourceTrackSpec(TrackType.VIDEO, identity="video-0"), SourceTrackSpec(TrackType.AUDIO, identity="audio-1", language="eng", title="First Audio"), SourceTrackSpec(TrackType.AUDIO, identity="audio-2", language="deu", title="Second Audio"), SourceTrackSpec(TrackType.SUBTITLE, identity="subtitle-3", language="eng", title="Subtitle"), ], ) completed = run_ffx_convert( self.workdir, self.home_dir, self.database_path, "--video-encoder", "copy", "--no-pattern", "--no-tmdb", "--no-prompt", "--no-signature", "--rearrange-streams", "0,2,1,3", str(source_path), ) self.assertCompleted(completed) output_path = expected_output_path(self.workdir, source_filename) streams = ffprobe_json(output_path)["streams"] self.assertEqual( [stream["codec_type"] for stream in streams], ["video", "audio", "audio", "subtitle"], ) self.assertEqual( [get_tag(stream, "THIS_IS") for stream in streams], ["video-0", "audio-2", "audio-1", "subtitle-3"], ) def test_no_pattern_stream_remove_list_clears_copied_stream_metadata(self): source_filename = "remove_tags_s01e01.mkv" self.write_config( { "metadata": { "streams": { "remove": ["BPS"], } } } ) source_path = create_source_fixture( self.workdir, source_filename, [ SourceTrackSpec( TrackType.VIDEO, identity="video-0", extra_tags={"BPS": "remove-me", "KEEP_ME": "video-keep"}, ), SourceTrackSpec( TrackType.AUDIO, identity="audio-1", language="eng", title="Main Audio", extra_tags={"BPS": "remove-me", "KEEP_ME": "audio-keep"}, ), ], ) completed = run_ffx_convert( self.workdir, self.home_dir, self.database_path, "--video-encoder", "copy", "--no-pattern", "--no-tmdb", "--no-prompt", "--no-signature", str(source_path), ) self.assertCompleted(completed) output_path = expected_output_path(self.workdir, source_filename) streams = ffprobe_json(output_path)["streams"] self.assertEqual( [stream["codec_type"] for stream in streams], ["video", "audio"], ) self.assertEqual(get_tag(streams[0], "THIS_IS"), "video-0") self.assertEqual(get_tag(streams[0], "KEEP_ME"), "video-keep") self.assertIsNone(get_tag(streams[0], "BPS")) self.assertEqual(get_tag(streams[1], "THIS_IS"), "audio-1") self.assertEqual(get_tag(streams[1], "KEEP_ME"), "audio-keep") self.assertIsNone(get_tag(streams[1], "BPS")) def test_pattern_validation_fails_for_nonexistent_source_track_reference(self): source_filename = "invalid_s01e01.mkv" source_path = create_source_fixture( self.workdir, source_filename, [ SourceTrackSpec(TrackType.VIDEO, identity="video-0"), SourceTrackSpec(TrackType.AUDIO, identity="audio-1"), SourceTrackSpec(TrackType.SUBTITLE, identity="subtitle-2"), ], ) prepare_pattern_database( self.database_path, r"^invalid_(s[0-9]+e[0-9]+)\.mkv$", [ PatternTrackSpec(index=0, source_index=0, track_type=TrackType.VIDEO), PatternTrackSpec(index=1, source_index=99, track_type=TrackType.SUBTITLE), ], ) completed = run_ffx_convert( self.workdir, self.home_dir, self.database_path, "--video-encoder", "copy", "--no-tmdb", "--no-prompt", "--no-signature", str(source_path), ) self.assertNotEqual(completed.returncode, 0) error_output = f"{completed.stdout}\n{completed.stderr}" self.assertIn("non-existent source track #99", error_output) self.assertFalse(expected_output_path(self.workdir, source_filename).exists()) def test_external_subtitle_file_replaces_payload_and_overrides_metadata(self): source_filename = "substitute_s01e01.mkv" self.write_config( { "metadata": { "streams": { "remove": ["BPS"], } } } ) source_path = create_source_fixture( self.workdir, source_filename, [ SourceTrackSpec(TrackType.VIDEO, identity="video-0"), SourceTrackSpec(TrackType.AUDIO, identity="audio-1", language="eng", title="Main Audio"), SourceTrackSpec( TrackType.SUBTITLE, identity="embedded-subtitle", language="eng", title="Embedded Title", extra_tags={"BPS": "remove-me", "EXTERNAL_KEEP": "keep-me"}, subtitle_lines=("embedded subtitle payload",), ), ], ) write_vtt( self.workdir / "substitute_s01e01_2_deu.vtt", ("external subtitle payload",), ) prepare_pattern_database( self.database_path, r"^substitute_(s[0-9]+e[0-9]+)\.mkv$", [ PatternTrackSpec(index=0, source_index=0, track_type=TrackType.VIDEO), PatternTrackSpec(index=1, source_index=1, track_type=TrackType.AUDIO), PatternTrackSpec(index=2, source_index=2, track_type=TrackType.SUBTITLE), ], ) completed = run_ffx_convert( self.workdir, self.home_dir, self.database_path, "--video-encoder", "copy", "--no-tmdb", "--no-prompt", "--no-signature", "--subtitle-directory", str(self.workdir), "--subtitle-prefix", "substitute", str(source_path), ) self.assertCompleted(completed) output_path = expected_output_path(self.workdir, source_filename) streams = ffprobe_json(output_path)["streams"] subtitle_stream = [stream for stream in streams if stream["codec_type"] == "subtitle"][0] self.assertEqual(get_tag(subtitle_stream, "language"), "deu") self.assertEqual(get_tag(subtitle_stream, "title"), "Embedded Title") self.assertEqual(get_tag(subtitle_stream, "THIS_IS"), "embedded-subtitle") self.assertEqual(get_tag(subtitle_stream, "EXTERNAL_KEEP"), "keep-me") self.assertIsNone(get_tag(subtitle_stream, "BPS")) extracted_subtitle = extract_first_subtitle_text(self.workdir, output_path) self.assertIn("external subtitle payload", extracted_subtitle) self.assertNotIn("embedded subtitle payload", extracted_subtitle) def test_subtitle_prefix_uses_configured_base_directory_when_directory_is_omitted(self): source_filename = "substitute_default_s01e01.mkv" subtitle_prefix = "substitute_default" subtitles_base_dir = self.home_dir / ".local" / "var" / "sync" / "subtitles" resolved_subtitle_dir = subtitles_base_dir / subtitle_prefix resolved_subtitle_dir.mkdir(parents=True, exist_ok=True) self.write_config( { "subtitlesDirectory": "~/.local/var/sync/subtitles", "metadata": { "streams": { "remove": ["BPS"], } } } ) source_path = create_source_fixture( self.workdir, source_filename, [ SourceTrackSpec(TrackType.VIDEO, identity="video-0"), SourceTrackSpec(TrackType.AUDIO, identity="audio-1", language="eng", title="Main Audio"), SourceTrackSpec( TrackType.SUBTITLE, identity="embedded-subtitle", language="eng", title="Embedded Title", extra_tags={"BPS": "remove-me", "EXTERNAL_KEEP": "keep-me"}, subtitle_lines=("embedded subtitle payload",), ), ], ) write_vtt( resolved_subtitle_dir / f"{subtitle_prefix}_s01e01_2_deu.vtt", ("external subtitle payload",), ) prepare_pattern_database( self.database_path, r"^substitute_default_(s[0-9]+e[0-9]+)\.mkv$", [ PatternTrackSpec(index=0, source_index=0, track_type=TrackType.VIDEO), PatternTrackSpec(index=1, source_index=1, track_type=TrackType.AUDIO), PatternTrackSpec(index=2, source_index=2, track_type=TrackType.SUBTITLE), ], ) completed = run_ffx_convert( self.workdir, self.home_dir, self.database_path, "--video-encoder", "copy", "--no-tmdb", "--no-prompt", "--no-signature", "--subtitle-prefix", subtitle_prefix, str(source_path), ) self.assertCompleted(completed) output_path = expected_output_path(self.workdir, source_filename) streams = ffprobe_json(output_path)["streams"] subtitle_stream = [stream for stream in streams if stream["codec_type"] == "subtitle"][0] self.assertEqual(get_tag(subtitle_stream, "language"), "deu") self.assertEqual(get_tag(subtitle_stream, "title"), "Embedded Title") self.assertEqual(get_tag(subtitle_stream, "THIS_IS"), "embedded-subtitle") self.assertEqual(get_tag(subtitle_stream, "EXTERNAL_KEEP"), "keep-me") self.assertIsNone(get_tag(subtitle_stream, "BPS")) extracted_subtitle = extract_first_subtitle_text(self.workdir, output_path) self.assertIn("external subtitle payload", extracted_subtitle) self.assertNotIn("embedded subtitle payload", extracted_subtitle) if __name__ == "__main__": unittest.main()