"""Test prep ddex.""" from unittest.mock import MagicMock from ddex_ingester_common.constants.ddex_providers import \ SME, SME_ANALYTICS_PROVIDER, SOM_LIVRE_VENDOR_ID from ddex_ingester_common.models.s3.track import Track from ddex_ingester_common.schemas.s3_schema import LocalizedTitleSchema, \ S3Schema from ddex_ingester_common.utils.prep_ddex import format_localizations, \ prep_artists, prep_cline, prep_composers, prep_forbidden_sequence, \ prep_lyrics_language, prep_missing_track_original_release_date, \ prep_original_release_date, prep_participants, prep_plines, \ prep_preorder_tracks, prep_track_lyrics, prep_track_name_length, \ prep_track_sequence_and_volume_number, prep_us_publishing_obligation, \ prep_video_content_language, remove_track_isrc_duplicates from freezegun import freeze_time import pytest @pytest.mark.parametrize('localization,expected', [ ( # No duplicates [ { 'language_code': 'JA-JPAN', 'title': 'Locale', 'version': None } ], [ { 'language_code': 'JA-JPAN', 'title': 'Locale', 'version': None } ]), ( # Duplicate Japanese [ { 'language_code': 'JA-JPAN', 'title': 'Locale', 'version': None }, { 'language_code': 'JA-KANA', 'title': 'Locale', 'version': None }, { 'language_code': 'RU', 'title': 'Locale', 'version': None }, { 'language_code': 'EN', 'title': 'Locale', 'version': None } ], [ { 'language_code': 'JA-JPAN', 'title': 'Locale', 'version': None }, { 'language_code': 'RU', 'title': 'Locale', 'version': None }, { 'language_code': 'EN', 'title': 'Locale', 'version': None } ]), ( # No Japanese [ { 'language_code': 'EN', 'title': 'Locale', 'version': None }, { 'language_code': 'RU', 'title': 'Locale', 'version': None } ], [ { 'language_code': 'EN', 'title': 'Locale', 'version': None }, { 'language_code': 'RU', 'title': 'Locale', 'version': None } ]), ( # Only Japanese katakana [ { 'language_code': 'JA-KANA', 'title': 'Locale', 'version': None }, { 'language_code': 'RU', 'title': 'Locale', 'version': None } ], [ { 'language_code': 'JA-KANA', 'title': 'Locale', 'version': None }, { 'language_code': 'RU', 'title': 'Locale', 'version': None } ]), ( # Multiple Japanese kana (This probably wouldn't happen) # noqa [ { 'language_code': 'JA-JPAN', 'title': 'Locale', 'version': None }, { 'language_code': 'JA-KANA', 'title': 'Locale', 'version': None }, { 'language_code': 'JA-KANA', 'title': 'Locale', 'version': None }, { 'language_code': 'RU', 'title': 'Locale', 'version': None } ], [ { 'language_code': 'JA-JPAN', 'title': 'Locale', 'version': None }, { 'language_code': 'RU', 'title': 'Locale', 'version': None } ]), ( # Empty localization [], [] ) ]) def test_format_localizations(localization, expected): """Test formatter deals with duplicates properly.""" logger = MagicMock() assert LocalizedTitleSchema().dump( format_localizations(SME, LocalizedTitleSchema().load(localization, many=True), logger), many=True) == expected @pytest.mark.parametrize('localization,expected', [([{'language_code': 'CHI', 'title': 'Locale', 'version': None}], [])]) def test_localizations_sme_analytics(localization, expected): """Test localization formatter removes all localizat for SME analytics.""" logger = MagicMock() assert LocalizedTitleSchema().dump( format_localizations(SME_ANALYTICS_PROVIDER, LocalizedTitleSchema().load(localization, many=True), logger), many=True) == expected def test_prep_lyrics_language_default(s3_prep_ddex): """Test prep_lyrics_language defaults to metadata_language.""" s3_data = S3Schema().load(s3_prep_ddex) language = 'ENG' s3_data.product.metadata_language = language prep_lyrics_language(s3_data) assert s3_data.tracks[0].lyrics_language == language def test_prep_lyrics_language_existing_value(s3_prep_ddex): """Test prep_lyrics_language with an existing lyrics_language.""" s3_data = S3Schema().load(s3_prep_ddex) language = 'ENG' s3_data.tracks[0].lyrics_language = language prep_lyrics_language(s3_data) assert s3_data.tracks[0].lyrics_language == language def test_prep_video_content_language(s3_prep_ddex): """Test prep_video_content_language.""" s3_data = S3Schema().load(s3_prep_ddex) s3_data.video.content_type = 'Music (instrumental)' prep_video_content_language(s3_data) assert s3_data.video.content_language == 'N/A' def test_prep_video_content_language_default(s3_prep_ddex): """Test prep_video_content_language defaults to title_language.""" s3_data = S3Schema().load(s3_prep_ddex) language = 'ENG' s3_data.video.title_language = language prep_video_content_language(s3_data) assert s3_data.video.content_language == language def test_prep_video_content_language_existing_value(s3_prep_ddex): """Test prep_video_content_language with an existing lyrics_language.""" s3_data = S3Schema().load(s3_prep_ddex) language = 'ENG' s3_data.video.content_language = language prep_video_content_language(s3_data) assert s3_data.video.content_language == language def test_prep_composers_default(s3_prep_ddex): """Test prep_composers defaults to track composers.""" s3_data = S3Schema().load(s3_prep_ddex) prep_composers(s3_data) assert 'Composer' in s3_data.product.display_artists[0].roles assert s3_data.product.display_artists[3]['roles'] == ['Composer'] def test_prep_composers_existing(s3_prep_ddex): """Test prep_composers uses product composers if available.""" s3_data = S3Schema().load(s3_prep_ddex) s3_data.product.display_artists[0].roles.append('Composer') nr_display_artists = len(s3_data.product.display_artists) prep_composers(s3_data) assert len(s3_data.product.display_artists) == nr_display_artists def test_prep_us_publishing_obligation( s3_prep_ddex): """Test prep_us_publishing_obligation.""" is_ccm_som_livre_grps_ddex_ingestion_enabled = True s3_data = S3Schema().load(s3_prep_ddex) s3_data.tracks[0].us_publishing_obligation = None prep_us_publishing_obligation(s3_data, is_ccm_som_livre_grps_ddex_ingestion_enabled) assert s3_data.tracks[ 0].us_publishing_obligation == 'HUNDREDPERCENTCONTROLLED' # noqa def test_prep_us_publishing_obligation_som_livre(s3_prep_ddex): """Test prep_us_publishing_obligation for Som Livre which is hard-coded.""" is_ccm_som_livre_grps_ddex_ingestion_enabled = True s3_data = S3Schema().load(s3_prep_ddex) s3_data.tracks[0].us_publishing_obligation = None s3_data.product.vendor_id = SOM_LIVRE_VENDOR_ID prep_us_publishing_obligation(s3_data, is_ccm_som_livre_grps_ddex_ingestion_enabled) assert s3_data.tracks[ 0].us_publishing_obligation == 'PARTIALLYCONTROLLED' # noqa def test_prep_artists(s3_prep_ddex_artists): """Test prep_artists.""" s3_data = S3Schema().load(s3_prep_ddex_artists) assert len(s3_data.tracks[0].display_artists) == 3 assert s3_data.tracks[0].display_artists[0].name == 'Various Artists' assert s3_data.tracks[0].display_artists[1].name == 'Composer Name' assert s3_data.tracks[0].display_artists[2].name == 'Artisti Vari' assert s3_data.tracks[0].resource_contributors[1].name == 'Various Artists' assert s3_data.tracks[0].resource_contributors[1].localized_names[0].name \ == 'Various Artists' prep_artists(s3_data, SME_ANALYTICS_PROVIDER) assert len(s3_data.tracks[0].display_artists) == 3 assert s3_data.tracks[0].display_artists[0].name == 'Multiple Artists' assert s3_data.tracks[0].display_artists[1].name == 'Composer Name' assert s3_data.tracks[0].display_artists[2].name == 'Multiple Artists' assert \ s3_data.tracks[0].resource_contributors[1].name == 'Multiple Artists' assert s3_data.tracks[0].resource_contributors[1].localized_names[0].name \ == 'Multiple Artists' @pytest.mark.parametrize( 'p_line,original_release_date,expected_p_line', [('test', '2024-01-01', '2024 test'), ('2025 valid ine', '2000-01-01', '2025 valid ine'), ('invalid p_line', '2023-01-01', '2023 invalid p_line'), ('invalid p_line', '2022-01-01', '2022 invalid p_line'), ('1990, 1999 Arista Records, Inc.', '2022-01-01', '1990, 1999 Arista Records, Inc.'), ('Sony BMG Music Entertainment', '2021-01-01', '2021 Sony BMG Music Entertainment'), ('1997 BMG Entertainment', '2020-01-01', '1997 BMG Entertainment'), ('Originally Recorded 1970. All rights reserved by BMG Music', '2020-01-01', '2020 Originally Recorded 1970. All rights reserved by BMG Music'), ('1515 invalid p_line', '2027-01-01', '2027 invalid p_line'), ('3000 ENTERTAINMENT - 2022', '2027-01-01', '2027 3000 ENTERTAINMENT - 2022 .'), ('3014 Sony Music Entertainment (Japan) Inc.', '2027-01-01', '2027 Sony Music Entertainment (Japan) Inc.'), (None, None, None) ], ) def test_prep_plines(s3_prep_ddex, p_line, original_release_date, expected_p_line): """Test prep_plines.""" s3_data = S3Schema().load(s3_prep_ddex) s3_data.tracks[0].p_line = p_line s3_data.tracks[0].original_release_date = original_release_date prep_plines(s3_data, SME_ANALYTICS_PROVIDER) assert s3_data.tracks[0].p_line == expected_p_line def test_prep_plines_without_original_release_date(s3_prep_ddex): """Test prep_plines without_original_release_date.""" s3_data = S3Schema().load(s3_prep_ddex) s3_data.tracks[0].p_line = 'invalid p_line' s3_data.tracks[0].original_release_date = None s3_data.product.original_release_date = '2024-01-01' prep_plines(s3_data, SME_ANALYTICS_PROVIDER) assert s3_data.tracks[0].p_line == '2024 invalid p_line' def test_prep_product_pline(s3_prep_ddex): """Test product prep_plines.""" s3_data = S3Schema().load(s3_prep_ddex) s3_data.product.p_line = '2025 ' prep_plines(s3_data, SME_ANALYTICS_PROVIDER) assert s3_data.product.p_line == '2025 .' def test_prep_plines_without_valid_original_release_date(s3_prep_ddex): """Test prep_plines without_original_release_date.""" s3_data = S3Schema().load(s3_prep_ddex) s3_data.tracks[0].p_line = '1775 invalid p_line' s3_data.tracks[0].original_release_date = '1004-01-01' s3_data.product.original_release_date = '2024-01-01' prep_plines(s3_data, SME_ANALYTICS_PROVIDER) assert s3_data.tracks[0].p_line == '2024 invalid p_line' def test_prep_plines_using_start_date_time(s3_prep_ddex): """Test prep_plines without_original_release_date.""" s3_data = S3Schema().load(s3_prep_ddex) s3_data.tracks[0].p_line = 'invalid p_line' s3_data.tracks[0].original_release_date = None s3_data.product.original_release_date = None prep_plines(s3_data, SME_ANALYTICS_PROVIDER) assert s3_data.tracks[0].p_line == '2020 invalid p_line' def test_prep_track_lyrics(s3_prep_ddex): """Test prep_track_lyrics.""" s3_data = S3Schema().load(s3_prep_ddex) s3_data.tracks[0].lyrics = 'Fuck bla bla fucking fo bar aSsfUCk' prep_track_lyrics(s3_data, SME_ANALYTICS_PROVIDER) assert s3_data.tracks[0].lyrics == '*** bla bla *** fo bar ***' def test_prep_participants_artist_with_multiple_ids( s3_prep_ddex_multiple_spotify_uri): """Test prep_participants.""" s3_data = S3Schema().load(s3_prep_ddex_multiple_spotify_uri) s3_data.tracks[0].display_artists[1].spotify_uri = 'test_4' s3_data.tracks[0].display_artists[1].name = 'Kizz Daniel' s3_data.product.display_artists[1].spotify_uri = 'test_3' s3_data.product.display_artists[1].name = 'Kizz Daniel' s3_data.tracks[0].display_artists[0].apple_id = 'apple_id_1' s3_data.tracks[0].display_artists[0].name = 'Blaqbonez' s3_data.product.display_artists[0].apple_id = 'apple_id_10' s3_data.product.display_artists[0].name = 'Blaqbonez' prep_participants(s3_data, SME_ANALYTICS_PROVIDER) assert s3_data.tracks[0].display_artists[1].spotify_uri == 'test_3' assert s3_data.tracks[0].display_artists[1].name == 'Kizz Daniel' assert s3_data.product.display_artists[1].spotify_uri == 'test_3' assert s3_data.tracks[0].display_artists[1].name == 'Kizz Daniel' assert s3_data.tracks[0].display_artists[0].apple_id == 'apple_id_1' assert s3_data.tracks[0].display_artists[0].name == 'Blaqbonez' assert s3_data.product.display_artists[0].apple_id == 'apple_id_1' assert s3_data.tracks[0].display_artists[0].name == 'Blaqbonez' def test_prep_participants_artist_spotify_uri_new( s3_prep_ddex_multiple_spotify_uri): """Test prep_participants.""" s3_data = S3Schema().load(s3_prep_ddex_multiple_spotify_uri) s3_data.tracks[0].display_artists[1].spotify_uri = 'new' s3_data.tracks[0].display_artists[1].name = 'Kizz Daniel' s3_data.product.display_artists[0].spotify_uri = 'NEW' s3_data.product.display_artists[0].name = 'Blaqbonez' prep_participants(s3_data, SME_ANALYTICS_PROVIDER) assert s3_data.tracks[0].display_artists[1].spotify_uri is None assert s3_data.tracks[0].display_artists[1].name == 'Kizz Daniel' assert s3_data.product.display_artists[0].spotify_uri is None assert s3_data.tracks[0].display_artists[0].name == 'Blaqbonez' @pytest.mark.parametrize('original_release_date,result_is_none', [(None, True), ('2025-01-01', False), ('1901-01-01', False), ('3025-01-01', True), ('1025-01-01', True), ]) def test_prep_original_release_date(s3_prep_ddex, original_release_date, result_is_none): """Test prep_original_release_date.""" s3_data = S3Schema().load(s3_prep_ddex) s3_data.product.original_release_date = original_release_date prep_original_release_date(s3_data) assert (s3_data.product.original_release_date is None) == result_is_none def test_prep_track_name_length_to_long(s3_prep_ddex): """Make sure we truncate track names that are too long.""" s3_data = S3Schema().load(s3_prep_ddex) s3_data.tracks[ 0].track_name = 'A very long track name that exceeds the 200 character limit 123456789 123456789 123456789 123456789 123456789 123456789 123456789 123456789 123456789 123456789 123456789 123456789 123456789 123456789 1' # noqa: E501 prep_track_name_length(s3_data) assert len(s3_data.tracks[0].track_name) <= 200 assert s3_data.tracks[0].track_name[-3:] == '...' def test_prep_track_name_length_not_to_long(s3_prep_ddex): """Make sure we truncate track names that are not too long.""" s3_data = S3Schema().load(s3_prep_ddex) track_name = 'Normal length track name' s3_data.tracks[0].track_name = track_name prep_track_name_length(s3_data) assert s3_data.tracks[0].track_name == track_name def test_prep_forbidden_sequence(s3_prep_ddex): """Make sure we break up the invalid sequence.""" s3_data = S3Schema().load(s3_prep_ddex) s3_data.project.description = 'Description with ../invalid seq' s3_data.tracks[0].track_name = 'Name with ../ invalid sequence' s3_data.tracks[0].lyrics = 'Lyrics with ../ invalid sequence' prep_forbidden_sequence(s3_data) assert s3_data.project.description == 'Description with .. / invalid seq' assert s3_data.tracks[0].track_name == 'Name with .. / invalid sequence' assert s3_data.tracks[0].lyrics == 'Lyrics with .. / invalid sequence' def test_prep_forbidden_sequence_new_line(s3_prep_ddex): """Make sure we break up the invalid sequence.""" s3_data = S3Schema().load(s3_prep_ddex) s3_data.project.description = 'Description with ..\\n invalid seq' s3_data.tracks[0].track_name = 'Name with ..\\n invalid sequence' s3_data.tracks[0].lyrics = 'Lyrics with ..\\n invalid sequence' s3_data.tracks[0].p_line = 'Italy S.p.A../ Nuvole Production srl' prep_forbidden_sequence(s3_data) assert s3_data.project.description == 'Description with .. \\n invalid seq' assert s3_data.tracks[0].track_name == 'Name with .. \\n invalid sequence' assert s3_data.tracks[0].lyrics == 'Lyrics with .. \\n invalid sequence' assert s3_data.tracks[0].p_line == 'Italy S.p.A.. / Nuvole Production srl' def test_prep_forbidden_sequence_null(s3_prep_ddex): """Make sure the invalid sequence logic supports null values.""" s3_data = S3Schema().load(s3_prep_ddex) s3_data.project.description = None s3_data.tracks[0].track_name = None prep_forbidden_sequence(s3_data) assert s3_data.project.description is None assert s3_data.tracks[0].track_name is None @pytest.mark.parametrize('now,expected_tracks_resources', [ ('2025-01-01', ['A1']), ('2025-01-02', ['A1']), ('2025-01-02 23:00:00', ['A1']), ('2025-01-03', ['A1', 'A2', 'A4']), ('2025-01-03 00:00:01', ['A1', 'A2', 'A4']), ('2025-01-04', ['A1', 'A2', 'A4']), ('2025-01-05', ['A1', 'A2', 'A3', 'A4']), ('2025-01-06', ['A1', 'A2', 'A3', 'A4']), ('2025-01-07', ['A1', 'A2', 'A3', 'A4', 'A5']), ('2025-01-12', ['A1', 'A2', 'A3', 'A4', 'A5', 'A7']), ('2025-01-15', ['A1', 'A2', 'A3', 'A4', 'A5', 'A7']), ]) def test_prep_preorder_tracks(s3_prep_preorder_tracks_ddex, now, expected_tracks_resources): """Test prep_preorder_tracks.""" logger = MagicMock() s3_data = S3Schema().load(s3_prep_preorder_tracks_ddex) s3_data.product.is_confidential = True s3_data.tracks[0].original_release_date = '2025-01-01' s3_data.tracks[1].original_release_date = '2025-01-03' s3_data.tracks[3].original_release_date = '2025-01-03' s3_data.tracks[2].original_release_date = '2025-01-05' s3_data.tracks[4].original_release_date = '2025-01-07' s3_data.tracks[5].original_release_date = None s3_data.tracks[6].original_release_date = '2025-01-12' with freeze_time(now): prep_preorder_tracks(s3_data, False, SME_ANALYTICS_PROVIDER, logger) tracks_resources = [t.resource_reference for t in s3_data.tracks] assert expected_tracks_resources == tracks_resources @pytest.mark.parametrize('now,expected_tracks_resources', [ ('2025-01-02', ['A1', 'A2', 'A4']), ('2025-01-02 23:00:00', ['A1', 'A2', 'A4']), ('2025-01-03', ['A1', 'A2', 'A4']), ('2025-01-03 00:00:01', ['A1', 'A2', 'A4']), ('2025-01-04', ['A1', 'A2', 'A3', 'A4']), ('2025-01-05', ['A1', 'A2', 'A3', 'A4']), ('2025-01-06', ['A1', 'A2', 'A3', 'A4', 'A5']), ('2025-01-07', ['A1', 'A2', 'A3', 'A4', 'A5']), ('2025-01-12', ['A1', 'A2', 'A3', 'A4', 'A5', 'A7']), ('2025-01-15', ['A1', 'A2', 'A3', 'A4', 'A5', 'A7']), ]) def test_prep_preorder_tracks_non_confidential(s3_prep_preorder_tracks_ddex, now, expected_tracks_resources): """Test prep_preorder_tracks.""" logger = MagicMock() s3_data = S3Schema().load(s3_prep_preorder_tracks_ddex) s3_data.product.is_confidential = False s3_data.tracks[0].original_release_date = '2025-01-01' s3_data.tracks[1].original_release_date = '2025-01-03' s3_data.tracks[3].original_release_date = '2025-01-03' s3_data.tracks[2].original_release_date = '2025-01-05' s3_data.tracks[4].original_release_date = '2025-01-07' s3_data.tracks[5].original_release_date = None s3_data.tracks[6].original_release_date = '2025-01-12' with freeze_time(now): prep_preorder_tracks(s3_data, False, SME_ANALYTICS_PROVIDER, logger) tracks_resources = [t.resource_reference for t in s3_data.tracks] assert expected_tracks_resources == tracks_resources def test_artwork_ingestion_only(s3_prep_preorder_tracks_ddex): """Test prep_preorder_tracks with artwork_ingestion_only.""" logger = MagicMock() s3_data = S3Schema().load(s3_prep_preorder_tracks_ddex) s3_data.product.is_confidential = True prep_preorder_tracks(s3_data, True, SME_ANALYTICS_PROVIDER, logger) assert s3_data.tracks == [] def test_prep_track_sequence_number(s3_prep_preorder_tracks_ddex): """Test test_prep_track_sequence_number with artwork_ingestion_only.""" s3_data = S3Schema().load(s3_prep_preorder_tracks_ddex) s3_data.tracks = [ Track(sequence_number=1, volume=1), Track(sequence_number=2, volume=1), Track(sequence_number=3, volume=1), Track(sequence_number=5, volume=1), Track(sequence_number=7, volume=1), Track(sequence_number=19, volume=1), Track(sequence_number=1, volume=2), Track(sequence_number=2, volume=2), Track(sequence_number=3, volume=2), Track(sequence_number=4, volume=2), Track(sequence_number=1, volume=3), Track(sequence_number=5, volume=3), Track(sequence_number=9, volume=3), Track(sequence_number=1, volume=4), Track(sequence_number=0, volume=4), Track(sequence_number=1, volume=4), Track(sequence_number=1, volume=4), Track(sequence_number=1, volume=4), ] prep_track_sequence_and_volume_number(s3_data) assert [track.sequence_number for track in s3_data.tracks if track.volume == 1] == [1, 2, 3, 4, 5, 6] assert [track.sequence_number for track in s3_data.tracks if track.volume == 2] == [1, 2, 3, 4] assert [track.sequence_number for track in s3_data.tracks if track.volume == 3] == [1, 2, 3] assert [track.sequence_number for track in s3_data.tracks if track.volume == 4] == [1, 2, 3, 4, 5] def test_prep_track_volume_number(s3_prep_preorder_tracks_ddex): """Test test_prep_track_sequence_number with missing volume.""" s3_data = S3Schema().load(s3_prep_preorder_tracks_ddex) s3_data.tracks = [ Track(sequence_number=1, volume=2), Track(sequence_number=2, volume=2), Track(sequence_number=4, volume=2), Track(sequence_number=5, volume=2), Track(sequence_number=1, volume=4), Track(sequence_number=0, volume=4), Track(sequence_number=1, volume=4), ] prep_track_sequence_and_volume_number(s3_data) assert [track.sequence_number for track in s3_data.tracks if track.volume == 1] == [1, 2, 3, 4] assert [track.sequence_number for track in s3_data.tracks if track.volume == 2] == [1, 2, 3] def test_prep_track_sequence_volume_do_nothing(s3_prep_preorder_tracks_ddex): """Test test_prep_track_sequence_number with valid data.""" s3_data = S3Schema().load(s3_prep_preorder_tracks_ddex) s3_data.tracks = [ Track(sequence_number=1, volume=1, isrc='1_1'), Track(sequence_number=2, volume=1, isrc='2_1'), Track(sequence_number=3, volume=1, isrc='3_1'), Track(sequence_number=1, volume=2, isrc='1_2'), Track(sequence_number=2, volume=2, isrc='2_2'), Track(sequence_number=3, volume=2, isrc='3_2'), Track(sequence_number=4, volume=2, isrc='4_2'), Track(sequence_number=1, volume=3, isrc='1_3'), Track(sequence_number=2, volume=3, isrc='2_3'), Track(sequence_number=3, volume=3, isrc='3_3'), Track(sequence_number=4, volume=3, isrc='4_3'), ] prep_track_sequence_and_volume_number(s3_data) for track in s3_data.tracks: assert track.isrc == f'{track.sequence_number}_{track.volume}' def test_prep_track_duplicates(s3_prep_preorder_tracks_ddex): """Test prep_track_duplicates.""" s3_data = S3Schema().load(s3_prep_preorder_tracks_ddex) s3_data.tracks = [ Track(isrc='ISRC_1'), Track(isrc='ISRC_2'), Track(isrc='ISRC_1'), Track(isrc='ISRC_3'), Track(isrc='ISRC_3'), Track(isrc='ISRC_4'), Track(isrc='ISRC_5'), Track(isrc='ISRC_1'), Track(isrc='ISRC_5'), ] remove_track_isrc_duplicates(s3_data, SME_ANALYTICS_PROVIDER) assert [track.isrc for track in s3_data.tracks] == ['ISRC_1', 'ISRC_2', 'ISRC_3', 'ISRC_4', 'ISRC_5'] @pytest.mark.parametrize( 'c_line,expected_c_line', [('2024', '2024 .'), ('2025 ', '2025 .'), ('2024 foo', '2024 foo')], ) def test_prep_cline(s3_prep_ddex, c_line, expected_c_line): """Test prep_cline.""" s3_data = S3Schema().load(s3_prep_ddex) s3_data.product.c_line = c_line prep_cline(s3_data) assert s3_data.product.c_line == expected_c_line def test_prep_missing_track_original_release_date( s3_prep_preorder_tracks_ddex): """Test prep_missing_track_original_release_date.""" s3_data = S3Schema().load(s3_prep_preorder_tracks_ddex) s3_data.product.original_release_date = '2018-01-01' s3_data.tracks = [ Track(isrc='ISRC_1', original_release_date='2025-01-01'), Track(isrc='ISRC_2', original_release_date='2025-02-01'), Track(isrc='ISRC_3', original_release_date=None), Track(isrc='ISRC_4', original_release_date='2025-03-01'), ] prep_missing_track_original_release_date(s3_data, SME_ANALYTICS_PROVIDER) assert [track.original_release_date for track in s3_data.tracks] == [ '2025-01-01', '2025-02-01', '2018-01-01', '2025-03-01']