"""Test delivery metadata model.""" import json import uuid from contextlib import nullcontext from typing import Any, ContextManager import pytest from pytest_mock import MockerFixture from syrupy.assertion import SnapshotAssertion from testfixtures import compare from delivery_metadata.clients.art_relations.distribution_features import ( DistributionFeatures, ) from delivery_metadata.clients.art_relations.product import ArtRelationsProduct from delivery_metadata.clients.art_relations.product_localized_metadata import ( ArtRelationsProductLocalizedMetadata, ) from delivery_metadata.clients.art_relations.product_provided_store_artists import ( ArtRelationsProductProvidedStoreArtists, ) from delivery_metadata.clients.art_relations.product_video import ( ArtRelationsProductVideo, ) from delivery_metadata.clients.art_relations.schemas import ArtRelationsArtist from delivery_metadata.clients.art_relations.store_customer import ( ArtRelationsStoreCustomer, ) from delivery_metadata.clients.art_relations.track import ArtRelationsTrack from delivery_metadata.clients.art_relations.track_localized_metadata import ( ArtRelationsTrackLocalizedMetadata, ) from delivery_metadata.clients.art_relations.track_publisher import ( ArtRelationsTrackPublisher, ) from delivery_metadata.clients.art_relations.track_writer import ArtRelationsTrackWriter from delivery_metadata.clients.graphql.product import ( GlobalParticipant, GraphQlProduct, LabelParticipant, LabelSoundRecordingParticipation, Role, ) from delivery_metadata.clients.ows_territories.territory import TerritoryResponse from delivery_metadata.clients.ows_timed_release.timed_release import ( ProductTimedReleaseByStore, TimedRelease, ) from delivery_metadata.clients.ows_track.instant_grats import InstantGrats from delivery_metadata.clients.ows_track.performer import ( TrackPerformer, ) from delivery_metadata.constants import ( CommercialModelType, DeliveryType, DistributionFeatureId, DownloadStreamRights, InstrumentRole, ParticipantRole, ProductTypeId, StoreIds, TerritoryStandard, TrackOfferType, TrackTypes, UseType, store_setting_overrides, ) from delivery_metadata.exceptions import ( NoLabelParticipantForParticipant, PricingDataNotFound, ProductIneligible, ) from delivery_metadata.models import delivery_metadata as delivery_metadata_model from delivery_metadata.models.delivery_metadata import _hydrate_track from delivery_metadata.models.schemas import ( ArtworkAsset, AudioAsset, Delivery, DeliveryMetadata, DigitalBookletAsset, GenreMapping, LabelParticipantData, LabelParticipantMap, Participant, ParticipantFullnameLocalization, Product, ProductAudio, ProductBundle, ProductVideo, ReleasePricing, Store, TerritoryCodeA2, TerritoryDates, Track, TrackReleasePricing, VideoArtworkAsset, VideoAsset, ) from tests.unit.conftest import ( get_art_relations_store_customer_mock, get_art_relations_tracks_mock, get_delivery_metadata_mock_by_store, get_dms_encoding_profile_mock, get_store_mock, update_mock_object, ) def test_get_feature_to_primary_names() -> None: assert delivery_metadata_model.get_feature_to_primary_names( [ ArtRelationsArtist( fullname="John Doe", role=ParticipantRole.FEATURING, is_feature_to_primary=True, ), ArtRelationsArtist(fullname="John Doe", role=ParticipantRole.REMIXER), ArtRelationsArtist(fullname="Jane Smith", role=ParticipantRole.ENSEMBLE), ArtRelationsArtist(fullname="The Band", role=ParticipantRole.ORCHESTRA), ] ) == {"johndoe"} @pytest.mark.parametrize( ( "test_description", "track_id", "expected_result", ), [ ( "missing track writer for track", 99999, [], ), ( "track writer exists for track", 12347, ["Jimmy Alpine"], ), ], ) def test_get_track_writers( test_description: str, track_id: int, expected_result: list[str], art_relations_track_writers_mock: dict[int, list[ArtRelationsTrackWriter]], ) -> None: result = delivery_metadata_model.get_track_writers( art_relations_track_writers_mock, track_id ) assert result == expected_result @pytest.mark.parametrize( ( "test_description", "track_id", "expected_result", ), [ ( "missing track publisher for track", 99999, [], ), ( "track publisher exists for track", 12347, ["Fire Pubs", "Pubbz House"], ), ], ) def test_get_track_publishers( test_description: str, track_id: int, expected_result: list[str], art_relations_track_publisher_mock: dict[int, list[ArtRelationsTrackPublisher]], ) -> None: result = delivery_metadata_model.get_track_publishers( art_relations_track_publisher_mock, track_id ) assert result == expected_result @pytest.mark.asyncio @pytest.mark.parametrize( ( "store_id", "territories_response_mock", "genre_mappings", "store_overrides", "ar_store_overrides", "track_types", ), [ pytest.param( StoreIds.NO_STORE, [TerritoryCodeA2(country_code=x) for x in ["FR", "CA"]], None, {}, {}, [TrackTypes.MUSIC], id="Test generic music product", ), pytest.param( StoreIds.NO_STORE, [TerritoryCodeA2(country_code=x) for x in ["FR", "CA"]], None, {"is_localization_supported": True}, {"supports_localization": "Y"}, [TrackTypes.VIDEO], id="Test video product with localization", ), pytest.param( StoreIds.AMAZON, [TerritoryCodeA2(country_code=x) for x in ["FR", "CA"]], None, {}, {}, [TrackTypes.MUSIC], id="Test music product for Amazon", ), ], ) async def test_get_delivery_metadata( store_id: int, territories_response_mock: list[TerritoryCodeA2], genre_mappings: list[GenreMapping], store_overrides: dict[str, Any], ar_store_overrides: dict[str, Any], track_types: list[TrackTypes], art_relations_track_artists_mock: dict[int, list[ArtRelationsArtist]], art_relations_release_artists_mock: list[ArtRelationsArtist], art_relations_product_provided_store_artists_mock: ArtRelationsProductProvidedStoreArtists, art_relations_product_mock: ArtRelationsProduct, art_relations_product_video_mock: ArtRelationsProductVideo, art_relations_tracks_mock: list[ArtRelationsTrack], art_relations_track_writers_mock: dict[int, list[ArtRelationsTrackWriter]], art_relations_track_publisher_mock: dict[int, list[ArtRelationsTrackPublisher]], graphql_product_mock: GraphQlProduct, timed_release_client_response_mock: TimedRelease, distribution_features_mock: dict[int, list[DistributionFeatures]], art_relations_territory_dates_mock: list[TerritoryDates], spotify_genre_mapping_mock: list[GenreMapping], instant_grats_client_response_mock: dict[int, set[InstantGrats]], product_pricing_response_mock: set[ReleasePricing], track_pricing_response_mock: set[TrackReleasePricing], video_track_pricing_response_mock: set[TrackReleasePricing], get_product_localized_metadata_result_mock: list[ ArtRelationsProductLocalizedMetadata ], get_track_localized_metadata_result_mock: list[ArtRelationsTrackLocalizedMetadata], get_product_artist_localized_metadata_result_mock: list[ ParticipantFullnameLocalization ], get_track_artist_localized_metadata_result_mock: list[ ParticipantFullnameLocalization ], get_track_performers_with_role_names_response_mock: dict[int, list[TrackPerformer]], mocker: MockerFixture, product_assets_mock: dict[ str, list[ArtworkAsset | DigitalBookletAsset | VideoArtworkAsset] | dict[int, list[AudioAsset | VideoAsset]], ], ) -> None: """Test get_delivery_metadata.""" mocker.patch( "delivery_metadata.models.delivery_metadata.USE_MERGED_TIMED_RELEASES_LOGIC", False, # Force legacy path ) store = update_mock_object( get_store_mock(store_id, track_types), Store, store_overrides, ) delivery_metadata_mock = get_delivery_metadata_mock_by_store(store, track_types) updates = { "delivery.allowed_territories": territories_response_mock, "product.genre_mappings": genre_mappings, } updated_delivery_metadata_mock = update_mock_object( delivery_metadata_mock, DeliveryMetadata, updates ) get_art_relations_product_mock = mocker.patch( "delivery_metadata.clients.art_relations.get_product", return_value=art_relations_product_mock, autospec=True, ) get_art_relations_product_video_mock = mocker.patch( "delivery_metadata.clients.art_relations.get_product_video", return_value=art_relations_product_video_mock, autospec=True, ) get_tracks_mock = mocker.patch( "delivery_metadata.clients.art_relations.get_tracks", return_value=get_art_relations_tracks_mock(track_types), autospec=True, ) get_art_relations_track_artists_mock = mocker.patch( "delivery_metadata.clients.art_relations.get_track_artists", return_value=art_relations_track_artists_mock, autospec=True, ) get_art_relations_release_artists_mock = mocker.patch( "delivery_metadata.clients.art_relations.get_release_artists", return_value=art_relations_release_artists_mock, autospec=True, ) get_art_relations_territory_dates_mock = mocker.patch( "delivery_metadata.clients.art_relations.get_territory_dates", return_value=art_relations_territory_dates_mock, autospec=True, ) get_art_relations_product_provided_store_artists_mock = mocker.patch( "delivery_metadata.clients.art_relations.get_product_provided_store_artists", return_value=art_relations_product_provided_store_artists_mock, autospec=True, ) get_ar_store_customer_mock = mocker.patch( "delivery_metadata.clients.art_relations.get_store_customer", return_value=update_mock_object( get_art_relations_store_customer_mock(store_id), ArtRelationsStoreCustomer, ar_store_overrides, ), autospec=True, ) get_art_relations_track_writers_mock = mocker.patch( "delivery_metadata.clients.art_relations.get_track_writers", return_value=art_relations_track_writers_mock, autospec=True, ) _get_track_writers_spy = mocker.spy(delivery_metadata_model, "get_track_writers") get_art_relations_track_publishers_mock = mocker.patch( "delivery_metadata.clients.art_relations.get_track_publishers", return_value=art_relations_track_publisher_mock, autospec=True, ) _get_track_publishers_spy = mocker.spy( delivery_metadata_model, "get_track_publishers" ) get_dms_encoding_profiles_mock = mocker.patch( "delivery_metadata.clients.direct_delivery.get_dms_encoding_profiles", return_value=get_dms_encoding_profile_mock(track_types, store), autospec=True, ) get_graphql_product_mock = mocker.patch( "delivery_metadata.clients.graphql.get_product", return_value=graphql_product_mock, autospec=True, ) get_genre_mapping_mock = mocker.patch( "delivery_metadata.clients.art_relations.get_genre_mapping", return_value=None, autospec=True, ) get_timed_release_mock = mocker.patch( "delivery_metadata.clients.ows_timed_release.get_timed_release", return_value=timed_release_client_response_mock, autospec=True, ) get_product_timed_release_by_store_id_mock = mocker.patch( "delivery_metadata.clients.ows_timed_release.get_product_timed_release_by_store_id", autospec=True, ) get_territory_rights_mock = mocker.patch( "delivery_metadata.clients.ows_carveouts.get_product_territory_rights", return_value=territories_response_mock, autospec=True, ) get_territory_conversion_mock = mocker.patch( "delivery_metadata.models.delivery_metadata.get_converted_territories", return_value=territories_response_mock, autospec=True, ) get_release_pricing_mock = mocker.patch( "delivery_metadata.clients.ows_pricing.get_product_pricing", return_value=product_pricing_response_mock, autospec=True, ) get_track_pricing_mock = mocker.patch( "delivery_metadata.clients.ows_pricing.get_track_pricing", return_value=track_pricing_response_mock, autospec=True, ) get_instant_grats_mock = mocker.patch( "delivery_metadata.clients.ows_track.get_instant_grats", return_value=instant_grats_client_response_mock, autospec=True, ) get_distribution_features_mock = mocker.patch( "delivery_metadata.clients.art_relations.get_distribution_features", return_value=distribution_features_mock.get(store_id, []), autospec=True, ) get_product_localized_metadata_mock = mocker.patch( "delivery_metadata.clients.art_relations.get_product_localized_metadata", return_value=get_product_localized_metadata_result_mock, autospec=True, ) get_track_localized_metadata_mock = mocker.patch( "delivery_metadata.clients.art_relations.get_track_localized_metadata", return_value=get_track_localized_metadata_result_mock, autospec=True, ) get_product_artist_localized_metadata_mock = mocker.patch( "delivery_metadata.clients.art_relations.get_product_artist_localized_metadata", return_value=get_product_artist_localized_metadata_result_mock, autospec=True, ) get_track_artist_localized_metadata_mock = mocker.patch( "delivery_metadata.clients.art_relations.get_track_artist_localized_metadata", return_value=get_track_artist_localized_metadata_result_mock, autospec=True, ) get_track_performers_with_role_names_mock = mocker.patch( "delivery_metadata.clients.ows_track.get_track_performers_with_role_names", return_value=get_track_performers_with_role_names_response_mock, autospec=True, ) get_assets_mock = mocker.patch( "delivery_metadata.clients.ows_assets.get_product_assets", return_value=product_assets_mock, autospec=True, ) get_video_assets_mock = mocker.patch( "delivery_metadata.clients.ows_video.get_product_video_assets", return_value={ "product_assets": product_assets_mock["product_assets"], "track_assets": {}, }, autospec=True, ) _hydrate_track_spy = mocker.spy(delivery_metadata_model, "_hydrate_track") upc = 75679660923 product_id = 4953147 delivery_metadata = await delivery_metadata_model.get_delivery_metadata( upc=upc, store_id=store_id, delivery_type=DeliveryType.COMPLETE_ALBUM, ) get_art_relations_product_mock.assert_called_once_with(upc) get_tracks_mock.assert_called_once_with(upc) get_art_relations_product_provided_store_artists_mock.assert_called_once_with( product_id ) get_ar_store_customer_mock.assert_called_once_with(store_id) get_dms_encoding_profiles_mock.assert_called_once_with( store_id, ProductTypeId.MUSIC ) get_art_relations_track_writers_mock.assert_called_once_with(upc) get_art_relations_track_publishers_mock.assert_called_once_with(upc) if track_types == [TrackTypes.VIDEO]: get_art_relations_release_artists_mock.assert_not_called() get_art_relations_track_artists_mock.assert_not_called() get_art_relations_product_video_mock.assert_called_once_with(upc) else: get_art_relations_release_artists_mock.assert_called_once_with(product_id) get_art_relations_track_artists_mock.assert_called_once_with(upc) assert _get_track_writers_spy.mock_calls == [ mocker.call(art_relations_track_writers_mock, 12345), mocker.call(art_relations_track_writers_mock, 12347), mocker.call(art_relations_track_writers_mock, 123410), mocker.call(art_relations_track_writers_mock, 12348), mocker.call(art_relations_track_writers_mock, 12349), mocker.call(art_relations_track_writers_mock, 123412), ] assert _get_track_publishers_spy.mock_calls == [ mocker.call(art_relations_track_publisher_mock, 12345), mocker.call(art_relations_track_publisher_mock, 12347), mocker.call(art_relations_track_publisher_mock, 123410), mocker.call(art_relations_track_publisher_mock, 12348), mocker.call(art_relations_track_publisher_mock, 12349), mocker.call(art_relations_track_publisher_mock, 123412), ] get_graphql_product_mock.assert_called_once_with(product_id) get_genre_mapping_mock.assert_called_once_with(store_id) get_timed_release_mock.assert_called_once_with(product_id) get_product_timed_release_by_store_id_mock.assert_not_called() get_territory_rights_mock.assert_called_once_with(upc, store_id) get_territory_conversion_mock.assert_called_once_with( territories_response_mock, updated_delivery_metadata_mock.delivery.store.territory_standard, ) if store.is_pricing_supported: get_release_pricing_mock.assert_called_once_with(product_id, store_id) if track_types == [TrackTypes.MUSIC]: get_track_pricing_mock.assert_called_once_with(product_id, store_id) if track_types == [TrackTypes.VIDEO]: get_track_pricing_mock.assert_called_once_with(product_id, store_id, True) else: get_release_pricing_mock.assert_not_called() get_track_pricing_mock.assert_not_called() if store.is_instant_gratification_supported: get_instant_grats_mock.assert_called_once_with(product_id) if store.is_localization_supported: get_product_localized_metadata_mock.assert_called_once_with(product_id) get_track_localized_metadata_mock.assert_called_once_with(product_id) if track_types == [TrackTypes.MUSIC]: get_product_artist_localized_metadata_mock.assert_called_once_with( product_id ) get_track_artist_localized_metadata_mock.assert_called_once_with(product_id) else: get_product_artist_localized_metadata_mock.assert_not_called() get_track_artist_localized_metadata_mock.assert_not_called() get_track_performers_with_role_names_mock.assert_called_once_with(product_id) get_distribution_features_mock.assert_called_once_with(store_id) get_art_relations_territory_dates_mock.assert_called_once_with(upc) if TrackTypes.MUSIC in track_types: get_assets_mock.assert_called_once_with(product_id) else: get_assets_mock.assert_not_called() if TrackTypes.VIDEO in track_types: get_video_assets_mock.assert_called_once_with(product_id) else: get_video_assets_mock.assert_not_called() compare( expected=updated_delivery_metadata_mock, actual=delivery_metadata, recursive=True, strict=True, raises=True, ) match track_types: case [TrackTypes.MUSIC]: assert isinstance(updated_delivery_metadata_mock.product, ProductAudio) case [TrackTypes.VIDEO]: assert isinstance(updated_delivery_metadata_mock.product, ProductVideo) case [TrackTypes.MUSIC, TrackTypes.VIDEO]: assert isinstance(updated_delivery_metadata_mock.product, ProductBundle) # TODO - exception (each role), no exception... @pytest.mark.parametrize( ( "participants", "expect_exception", ), [ ( [ Participant( fullname="Label Artist", roles={ParticipantRole.PERFORMER}, instruments=set(), label_participant_id=None, ) ], True, ), ( [ Participant( fullname="Label Artist", roles={ParticipantRole.FEATURING}, instruments=set(), label_participant_id=None, ) ], True, ), ( [ Participant( fullname="Label Artist", roles={ParticipantRole.REMIXER}, instruments=set(), label_participant_id=None, ) ], False, ), ( [ Participant( fullname="Label Artist", roles={ParticipantRole.PERFORMER}, instruments=set(), label_participant_id=uuid.UUID( "d287337c-c7f1-4660-9d0c-e82758c10183" ), ) ], False, ), ], ) def test_check_label_participant_requirements( participants: list[Participant], expect_exception: bool ) -> None: raised = False try: delivery_metadata_model.check_label_participant_requirements( participants=participants, ) except NoLabelParticipantForParticipant: raised = True assert raised == expect_exception def test_get_participants_for_product(snapshot: SnapshotAssertion) -> None: art_relations_artists = [ ArtRelationsArtist( fullname="Very Big Star", role=ParticipantRole.PERFORMER, ), ArtRelationsArtist(fullname="Nobody Special", role=ParticipantRole.ENSEMBLE), ArtRelationsArtist(fullname="nobody special", role=ParticipantRole.ORCHESTRA), ] participant_fullname_localizations = [ ParticipantFullnameLocalization( fullname="very big star", language_code="fr", localized_fullname="Très Grande Étoile", ) ] label_participant_map = LabelParticipantMap( map={ "verybigstar": LabelParticipantData( label_participant_id=uuid.UUID("d287337c-c7f1-4660-9d0c-e82758c10183"), apple_music_id="apple123", spotify_id="spotify123", spotify_artist_key="spotify_artist_key", global_participant_id=uuid.UUID("fed7a0bf-6e0d-4b41-bd4b-e1d4ee6dc401"), ) } ) result = delivery_metadata_model.get_product_participants( art_relations_artists=art_relations_artists, participant_fullname_localizations=frozenset( participant_fullname_localizations ), lp_map=label_participant_map, is_feature_to_primary_supported=False, ) assert sorted(result, key=lambda x: x.fullname) == snapshot def test_get_participants_for_track(snapshot: SnapshotAssertion) -> None: art_relations_artists = [ ArtRelationsArtist( fullname="Very Big Star", role=ParticipantRole.PERFORMER, ), ArtRelationsArtist(fullname="Nobody Special", role=ParticipantRole.ENSEMBLE), ] track_performers = [ TrackPerformer( role="Vocals - Lead Vocals", birth_name="Very Big Star", ), TrackPerformer( role="Percussion - Congas", birth_name="Nobody Special", ), TrackPerformer( role="Vocals - Spoken Word", birth_name="Speaker", ), TrackPerformer( role="Vocals - Backing Vocals", birth_name="Background Vocalist", ), ] lsr_participations = [ LabelSoundRecordingParticipation( role=Role(name=ParticipantRole.VOICE_ACTOR.value), participant=LabelParticipant( uuid=uuid.UUID("2567e78e-7650-43cd-aec2-361e2d6c2bd2"), name="Neutron Star", appleMusicId="apple001", spotifyId="spotify001", spotifyArtistKey="spotify_artist_key", globalParticipant=GlobalParticipant( id=uuid.UUID("29c3e618-f1a5-4692-a257-32ede48cf3de") ), ), ), LabelSoundRecordingParticipation( role=Role(name=ParticipantRole.AUTHOR_OF_AFTERWORD.value), participant=LabelParticipant( uuid=uuid.UUID("113911e3-171c-4660-b0b5-35ae2ee4b8a1"), name="very Big star", appleMusicId="apple456", spotifyId="spotify456", spotifyArtistKey="spotify_artist_key", globalParticipant=GlobalParticipant( id=uuid.UUID("c3e972a5-dc8d-447e-9535-d9e618b59f51") ), ), ), ] label_participant_map = LabelParticipantMap( map={ "verybigstar": LabelParticipantData( label_participant_id=uuid.UUID("d287337c-c7f1-4660-9d0c-e82758c10183"), apple_music_id="apple123", spotify_id="spotify123", spotify_artist_key="spotify_artist_key", global_participant_id=uuid.UUID("fed7a0bf-6e0d-4b41-bd4b-e1d4ee6dc401"), ), "neutronstar": LabelParticipantData( label_participant_id=uuid.UUID("2567e78e-7650-43cd-aec2-361e2d6c2bd1"), apple_music_id="apple001", spotify_id="spotify001", spotify_artist_key="spotify_artist_key", global_participant_id=uuid.UUID("29c3e618-f1a5-4692-a257-32ede48cf3dd"), ), } ) result = delivery_metadata_model.get_track_participants( art_relations_artists=art_relations_artists, track_performers=track_performers, lsr_participations=lsr_participations, participant_fullname_localizations=frozenset(), lp_map=label_participant_map, is_feature_to_primary_supported=False, ) assert sorted(result, key=lambda x: x.fullname) == snapshot @pytest.mark.asyncio @pytest.mark.parametrize( ( "_description", "use_merged_logic", "mock_scenario", ), [ ( "Legacy endpoint - uses get_timed_release()", False, # Legacy path "legacy_success", ), ( "Merged endpoint - successful response", True, # Merged path - success "merged_success", ), ( "Merged endpoint - handles 404 correctly", True, # Merged path - 404 "merged_404", ), ], ) async def test__fetch_timed_release_or_none( _description: str, use_merged_logic: bool, mock_scenario: str, timed_release_client_response_mock: TimedRelease, product_timed_release_by_store_client_response_mock: ProductTimedReleaseByStore, mocker: MockerFixture, ) -> None: """Test get_delivery_metadata with different timed-release scenarios.""" # Force merged logic path mocker.patch( "delivery_metadata.models.delivery_metadata.USE_MERGED_TIMED_RELEASES_LOGIC", use_merged_logic, ) store_id = 286 product_id = 4953147 # Set up timed-release mocks based on scenarios get_timed_release_mock = mocker.patch( "delivery_metadata.models.delivery_metadata.ows_timed_release.get_timed_release", autospec=True, ) get_product_timed_release_by_store_id_mock = mocker.patch( "delivery_metadata.models.delivery_metadata.ows_timed_release.get_product_timed_release_by_store_id", autospec=True, ) if mock_scenario == "legacy_success": get_timed_release_mock.return_value = timed_release_client_response_mock elif mock_scenario == "merged_success": get_product_timed_release_by_store_id_mock.return_value = ( product_timed_release_by_store_client_response_mock ) elif mock_scenario == "merged_404": get_product_timed_release_by_store_id_mock.return_value = None mocker.patch( "delivery_metadata.clients.ows_assets.get_product_assets", return_value={"product_assets": [], "track_assets": {}}, autospec=True, ) _ = await delivery_metadata_model._fetch_timed_release_or_none(product_id, store_id) if use_merged_logic: get_product_timed_release_by_store_id_mock.assert_called_once_with( product_id, store_id ) get_timed_release_mock.assert_not_called() else: get_timed_release_mock.assert_called_once_with(product_id) get_product_timed_release_by_store_id_mock.assert_not_called() @pytest.mark.asyncio async def test_get_converted_territories( mocker: MockerFixture, ows_territories_response_mock: TerritoryResponse ) -> None: input_codes = [TerritoryCodeA2(country_code=x) for x in ["SD", "CA"]] get_territory_conversion_mock = mocker.patch( "delivery_metadata.clients.ows_territories.convert_territories", return_value=ows_territories_response_mock, autospec=True, ) result = await delivery_metadata_model.get_converted_territories( input_codes, TerritoryStandard.ISO_3166_2016 ) get_territory_conversion_mock.assert_called_once_with( TerritoryStandard.ORCHARD_2016, TerritoryStandard.ISO_3166_2016, input_codes ) assert result == [TerritoryCodeA2(country_code="US")] @pytest.mark.asyncio async def test_get_converted_territories_empty( mocker: MockerFixture, ows_territories_response_mock: TerritoryResponse ) -> None: get_territory_conversion_mock = mocker.patch( "delivery_metadata.clients.ows_territories.convert_territories", return_value=ows_territories_response_mock, autospec=True, ) result = await delivery_metadata_model.get_converted_territories( [], TerritoryStandard.ISO_3166_2016 ) assert not get_territory_conversion_mock.called assert result == [] def test_get_distribution_feature_ids( distribution_features_mock: dict[int, list[DistributionFeatures]], snapshot: SnapshotAssertion, ) -> None: result = delivery_metadata_model.get_distribution_feature_ids( distribution_features_mock.get(StoreIds.NO_STORE, []) ) assert result == snapshot @pytest.mark.parametrize( ("extras", "expected_result"), [ (None, {UseType.STREAM, UseType.ON_DEMAND, UseType.NON_INTERACTIVE}), ( { DownloadStreamRights.STREAM: { CommercialModelType.SUBSCRIPTION: { UseType.NON_INTERACTIVE, UseType.ON_DEMAND, } }, }, {UseType.STREAM, UseType.NON_INTERACTIVE, UseType.ON_DEMAND}, ), ], ) def test__get_commercial_model_type_to_use_types_map( extras: dict[DownloadStreamRights, dict[CommercialModelType, set[UseType]]], expected_result: set[UseType], delivery_metadata_mock: DeliveryMetadata, ) -> None: updated_delivery_metadata_mock = delivery_metadata_mock if extras: updated_delivery_metadata_mock = update_mock_object( delivery_metadata_mock, DeliveryMetadata, { "delivery.store.ddex_commercial_model_to_use_type_map_overrides": extras, }, ) assert ( updated_delivery_metadata_mock.delivery.store.ddex_commercial_model_to_use_type_map == { DownloadStreamRights.STREAM: { CommercialModelType.AD_SUPPORTED: { UseType.STREAM, }, CommercialModelType.SUBSCRIPTION: expected_result, }, DownloadStreamRights.DOWNLOAD: { CommercialModelType.PAY: { UseType.PERMANENT_DOWNLOAD, UseType.RINGTONE, }, CommercialModelType.SUBSCRIPTION: { UseType.CONDITIONAL_DOWNLOAD, }, }, } ) def test_store_overrides(delivery_metadata_mock: DeliveryMetadata) -> None: """Test that all store setting overrides are valid.""" for _, values in store_setting_overrides.items(): params = dict(delivery_metadata_mock.delivery.store) | values Store(**params) def test_delivery_validate_model(delivery_metadata_mock: DeliveryMetadata) -> None: with pytest.raises(ProductIneligible) as e: Delivery( **{ **delivery_metadata_mock.delivery.model_dump(), "allowed_territories": [], } ) assert ( str(e.value) == "allowed_territories cannot be empty for delivery_type: complete_album" ) def test_promote_or_demote_feature_to_primary_roles( snapshot: SnapshotAssertion, ) -> None: """Test processing of feature_to_primary roles.""" participants = [ Participant( fullname="Very Big Artist", roles={ParticipantRole.FEATURING, ParticipantRole.REMIXER}, instruments={InstrumentRole.VOCALS_YODELING}, ), Participant( fullname="Neutron Star", roles={ParticipantRole.FEATURING, ParticipantRole.ADAPTER}, instruments={InstrumentRole.BRASS_BUGLE}, ), Participant( fullname="Background Vocalist", roles={ParticipantRole.BACKGROUND_VOCALIST}, instruments=set(), ), ] results = delivery_metadata_model.handle_feature_to_primary_role( participants, {"verybigartist", "backgroundvocalist"} ) assert results == snapshot def test_get_placeholder_replacements_product( delivery_metadata_mock: DeliveryMetadata, ) -> None: result = delivery_metadata_mock._get_placeholder_replacements() assert result == { "content_id": 3452323, "upc": 75679660923, "display_upc": "075679660923", "vendor_id": 7123, } def test_get_placeholder_replacements_track( delivery_metadata_mock: DeliveryMetadata, ) -> None: result = delivery_metadata_mock._get_placeholder_replacements( delivery_metadata_mock.product.first_track ) assert result == { "content_id": 3452323, "upc": 75679660923, "display_upc": "075679660923", "cd": 1, "track_id": 1, "isrc": "USAT22403171", "vendor_id": 7123, } @pytest.mark.parametrize( ("store_id", "expected_time"), [ (StoreIds.SPOTIFY, "2024-08-14T09:30:00Z"), (StoreIds.NO_STORE, None), ], ) def test_get_timed_release_by_dms( store_id: int, expected_time: str, timed_release_client_response_mock: TimedRelease ) -> None: actual_time = delivery_metadata_model.get_timed_release_by_dms( store_id, timed_release_client_response_mock ) assert actual_time == expected_time @pytest.mark.parametrize( ("store_id", "expected_time"), [ (1942, "2024-01-01"), (StoreIds.NO_STORE, None), ], ) def test_get_staggered_release_by_dms( store_id: int, expected_time: str, timed_release_client_response_mock: TimedRelease ) -> None: result = delivery_metadata_model.get_staggered_release_by_dms( store_id, timed_release_client_response_mock ) assert result == expected_time def test_get_timed_release_by_dms_no_timed_data() -> None: assert not delivery_metadata_model.get_timed_release_by_dms( 0, TimedRelease(staggered=None, timed=None, unsupported_timed=None) ) def test_get_staggered_release_by_dms_no_timed_data() -> None: assert not delivery_metadata_model.get_staggered_release_by_dms( 0, TimedRelease(staggered=None, timed=None, unsupported_timed=None) ) def test_product_distro_rights_for_non_distro_store( delivery_metadata_mock: DeliveryMetadata, ) -> None: updated_delivery_metadata_mock = update_mock_object( delivery_metadata_mock, DeliveryMetadata, {"delivery.store.distribution_feature_ids": set()}, ) distro_rights = delivery_metadata_model._get_release_distribution_rights( delivery_metadata_mock.product.tracks, updated_delivery_metadata_mock.delivery.store, DeliveryType.METADATA_UPDATE, ) assert distro_rights == {DownloadStreamRights.DOWNLOAD, DownloadStreamRights.STREAM} def test_track_distro_rights_for_non_distro_store( delivery_metadata_mock: DeliveryMetadata, art_relations_tracks_mock: list[ArtRelationsTrack], ) -> None: updated_delivery_metadata_mock = update_mock_object( delivery_metadata_mock, DeliveryMetadata, {"delivery.store.distribution_feature_ids": set()}, ) track_data = art_relations_tracks_mock[0].__dict__ track_data["offer_type"] = TrackOfferType.NONE distro_rights = delivery_metadata_model._get_track_distribution_rights( ArtRelationsTrack(**track_data), updated_delivery_metadata_mock.delivery.store, ) assert distro_rights == {DownloadStreamRights.DOWNLOAD, DownloadStreamRights.STREAM} @pytest.mark.parametrize( ( "_test_description", "all_tracks", "track_offer_type", "product_distribution_rights", "track_distribution_rights", "raises_context", ), [ ( "all tracks have none offer type", True, TrackOfferType.NONE, set(), set(), pytest.raises(ProductIneligible), ), ( "all tracks have album_download_only offer type", True, TrackOfferType.ALBUM_DOWNLOAD_ONLY, {DownloadStreamRights.DOWNLOAD}, set(), nullcontext(), ), ( "some tracks have album_download_only offer type", False, TrackOfferType.ALBUM_TRACK_DOWNLOAD, {DownloadStreamRights.DOWNLOAD}, {DownloadStreamRights.DOWNLOAD}, nullcontext(), ), ], ) def test_product_distro_rights( _test_description: str, all_tracks: bool, track_offer_type: TrackOfferType, product_distribution_rights: set[DownloadStreamRights], track_distribution_rights: set[DownloadStreamRights], raises_context: ContextManager[Exception], delivery_metadata_mock: DeliveryMetadata, ) -> None: updated_tracks: list[Track] = [] for track in get_art_relations_tracks_mock([TrackTypes.MUSIC]): new_track_dict = track.__dict__ new_track_dict["offer_type"] = track_offer_type updated_tracks.append( _hydrate_track( ArtRelationsTrack(**new_track_dict), delivery_metadata_mock.delivery.store, [], {}, {}, [], [], ) ) if not all_tracks: break product_data = delivery_metadata_mock.product.__dict__ product_data["tracks"] = updated_tracks updated_product = Product(**product_data) updated_delivery_metadata_mock = update_mock_object( delivery_metadata_mock, DeliveryMetadata, {"delivery.product": updated_product}, ) for new_track in updated_tracks: assert new_track.distribution_rights == track_distribution_rights with raises_context as exc: distro_rights = delivery_metadata_model._get_release_distribution_rights( updated_delivery_metadata_mock.product.tracks, updated_delivery_metadata_mock.delivery.store, DeliveryType.COMPLETE_ALBUM, ) assert distro_rights == product_distribution_rights if exc: assert isinstance(exc, ProductIneligible) def test_hydrate_track_instant_grats_for_other_store( art_relations_tracks_mock: list[ArtRelationsTrack], ) -> None: store = update_mock_object( get_store_mock(StoreIds.AMAZON, [TrackTypes.MUSIC]), Store, {"is_pricing_supported": False}, ) # instant grats data exists but is keyed to Apple — not to Amazon (the requested store) instant_grats_for_other_store: dict[int, set[InstantGrats]] = { StoreIds.APPLE: { InstantGrats( tuid=12345, store_id=StoreIds.APPLE, date="2024-08-10T00:00:00", active="Y", ), } } track = _hydrate_track( art_relations_tracks_mock[0], store, [], {}, instant_grats_for_other_store, [], [], ) assert track.instant_grats is None @pytest.mark.parametrize( ("_test_description", "distro_features", "offer_types", "expected_result"), [ ( "track_download_only and album track download - download only rights", {DistributionFeatureId.A_LA_CARTE_DOWNLOAD}, [TrackOfferType.TRACK_DOWNLOAD_ONLY, TrackOfferType.ALBUM_TRACK_DOWNLOAD], {DownloadStreamRights.DOWNLOAD}, ), ( "track_download_only and album_track_download - streaming no rights", {DistributionFeatureId.SUBSCRIPTION_STREAMING}, [TrackOfferType.TRACK_DOWNLOAD_ONLY, TrackOfferType.ALBUM_TRACK_DOWNLOAD], set(), ), ( "album_download_stream - download no rights", {DistributionFeatureId.A_LA_CARTE_DOWNLOAD}, [TrackOfferType.ALBUM_DOWNLOAD_STREAM], set(), ), ( "album_download_stream - streaming only rights", {DistributionFeatureId.SUBSCRIPTION_STREAMING}, [TrackOfferType.ALBUM_DOWNLOAD_STREAM], {DownloadStreamRights.STREAM}, ), ( "album_download_only and none - no rights", { DistributionFeatureId.A_LA_CARTE_DOWNLOAD, DistributionFeatureId.SUBSCRIPTION_STREAMING, }, [TrackOfferType.ALBUM_DOWNLOAD_ONLY, TrackOfferType.NONE], set(), ), ], ) def test_track_distro_rights( _test_description: str, distro_features: set[DistributionFeatureId], offer_types: list[TrackOfferType], expected_result: set[DownloadStreamRights], delivery_metadata_mock: DeliveryMetadata, art_relations_tracks_mock: list[ArtRelationsTrack], ) -> None: updated_delivery_metadata_mock = update_mock_object( delivery_metadata_mock, DeliveryMetadata, {"delivery.store.distribution_feature_ids": distro_features}, ) for offer_type in offer_types: track_data = art_relations_tracks_mock[0].__dict__ track_data["offer_type"] = offer_type distro_rights = delivery_metadata_model._get_track_distribution_rights( ArtRelationsTrack(**track_data), updated_delivery_metadata_mock.delivery.store, ) assert distro_rights == expected_result def test_get_label_participant_map( graphql_json_data: str, snapshot: SnapshotAssertion, ) -> None: graphql_data = json.loads(graphql_json_data) result = delivery_metadata_model.get_label_participant_map( GraphQlProduct(**graphql_data) ) assert result.map == snapshot def test_group_track_pricing_by_track_id_audio( track_pricing_response_mock: set[TrackReleasePricing], audio_track_pricing_dict_mock: dict[int, set[ReleasePricing]], ) -> None: result = delivery_metadata_model.group_track_pricing_by_track_id( track_pricing_response_mock, ) assert result == audio_track_pricing_dict_mock def test_group_track_pricing_by_track_id_video( video_track_pricing_response_mock: set[TrackReleasePricing], video_track_pricing_dict_mock: dict[int, set[ReleasePricing]], ) -> None: result = delivery_metadata_model.group_track_pricing_by_track_id( video_track_pricing_response_mock, ) assert result == video_track_pricing_dict_mock @pytest.mark.parametrize( "track_id", [12345, 12347, 12348], ) def test_get_track_pricing_by_track_id( track_id: int, audio_track_pricing_dict_mock: dict[int, set[ReleasePricing]], snapshot: SnapshotAssertion, ) -> None: result = delivery_metadata_model.get_track_pricing_by_track_id( track_id, audio_track_pricing_dict_mock, ) assert result == snapshot @pytest.mark.parametrize( "track_id", [11111, 22222, 33333], ) def test_get_track_pricing_by_track_id_exception( track_id: int, audio_track_pricing_dict_mock: dict[int, set[ReleasePricing]], ) -> None: with pytest.raises( PricingDataNotFound, match=f"^No track pricing found for track {track_id}.$" ): delivery_metadata_model.get_track_pricing_by_track_id( track_id, audio_track_pricing_dict_mock, ) @pytest.mark.parametrize( "preorder_mock, timed_release_mock, expected_result", [ ("2025-07-04", "2025-07-07T00:00:00Z", "2025-07-04"), ("2025-07-04", "2025-07-05T00:00:00Z", None), ("2025-07-04", "2025-07-06T00:00:00Z", None), ("2025-07-04", None, "2025-07-04"), (None, "2025-08-24T00:00:00Z", None), ], ) def test_get_preorder_date_or_none( preorder_mock: str | None, timed_release_mock: str | None, expected_result: str | None, ) -> None: result = delivery_metadata_model.get_preorder_date_or_none( preorder_mock, timed_release_mock, ) assert result == expected_result