from __future__ import annotations from pathlib import Path import sys import unittest 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.media_descriptor import MediaDescriptor # noqa: E402 from ffx.media_descriptor_change_set import MediaDescriptorChangeSet # noqa: E402 from ffx.track_descriptor import TrackDescriptor # noqa: E402 from ffx.track_type import TrackType # noqa: E402 from ffx.i18n import set_current_language # noqa: E402 from ffx.logging_utils import get_ffx_logger # noqa: E402 class StaticConfig: def __init__(self, data: dict): self._data = data def getData(self): return self._data class MediaDescriptorChangeSetTests(unittest.TestCase): def tearDown(self): set_current_language("de") def test_non_primary_source_language_code_is_normalized_in_changed_track_metadata(self): context = { "logger": get_ffx_logger(), "config": StaticConfig({}), } source_track = TrackDescriptor( index=0, source_index=0, sub_index=0, track_type=TrackType.AUDIO, tags={"language": "ger", "title": "German Main"}, ) target_track = TrackDescriptor( index=0, source_index=0, sub_index=0, track_type=TrackType.AUDIO, tags={"language": "ger", "title": "German Main"}, ) change_set = MediaDescriptorChangeSet( context, MediaDescriptor(track_descriptors=[target_track]), MediaDescriptor(track_descriptors=[source_track]), ) metadata_tokens = change_set.generateMetadataTokens() self.assertIn("-metadata:s:a:0", metadata_tokens) self.assertIn("language=deu", metadata_tokens) self.assertNotIn("language=ger", metadata_tokens) def test_target_only_track_language_metadata_uses_primary_code(self): context = { "logger": get_ffx_logger(), "config": StaticConfig({}), } target_track = TrackDescriptor( index=0, source_index=0, sub_index=0, track_type=TrackType.AUDIO, tags={"language": "ger", "title": "German Main"}, ) change_set = MediaDescriptorChangeSet( context, MediaDescriptor(track_descriptors=[target_track]), ) metadata_tokens = change_set.generateMetadataTokens() self.assertIn("-metadata:s:a:0", metadata_tokens) self.assertIn("language=deu", metadata_tokens) self.assertNotIn("language=ger", metadata_tokens) def test_external_subtitle_preserves_source_only_tags_except_removed_keys(self): context = { "logger": get_ffx_logger(), "config": StaticConfig( { "metadata": { "streams": { "remove": ["BPS"], } } } ), } source_track = TrackDescriptor( index=0, source_index=0, sub_index=0, track_type=TrackType.SUBTITLE, tags={ "language": "eng", "title": "Embedded Title", "THIS_IS": "embedded-subtitle", "EXTERNAL_KEEP": "keep-me", "BPS": "remove-me", }, ) target_track = TrackDescriptor( index=0, source_index=0, sub_index=0, track_type=TrackType.SUBTITLE, tags={"language": "deu"}, external_source_file="/tmp/external-subtitle.vtt", ) change_set = MediaDescriptorChangeSet( context, MediaDescriptor(track_descriptors=[target_track]), MediaDescriptor(track_descriptors=[source_track]), ) metadata_tokens = change_set.generateMetadataTokens() self.assertIn("-metadata:s:s:0", metadata_tokens) self.assertIn("language=deu", metadata_tokens) self.assertIn("title=Embedded Title", metadata_tokens) self.assertIn("THIS_IS=embedded-subtitle", metadata_tokens) self.assertIn("EXTERNAL_KEEP=keep-me", metadata_tokens) self.assertNotIn("BPS=remove-me", metadata_tokens) self.assertIn("BPS=", metadata_tokens) def test_external_subtitle_normalizes_preserved_source_language_metadata(self): context = { "logger": get_ffx_logger(), "config": StaticConfig({}), } source_track = TrackDescriptor( index=0, source_index=0, sub_index=0, track_type=TrackType.SUBTITLE, tags={"language": "ger", "title": "German Subtitle"}, ) target_track = TrackDescriptor( index=0, source_index=0, sub_index=0, track_type=TrackType.SUBTITLE, tags={}, external_source_file="/tmp/external-subtitle.vtt", ) change_set = MediaDescriptorChangeSet( context, MediaDescriptor(track_descriptors=[target_track]), MediaDescriptor(track_descriptors=[source_track]), ) metadata_tokens = change_set.generateMetadataTokens() self.assertIn("-metadata:s:s:0", metadata_tokens) self.assertIn("language=deu", metadata_tokens) self.assertNotIn("language=ger", metadata_tokens) def test_subtitle_without_title_gets_language_name_when_normalization_enabled(self): set_current_language("de") context = { "logger": get_ffx_logger(), "config": StaticConfig({}), } source_track = TrackDescriptor( index=0, source_index=0, sub_index=0, track_type=TrackType.SUBTITLE, tags={"language": "ger"}, ) target_track = TrackDescriptor( index=0, source_index=0, sub_index=0, track_type=TrackType.SUBTITLE, tags={"language": "ger"}, ) change_set = MediaDescriptorChangeSet( context, MediaDescriptor(track_descriptors=[target_track]), MediaDescriptor(track_descriptors=[source_track]), ) metadata_tokens = change_set.generateMetadataTokens() change_set_obj = change_set.getChangeSetObj() self.assertIn("-metadata:s:s:0", metadata_tokens) self.assertIn("language=deu", metadata_tokens) self.assertIn("title=Deutsch", metadata_tokens) self.assertEqual( "Deutsch", change_set_obj["tracks"]["changed"][0]["tags"]["added"]["title"], ) def test_subtitle_without_title_uses_current_language_for_generated_title(self): set_current_language("en") context = { "logger": get_ffx_logger(), "config": StaticConfig({}), } source_track = TrackDescriptor( index=0, source_index=0, sub_index=0, track_type=TrackType.SUBTITLE, tags={"language": "ger"}, ) target_track = TrackDescriptor( index=0, source_index=0, sub_index=0, track_type=TrackType.SUBTITLE, tags={"language": "ger"}, ) change_set = MediaDescriptorChangeSet( context, MediaDescriptor(track_descriptors=[target_track]), MediaDescriptor(track_descriptors=[source_track]), ) metadata_tokens = change_set.generateMetadataTokens() self.assertIn("title=German", metadata_tokens) self.assertNotIn("title=Deutsch", metadata_tokens) def test_non_subtitle_track_without_title_does_not_get_language_name(self): context = { "logger": get_ffx_logger(), "config": StaticConfig({}), } source_track = TrackDescriptor( index=0, source_index=0, sub_index=0, track_type=TrackType.AUDIO, tags={"language": "ger"}, ) target_track = TrackDescriptor( index=0, source_index=0, sub_index=0, track_type=TrackType.AUDIO, tags={"language": "ger"}, ) change_set = MediaDescriptorChangeSet( context, MediaDescriptor(track_descriptors=[target_track]), MediaDescriptor(track_descriptors=[source_track]), ) metadata_tokens = change_set.generateMetadataTokens() self.assertIn("-metadata:s:a:0", metadata_tokens) self.assertIn("language=deu", metadata_tokens) self.assertNotIn("title=Deutsch", metadata_tokens) def test_target_only_tracks_still_emit_remove_tokens_for_configured_stream_keys(self): context = { "logger": get_ffx_logger(), "config": StaticConfig( { "metadata": { "remove": ["creation_time"], "streams": { "remove": ["BPS"], } } } ), } target_track = TrackDescriptor( index=0, source_index=0, sub_index=0, track_type=TrackType.AUDIO, tags={ "language": "eng", "title": "Main Audio", "BPS": "remove-me", "KEEP_ME": "keep-me", }, ) change_set = MediaDescriptorChangeSet( context, MediaDescriptor(tags={"creation_time": "remove-me"}, track_descriptors=[target_track]), ) metadata_tokens = change_set.generateMetadataTokens() self.assertIn("-metadata:g", metadata_tokens) self.assertIn("creation_time=", metadata_tokens) self.assertIn("-metadata:s:a:0", metadata_tokens) self.assertIn("BPS=", metadata_tokens) self.assertIn("KEEP_ME=keep-me", metadata_tokens) def test_cleanup_can_be_disabled_per_context(self): context = { "logger": get_ffx_logger(), "config": StaticConfig( { "metadata": { "remove": ["creation_time"], "streams": { "remove": ["BPS"], }, } } ), "apply_metadata_cleanup": False, } source_track = TrackDescriptor( index=0, source_index=0, sub_index=0, track_type=TrackType.AUDIO, tags={"BPS": "keep-me"}, ) target_track = TrackDescriptor( index=0, source_index=0, sub_index=0, track_type=TrackType.AUDIO, tags={"BPS": "keep-me"}, ) change_set = MediaDescriptorChangeSet( context, MediaDescriptor( tags={"creation_time": "keep-me"}, track_descriptors=[target_track], ), MediaDescriptor( tags={"creation_time": "keep-me"}, track_descriptors=[source_track], ), ) metadata_tokens = change_set.generateMetadataTokens() self.assertNotIn("creation_time=", metadata_tokens) self.assertNotIn("BPS=", metadata_tokens) def test_normalization_can_be_disabled_per_context(self): context = { "logger": get_ffx_logger(), "config": StaticConfig({}), "apply_metadata_normalization": False, } target_track = TrackDescriptor( index=0, source_index=0, sub_index=0, track_type=TrackType.AUDIO, tags={"language": "ger", "title": "German Main"}, ) change_set = MediaDescriptorChangeSet( context, MediaDescriptor(track_descriptors=[target_track]), ) metadata_tokens = change_set.generateMetadataTokens() self.assertIn("-metadata:s:a:0", metadata_tokens) self.assertIn("language=ger", metadata_tokens) self.assertNotIn("language=deu", metadata_tokens) if __name__ == "__main__": unittest.main()