ffx/tests/unit/test_media_descriptor_change_set.py

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.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 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_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)


if __name__ == "__main__":
    unittest.main()