"""Tests for contributor logic layer.""" from unittest.mock import MagicMock from uuid import UUID import pytest from fastapi import HTTPException from pytest_mock import MockerFixture from contributor.api.schemas.contributors import Contributor, Label from contributor.logic.contributors import ( assert_authorization, check_vendor_access, get_contributor, get_contributors, get_contributors_for_global_participant, get_global_participant, get_global_participants, get_product_contributors, get_product_participations, get_project_contributors, get_projects, get_sound_recording_participations, get_track_participations, get_track_participations_by_track_ids, get_track_participations_for_track, merge_contributors, rename_contributor, update_contributor, ) @pytest.fixture def contributor_uuid(): return "550e8400-e29b-41d4-a716-446655440000" @pytest.fixture def contributor(contributor_uuid): return { "id": 42, "uuid": contributor_uuid, "name": "Test Artist", "spotify_id": "3TVXtAsR1Inumwj472S9r4", "apple_music_id": None, "spotify_artist_key": None, "isni": None, "company_brand_uuid": None, "parent_company_uuid": None, "artist_info_ids": [], "global_participant": None, "label": { "uuid": "cccc0000-dddd-4eee-8fff-aaaaaaaaaaaa", "vendor_id": 100, "subaccount_id": 0, }, } @pytest.fixture def session(): return MagicMock() @pytest.fixture def label_profiles(): return [{"profile_type": "LabelProfile", "profile_id": 7123, "roles": []}] @pytest.fixture def admin_profiles(): return [{"profile_type": "ContentApp", "profile_id": 99, "roles": []}] class TestCheckVendorAccess: def test_returns_true_when_authorized( self, mocker: MockerFixture, session, label_profiles ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mock = mocker.patch( "contributor.logic.contributors.neo4j_queries.check_vendor_access", return_value=True, ) result = check_vendor_access(session, profiles=label_profiles, vendor_id=100) mock.assert_called_once_with(session, profile_ids=[7123], vendor_id=100) assert result is True def test_returns_false_when_not_authorized( self, mocker: MockerFixture, session, label_profiles ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mocker.patch( "contributor.logic.contributors.neo4j_queries.check_vendor_access", return_value=False, ) assert ( check_vendor_access(session, profiles=label_profiles, vendor_id=100) is False ) def test_returns_true_for_full_catalog_access_without_profile_check( self, mocker: MockerFixture, session, admin_profiles ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=True, ) vendor_check = mocker.patch( "contributor.logic.contributors.neo4j_queries.check_vendor_access" ) result = check_vendor_access(session, profiles=admin_profiles, vendor_id=100) vendor_check.assert_not_called() assert result is True class TestAssertAuthorization: def test_does_not_raise_when_user_can_access_vendor( self, mocker: MockerFixture, contributor ) -> None: mock = mocker.patch( "contributor.logic.contributors.ows_permissions.can_access_vendor", return_value=True, ) contributor_model = Contributor.model_validate(contributor) assert_authorization(contributor=contributor_model) mock.assert_called_once_with(vendor_id=100) def test_raises_404_when_user_cannot_access_vendor( self, mocker: MockerFixture, contributor ) -> None: mock = mocker.patch( "contributor.logic.contributors.ows_permissions.can_access_vendor", return_value=False, ) contributor_model = Contributor.model_validate(contributor) with pytest.raises(HTTPException) as exc_info: assert_authorization(contributor=contributor_model) mock.assert_called_once_with(vendor_id=100) assert exc_info.value.status_code == 404 assert ( exc_info.value.detail == f"Contributor {contributor_model.uuid} not found" ) class TestGetContributor: def test_returns_first_result_from_get_contributors( self, mocker: MockerFixture, session, contributor_uuid, contributor, label_profiles, identity_uuid, ) -> None: mock = mocker.patch( "contributor.logic.contributors.get_contributors", return_value=[contributor], ) result = get_contributor( session, contributor_uuid, profiles=label_profiles, identity_uuid=identity_uuid, ) mock.assert_called_once_with( session, [contributor_uuid], label_profiles, identity_uuid=identity_uuid, oa_user=False, ) assert result == contributor def test_returns_none_when_not_found( self, mocker: MockerFixture, session, contributor_uuid, label_profiles, identity_uuid, ) -> None: mocker.patch( "contributor.logic.contributors.get_contributors", return_value=[], ) assert ( get_contributor( session, contributor_uuid, profiles=label_profiles, identity_uuid=identity_uuid, ) is None ) def test_passes_oa_user_to_get_contributors( self, mocker: MockerFixture, session, contributor_uuid, contributor, label_profiles, identity_uuid, ) -> None: mock = mocker.patch( "contributor.logic.contributors.get_contributors", return_value=[Contributor.model_validate(contributor)], ) result = get_contributor( session, contributor_uuid, profiles=label_profiles, identity_uuid=identity_uuid, oa_user=True, ) mock.assert_called_once_with( session, [contributor_uuid], label_profiles, identity_uuid=identity_uuid, oa_user=True, ) assert result == Contributor.model_validate(contributor) def test_strips_spotify_artist_key_for_all_results_when_not_oa_user( self, mocker: MockerFixture, session, contributor_uuid, contributor, label_profiles, identity_uuid, ) -> None: mocker.patch( "contributor.logic.contributors.datasources.get_splitio_client", return_value=MagicMock(), ) mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) second_uuid = UUID("aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee") mocker.patch( "contributor.logic.contributors.neo4j_queries.get_by_uuids", return_value=[ {**contributor, "spotify_artist_key": "Q1W2E-R3T4Y-U5I6O-P7A8S-D9F0G"}, { **contributor, "uuid": str(second_uuid), "spotify_artist_key": "Z9X8C-V7B6N-M5L4K-J3H2G-F1D0S", }, ], ) results = get_contributors( session, [UUID(contributor_uuid), second_uuid], profiles=label_profiles, identity_uuid=identity_uuid, oa_user=False, ) assert all(r.spotify_artist_key is None for r in results) def test_keeps_spotify_artist_key_when_oa_user( self, mocker: MockerFixture, session, contributor_uuid, contributor, label_profiles, identity_uuid, ) -> None: mocker.patch( "contributor.logic.contributors.datasources.get_splitio_client", return_value=MagicMock(), ) mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mocker.patch( "contributor.logic.contributors.neo4j_queries.get_by_uuids", return_value=[ {**contributor, "spotify_artist_key": "P0O9I-U8Y7T-R6E5W-Q4A3S-D2F1G"} ], ) results = get_contributors( session, [contributor_uuid], profiles=label_profiles, identity_uuid=identity_uuid, oa_user=True, ) assert results[0].spotify_artist_key == "P0O9I-U8Y7T-R6E5W-Q4A3S-D2F1G" def test_returns_empty_list_when_none_found( self, mocker: MockerFixture, session, contributor_uuid, label_profiles, identity_uuid, ) -> None: mocker.patch( "contributor.logic.contributors.datasources.get_splitio_client", return_value=MagicMock(), ) mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mocker.patch( "contributor.logic.contributors.neo4j_queries.get_by_uuids", return_value=[] ) assert ( get_contributors( session, [contributor_uuid], profiles=label_profiles, identity_uuid=identity_uuid, ) == [] ) class TestUpdateContributor: def test_passes_is_admin_false_for_label_profiles( self, mocker: MockerFixture, session, contributor_uuid, contributor, label_profiles, ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mock = mocker.patch( "contributor.logic.contributors.neo4j_queries.update", return_value=contributor, ) result = update_contributor( session, contributor_uuid=contributor_uuid, profiles=label_profiles, spotify_id="3TVXtAsR1Inumwj472S9r4", apple_music_id=None, spotify_artist_key=None, isni=None, global_participant_uuid=None, updated_by="ows-contributor/test-uuid", ) mock.assert_called_once_with( session, uuid=contributor_uuid, profile_ids=[7123], is_admin=False, spotify_id="3TVXtAsR1Inumwj472S9r4", apple_music_id=None, spotify_artist_key=None, isni=None, global_participant_uuid=None, updated_by="ows-contributor/test-uuid", ) assert result == contributor def test_passes_is_admin_true_for_full_catalog_access( self, mocker: MockerFixture, session, contributor_uuid, contributor, admin_profiles, ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=True, ) mock = mocker.patch( "contributor.logic.contributors.neo4j_queries.update", return_value=contributor, ) result = update_contributor( session, contributor_uuid=contributor_uuid, profiles=admin_profiles, spotify_id="3TVXtAsR1Inumwj472S9r4", apple_music_id=None, spotify_artist_key=None, isni=None, global_participant_uuid=None, updated_by="ows-contributor/test-uuid", ) mock.assert_called_once_with( session, uuid=contributor_uuid, profile_ids=[], is_admin=True, spotify_id="3TVXtAsR1Inumwj472S9r4", apple_music_id=None, spotify_artist_key=None, isni=None, global_participant_uuid=None, updated_by="ows-contributor/test-uuid", ) assert result == contributor def test_returns_none_when_not_found( self, mocker: MockerFixture, session, contributor_uuid, label_profiles ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mocker.patch( "contributor.logic.contributors.neo4j_queries.update", return_value=None ) result = update_contributor( session, contributor_uuid=contributor_uuid, profiles=label_profiles, spotify_id=None, apple_music_id=None, spotify_artist_key=None, isni=None, global_participant_uuid=None, updated_by="ows-contributor/test-uuid", ) assert result is None class TestGetGlobalParticipant: @pytest.fixture def global_participant_uuid(self): return "b1a2c3d4-e5f6-4a7b-8c9d-0e1f2a3b4c5d" @pytest.fixture def global_participant(self): return { "id": "b1a2c3d4-e5f6-4a7b-8c9d-0e1f2a3b4c5d", "name": "Test Artist Global", "spotify_id": "4iV5W9uYEdYUVa79Axb7Rh", } def test_returns_global_participant( self, mocker: MockerFixture, session, global_participant_uuid, global_participant, label_profiles, ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mock = mocker.patch( "contributor.logic.contributors.neo4j_queries.get_gp_by_ids", return_value=[global_participant], ) result = get_global_participant( session, global_participant_id=global_participant_uuid, profiles=label_profiles, ) mock.assert_called_once_with( session, ids=[global_participant_uuid], profile_ids=[7123], is_admin=False ) assert result == global_participant def test_returns_none_when_not_found( self, mocker: MockerFixture, session, global_participant_uuid, label_profiles ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mocker.patch( "contributor.logic.contributors.neo4j_queries.get_gp_by_ids", return_value=[], ) assert ( get_global_participant( session, global_participant_id=global_participant_uuid, profiles=label_profiles, ) is None ) def test_passes_is_admin_true_for_full_catalog_access( self, mocker: MockerFixture, session, global_participant_uuid, global_participant, admin_profiles, ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=True, ) mock = mocker.patch( "contributor.logic.contributors.neo4j_queries.get_gp_by_ids", return_value=[global_participant], ) result = get_global_participant( session, global_participant_id=global_participant_uuid, profiles=admin_profiles, ) mock.assert_called_once_with( session, ids=[global_participant_uuid], profile_ids=[], is_admin=True ) assert result == global_participant class TestGetGlobalParticipants: @pytest.fixture def global_participant(self): return { "id": "b1a2c3d4-e5f6-4a7b-8c9d-0e1f2a3b4c5d", "name": "Test Artist Global", "spotify_id": "4iV5W9uYEdYUVa79Axb7Rh", } def test_calls_get_gp_by_ids_with_all_ids( self, mocker: MockerFixture, session, global_participant, label_profiles, ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mock = mocker.patch( "contributor.logic.contributors.neo4j_queries.get_gp_by_ids", return_value=[global_participant], ) second_id = "c2b3d4e5-f6a7-5b8c-9d0e-1f2a3b4c5d6e" result = get_global_participants( session, global_participant_ids=["b1a2c3d4-e5f6-4a7b-8c9d-0e1f2a3b4c5d", second_id], profiles=label_profiles, ) mock.assert_called_once_with( session, ids=["b1a2c3d4-e5f6-4a7b-8c9d-0e1f2a3b4c5d", second_id], profile_ids=[7123], is_admin=False, ) assert result == [global_participant] def test_passes_is_admin_true_for_full_catalog_access( self, mocker: MockerFixture, session, global_participant, admin_profiles, ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=True, ) mock = mocker.patch( "contributor.logic.contributors.neo4j_queries.get_gp_by_ids", return_value=[global_participant], ) result = get_global_participants( session, global_participant_ids=["b1a2c3d4-e5f6-4a7b-8c9d-0e1f2a3b4c5d"], profiles=admin_profiles, ) mock.assert_called_once_with( session, ids=["b1a2c3d4-e5f6-4a7b-8c9d-0e1f2a3b4c5d"], profile_ids=[], is_admin=True, ) assert result == [global_participant] def test_returns_empty_list_when_none_found( self, mocker: MockerFixture, session, label_profiles, ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mocker.patch( "contributor.logic.contributors.neo4j_queries.get_gp_by_ids", return_value=[], ) result = get_global_participants( session, global_participant_ids=["no-match"], profiles=label_profiles, ) assert result == [] class TestGetContributorsForGlobalParticipant: @pytest.fixture def global_participant_uuid(self): return "b1a2c3d4-e5f6-4a7b-8c9d-0e1f2a3b4c5d" @pytest.fixture def contributors(self, contributor): return [contributor] def test_returns_contributors( self, mocker: MockerFixture, session, global_participant_uuid, contributors, label_profiles, ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mock = mocker.patch( "contributor.logic.contributors.neo4j_queries.get_contributors_for_global_participant", return_value=contributors, ) result = get_contributors_for_global_participant( session, global_participant_id=global_participant_uuid, profiles=label_profiles, ) mock.assert_called_once_with( session, global_participant_id=global_participant_uuid, profile_ids=[7123], is_admin=False, ) assert result == contributors def test_passes_is_admin_true_for_full_catalog_access( self, mocker: MockerFixture, session, global_participant_uuid, contributors, admin_profiles, ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=True, ) mock = mocker.patch( "contributor.logic.contributors.neo4j_queries.get_contributors_for_global_participant", return_value=contributors, ) result = get_contributors_for_global_participant( session, global_participant_id=global_participant_uuid, profiles=admin_profiles, ) mock.assert_called_once_with( session, global_participant_id=global_participant_uuid, profile_ids=[], is_admin=True, ) assert result == contributors def test_returns_none_when_not_found( self, mocker: MockerFixture, session, global_participant_uuid, label_profiles, ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mocker.patch( "contributor.logic.contributors.neo4j_queries.get_contributors_for_global_participant", return_value=None, ) assert ( get_contributors_for_global_participant( session, global_participant_id=global_participant_uuid, profiles=label_profiles, ) is None ) class TestGetTrackParticipations: @pytest.fixture def track_participations(self, contributor_uuid): return { "contributor": {"uuid": contributor_uuid}, "participations": [ { "contributor": {"uuid": contributor_uuid}, "participated_as": "performer", "track": {"tuid": 101}, }, { "contributor": {"uuid": contributor_uuid}, "participated_as": "featuring", "track": {"tuid": 202}, }, ], } def test_returns_participations( self, mocker: MockerFixture, session, contributor_uuid, track_participations, label_profiles, identity_uuid, ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mock = mocker.patch( "contributor.logic.contributors.tp_queries.get_by_label_participant_uuid", return_value=track_participations, ) result = get_track_participations( session, contributor_uuid, profiles=label_profiles, identity_uuid=identity_uuid, ) mock.assert_called_once_with( session, uuid=contributor_uuid, identity_uuid=identity_uuid, profile_ids=[7123], is_admin=False, ) assert result == track_participations def test_returns_none_when_not_found( self, mocker: MockerFixture, session, contributor_uuid, label_profiles, identity_uuid, ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mocker.patch( "contributor.logic.contributors.tp_queries.get_by_label_participant_uuid", return_value=None, ) assert ( get_track_participations( session, contributor_uuid, profiles=label_profiles, identity_uuid=identity_uuid, ) is None ) def test_passes_is_admin_true_for_full_catalog_access( self, mocker: MockerFixture, session, contributor_uuid, track_participations, admin_profiles, identity_uuid, ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=True, ) mock = mocker.patch( "contributor.logic.contributors.tp_queries.get_by_label_participant_uuid", return_value=track_participations, ) result = get_track_participations( session, contributor_uuid, profiles=admin_profiles, identity_uuid=identity_uuid, ) mock.assert_called_once_with( session, uuid=contributor_uuid, identity_uuid=identity_uuid, profile_ids=[], is_admin=True, ) assert result == track_participations def test_filters_out_feature_to_primary( self, mocker: MockerFixture, session, contributor_uuid, label_profiles, identity_uuid, ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mocker.patch( "contributor.logic.contributors.tp_queries.get_by_label_participant_uuid", return_value={ "contributor": {"uuid": contributor_uuid}, "participations": [ { "contributor": {"uuid": contributor_uuid}, "participated_as": "performer", "track": {"tuid": 101}, }, { "contributor": {"uuid": contributor_uuid}, "participated_as": "feature_to_primary", "track": {"tuid": 303}, }, { "contributor": {"uuid": contributor_uuid}, "participated_as": "featuring", "track": {"tuid": 202}, }, ], }, ) result = get_track_participations( session, contributor_uuid, profiles=label_profiles, identity_uuid=identity_uuid, ) assert result is not None assert result["participations"] == [ { "contributor": {"uuid": contributor_uuid}, "participated_as": "performer", "track": {"tuid": 101}, }, { "contributor": {"uuid": contributor_uuid}, "participated_as": "featuring", "track": {"tuid": 202}, }, ] def test_filters_out_all_feature_to_primary( self, mocker: MockerFixture, session, contributor_uuid, label_profiles, identity_uuid, ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mocker.patch( "contributor.logic.contributors.tp_queries.get_by_label_participant_uuid", return_value={ "contributor": {"uuid": contributor_uuid}, "participations": [ { "contributor": {"uuid": contributor_uuid}, "participated_as": "feature_to_primary", "track": {"tuid": 303}, }, { "contributor": {"uuid": contributor_uuid}, "participated_as": "feature_to_primary", "track": {"tuid": 404}, }, ], }, ) result = get_track_participations( session, contributor_uuid, profiles=label_profiles, identity_uuid=identity_uuid, ) assert result is not None assert result["participations"] == [] class TestGetTrackParticipationsForTrack: @pytest.fixture def track_participations(self): return { "track": {"tuid": 101}, "participations": [ { "contributor": {"uuid": "550e8400-e29b-41d4-a716-446655440000"}, "participated_as": "performer", "tuid": 101, "track_artist_id": 1001, }, { "contributor": {"uuid": "aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee"}, "participated_as": "featuring", "tuid": 101, "track_artist_id": 1002, }, ], } def test_returns_track_participations( self, mocker: MockerFixture, session, track_participations, label_profiles, identity_uuid, ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mock = mocker.patch( "contributor.logic.contributors.tp_queries.get_by_track_ids", return_value=[track_participations], ) result = get_track_participations_for_track( session, track_id=101, profiles=label_profiles, identity_uuid=identity_uuid ) mock.assert_called_once_with( session, track_ids=[101], identity_uuid=identity_uuid, profile_ids=[7123], is_admin=False, ) assert result == track_participations def test_returns_none_when_not_found( self, mocker: MockerFixture, session, label_profiles, identity_uuid ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mocker.patch( "contributor.logic.contributors.tp_queries.get_by_track_ids", return_value=None, ) assert ( get_track_participations_for_track( session, track_id=101, profiles=label_profiles, identity_uuid=identity_uuid, ) is None ) def test_passes_is_admin_true_for_full_catalog_access( self, mocker: MockerFixture, session, track_participations, admin_profiles, identity_uuid, ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=True, ) mock = mocker.patch( "contributor.logic.contributors.tp_queries.get_by_track_ids", return_value=[track_participations], ) result = get_track_participations_for_track( session, track_id=101, profiles=admin_profiles, identity_uuid=identity_uuid ) mock.assert_called_once_with( session, track_ids=[101], identity_uuid=identity_uuid, profile_ids=[], is_admin=True, ) assert result == track_participations def test_filters_null_and_feature_to_primary( self, mocker: MockerFixture, session, label_profiles, identity_uuid, ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mocker.patch( "contributor.logic.contributors.tp_queries.get_by_track_ids", return_value=[ { "track": {"tuid": 101}, "participations": [ None, { "contributor": { "uuid": "550e8400-e29b-41d4-a716-446655440000" }, "participated_as": "feature_to_primary", "tuid": 101, "track_artist_id": 1001, }, { "contributor": { "uuid": "aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee" }, "participated_as": "performer", "tuid": 101, "track_artist_id": 1002, }, ], } ], ) result = get_track_participations_for_track( session, track_id=101, profiles=label_profiles, identity_uuid=identity_uuid ) assert result is not None assert result["participations"] == [ { "contributor": {"uuid": "aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee"}, "participated_as": "performer", "tuid": 101, "track_artist_id": 1002, } ] class TestGetTrackParticipationsByTrackIds: def test_returns_track_participations( self, mocker: MockerFixture, session, label_profiles, identity_uuid, ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mock = mocker.patch( "contributor.logic.contributors.tp_queries.get_by_track_ids", return_value=[ { "track": {"tuid": 101}, "participations": [ { "contributor": { "uuid": "550e8400-e29b-41d4-a716-446655440000" }, "participated_as": "performer", "tuid": 101, "track_artist_id": 1001, } ], }, { "track": {"tuid": 202}, "participations": [ { "contributor": { "uuid": "aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee" }, "participated_as": "featuring", "tuid": 202, "track_artist_id": 2002, } ], }, ], ) result = get_track_participations_by_track_ids( session, track_ids=[101, 202], profiles=label_profiles, identity_uuid=identity_uuid, ) mock.assert_called_once_with( session, track_ids=[101, 202], identity_uuid=identity_uuid, profile_ids=[7123], is_admin=False, ) assert result == [ { "track": {"tuid": 101}, "participations": [ { "contributor": {"uuid": "550e8400-e29b-41d4-a716-446655440000"}, "participated_as": "performer", "tuid": 101, "track_artist_id": 1001, } ], }, { "track": {"tuid": 202}, "participations": [ { "contributor": {"uuid": "aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee"}, "participated_as": "featuring", "tuid": 202, "track_artist_id": 2002, } ], }, ] def test_returns_empty_list_when_none_found( self, mocker: MockerFixture, session, label_profiles, identity_uuid ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mocker.patch( "contributor.logic.contributors.tp_queries.get_by_track_ids", return_value=[], ) assert ( get_track_participations_by_track_ids( session, track_ids=[999], profiles=label_profiles, identity_uuid=identity_uuid, ) == [] ) class TestGetProductParticipations: @pytest.fixture def product_participations(self, contributor_uuid): return { "contributor": {"uuid": contributor_uuid}, "participations": [ { "contributor": {"uuid": contributor_uuid}, "participated_as": "performer", "product": {"product_id": 101}, "release_artist_id": 1001, }, { "contributor": {"uuid": contributor_uuid}, "participated_as": "featuring", "product": {"product_id": 202}, "release_artist_id": 2002, }, ], } def test_returns_participations( self, mocker: MockerFixture, session, contributor_uuid, product_participations, label_profiles, ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mock = mocker.patch( "contributor.logic.contributors.pp_queries.get_by_label_participant_uuid", return_value=product_participations, ) result = get_product_participations( session, contributor_uuid, profiles=label_profiles ) mock.assert_called_once_with( session, uuid=contributor_uuid, profile_ids=[7123], is_admin=False ) assert result == product_participations def test_returns_none_when_not_found( self, mocker: MockerFixture, session, contributor_uuid, label_profiles ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mocker.patch( "contributor.logic.contributors.pp_queries.get_by_label_participant_uuid", return_value=None, ) assert ( get_product_participations( session, contributor_uuid, profiles=label_profiles ) is None ) def test_passes_is_admin_true_for_full_catalog_access( self, mocker: MockerFixture, session, contributor_uuid, product_participations, admin_profiles, ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=True, ) mock = mocker.patch( "contributor.logic.contributors.pp_queries.get_by_label_participant_uuid", return_value=product_participations, ) result = get_product_participations( session, contributor_uuid, profiles=admin_profiles ) mock.assert_called_once_with( session, uuid=contributor_uuid, profile_ids=[], is_admin=True ) assert result == product_participations def test_filters_out_feature_to_primary( self, mocker: MockerFixture, session, contributor_uuid, label_profiles, ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mocker.patch( "contributor.logic.contributors.pp_queries.get_by_label_participant_uuid", return_value={ "contributor": {"uuid": contributor_uuid}, "participations": [ { "contributor": {"uuid": contributor_uuid}, "participated_as": "performer", "product": {"product_id": 101}, "release_artist_id": 1001, }, { "contributor": {"uuid": contributor_uuid}, "participated_as": "feature_to_primary", "product": {"product_id": 303}, "release_artist_id": 3003, }, { "contributor": {"uuid": contributor_uuid}, "participated_as": "featuring", "product": {"product_id": 202}, "release_artist_id": 2002, }, ], }, ) result = get_product_participations( session, contributor_uuid, profiles=label_profiles ) assert result is not None assert result["participations"] == [ { "contributor": {"uuid": contributor_uuid}, "participated_as": "performer", "product": {"product_id": 101}, "release_artist_id": 1001, }, { "contributor": {"uuid": contributor_uuid}, "participated_as": "featuring", "product": {"product_id": 202}, "release_artist_id": 2002, }, ] def test_filters_out_all_feature_to_primary( self, mocker: MockerFixture, session, contributor_uuid, label_profiles, ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mocker.patch( "contributor.logic.contributors.pp_queries.get_by_label_participant_uuid", return_value={ "contributor": {"uuid": contributor_uuid}, "participations": [ { "contributor": {"uuid": contributor_uuid}, "participated_as": "feature_to_primary", "product": {"product_id": 303}, "release_artist_id": 3003, }, { "contributor": {"uuid": contributor_uuid}, "participated_as": "feature_to_primary", "product": {"product_id": 404}, "release_artist_id": 4004, }, ], }, ) result = get_product_participations( session, contributor_uuid, profiles=label_profiles ) assert result is not None assert result["participations"] == [] class TestGetSoundRecordingParticipations: @pytest.fixture def sound_recording_participations(self, contributor_uuid): return { "contributor": {"uuid": contributor_uuid}, "participations": [ { "contributor": {"uuid": contributor_uuid}, "label_sound_recording": { "id": "0457b5bb-b032-464d-a232-7cf8100a057a" }, "role": { "apple_role_name": "Recording Engineer", "category": {"name": "Studio Personnel", "uuid": "uuid-1"}, "ddex_role_name": "RecordingEngineer", "name": "Recording Engineer", }, "sequence_number": 1, }, { "contributor": {"uuid": contributor_uuid}, "label_sound_recording": { "id": "891e83b4-0502-4230-8fa9-733ee70949ea" }, "role": { "apple_role_name": "Mixing Engineer", "category": {"name": "Studio Personnel", "uuid": "uuid-2"}, "ddex_role_name": "MixingEngineer", "name": "Mixing Engineer", }, "sequence_number": 2, }, ], } def test_returns_participations( self, mocker: MockerFixture, session, contributor_uuid, sound_recording_participations, label_profiles, ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mock = mocker.patch( "contributor.logic.contributors.srp_queries.get_by_label_participant_uuid", return_value=sound_recording_participations, ) result = get_sound_recording_participations( session, contributor_uuid, profiles=label_profiles ) mock.assert_called_once_with( session, uuid=contributor_uuid, profile_ids=[7123], is_admin=False ) assert result == sound_recording_participations def test_returns_none_when_not_found( self, mocker: MockerFixture, session, contributor_uuid, label_profiles ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mocker.patch( "contributor.logic.contributors.srp_queries.get_by_label_participant_uuid", return_value=None, ) assert ( get_sound_recording_participations( session, contributor_uuid, profiles=label_profiles ) is None ) def test_passes_is_admin_true_for_full_catalog_access( self, mocker: MockerFixture, session, contributor_uuid, sound_recording_participations, admin_profiles, ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=True, ) mock = mocker.patch( "contributor.logic.contributors.srp_queries.get_by_label_participant_uuid", return_value=sound_recording_participations, ) result = get_sound_recording_participations( session, contributor_uuid, profiles=admin_profiles ) mock.assert_called_once_with( session, uuid=contributor_uuid, profile_ids=[], is_admin=True ) assert result == sound_recording_participations class TestGetProjects: @pytest.fixture def projects(self, contributor_uuid): return { "contributor": {"uuid": contributor_uuid}, "projects": [ {"project_id": 101}, {"project_id": 202}, ], } def test_returns_projects( self, mocker: MockerFixture, session, contributor_uuid, projects, label_profiles, ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mock = mocker.patch( "contributor.logic.contributors.project_queries.get_by_label_participant_uuid", return_value=projects, ) result = get_projects(session, contributor_uuid, profiles=label_profiles) mock.assert_called_once_with( session, uuid=contributor_uuid, profile_ids=[7123], is_admin=False ) assert result == projects def test_returns_none_when_not_found( self, mocker: MockerFixture, session, contributor_uuid, label_profiles ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mocker.patch( "contributor.logic.contributors.project_queries.get_by_label_participant_uuid", return_value=None, ) assert get_projects(session, contributor_uuid, profiles=label_profiles) is None def test_passes_is_admin_true_for_full_catalog_access( self, mocker: MockerFixture, session, contributor_uuid, projects, admin_profiles, ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=True, ) mock = mocker.patch( "contributor.logic.contributors.project_queries.get_by_label_participant_uuid", return_value=projects, ) result = get_projects(session, contributor_uuid, profiles=admin_profiles) mock.assert_called_once_with( session, uuid=contributor_uuid, profile_ids=[], is_admin=True ) assert result == projects class TestGetProjectContributors: @pytest.fixture def project_contributors(self): return [ { "project": {"project_id": 101}, "contributors": [ { "id": 42, "uuid": "550e8400-e29b-41d4-a716-446655440000", "name": "Test Artist", "spotify_id": "3TVXtAsR1Inumwj472S9r4", "apple_music_id": None, "spotify_artist_key": None, "isni": None, "company_brand_uuid": "11110000-2222-4333-8444-555555555555", "parent_company_uuid": "66660000-7777-4888-8999-aaaaaaaaaaaa", "artist_info_ids": [123], "global_participant": None, "label": { "uuid": "cccc0000-dddd-4eee-8fff-aaaaaaaaaaaa", "vendor_id": 100, "subaccount_id": 0, }, } ], } ] def test_calls_get_contributors_by_project_ids_with_all_ids( self, mocker: MockerFixture, session, project_contributors, label_profiles, ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mock = mocker.patch( "contributor.logic.contributors.project_queries.get_contributors_by_project_ids", return_value=project_contributors, ) result = get_project_contributors( session, project_ids=[101, 202], profiles=label_profiles, ) mock.assert_called_once_with( session, project_ids=[101, 202], profile_ids=[7123], is_admin=False, ) assert result == project_contributors def test_passes_is_admin_true_for_full_catalog_access( self, mocker: MockerFixture, session, project_contributors, admin_profiles, ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=True, ) mock = mocker.patch( "contributor.logic.contributors.project_queries.get_contributors_by_project_ids", return_value=project_contributors, ) result = get_project_contributors( session, project_ids=[101], profiles=admin_profiles, ) mock.assert_called_once_with( session, project_ids=[101], profile_ids=[], is_admin=True, ) assert result == project_contributors def test_returns_empty_list_when_none_found( self, mocker: MockerFixture, session, label_profiles, ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mocker.patch( "contributor.logic.contributors.project_queries.get_contributors_by_project_ids", return_value=[], ) result = get_project_contributors( session, project_ids=[999], profiles=label_profiles, ) assert result == [] class TestGetProductContributors: @pytest.fixture def product_contributors(self): return [ { "product": {"product_id": 12345}, "contributors": [ { "contributor": {"uuid": "550e8400-e29b-41d4-a716-446655440000"}, "participated_as": "performer", "product": {"product_id": 12345}, "release_artist_id": 1001, }, { "contributor": {"uuid": "aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee"}, "participated_as": "featuring", "product": {"product_id": 12345}, "release_artist_id": 1002, }, ], "display_artists": [], } ] @pytest.fixture(autouse=True) def display_artists(self, mocker: MockerFixture) -> None: mocker.patch( "contributor.logic.contributors.display_artist_queries.get_by_product_ids", return_value=[], ) def test_calls_get_by_product_ids_with_all_ids( self, mocker: MockerFixture, session, product_contributors, label_profiles ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mock = mocker.patch( "contributor.logic.contributors.pp_queries.get_by_product_ids", return_value=product_contributors, ) result = get_product_contributors( session, product_ids=[12345, 67890], profiles=label_profiles ) mock.assert_called_once_with( session, product_ids=[12345, 67890], profile_ids=[7123], is_admin=False ) assert result == product_contributors def test_passes_is_admin_true_for_full_catalog_access( self, mocker: MockerFixture, session, product_contributors, admin_profiles ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=True, ) mock = mocker.patch( "contributor.logic.contributors.pp_queries.get_by_product_ids", return_value=product_contributors, ) result = get_product_contributors( session, product_ids=[12345], profiles=admin_profiles ) mock.assert_called_once_with( session, product_ids=[12345], profile_ids=[], is_admin=True ) assert result == product_contributors def test_filters_out_feature_to_primary_from_all_products( self, mocker: MockerFixture, session, label_profiles ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mocker.patch( "contributor.logic.contributors.pp_queries.get_by_product_ids", return_value=[ { "product": {"product_id": 12345}, "contributors": [ { "contributor": { "uuid": "550e8400-e29b-41d4-a716-446655440000" }, "participated_as": "performer", "product": {"product_id": 12345}, "release_artist_id": 1001, }, { "contributor": { "uuid": "bbbbbbbb-bbbb-4bbb-8bbb-bbbbbbbbbbbb" }, "participated_as": "feature_to_primary", "product": {"product_id": 12345}, "release_artist_id": 2002, }, ], }, { "product": {"product_id": 67890}, "contributors": [ { "contributor": { "uuid": "cccccccc-cccc-4ccc-8ccc-cccccccccccc" }, "participated_as": "feature_to_primary", "product": {"product_id": 67890}, "release_artist_id": 3003, }, ], }, ], ) result = get_product_contributors( session, product_ids=[12345, 67890], profiles=label_profiles ) assert len(result) == 2 assert len(result[0]["contributors"]) == 1 assert result[0]["contributors"][0]["participated_as"] == "performer" assert len(result[1]["contributors"]) == 0 assert result[0]["display_artists"] == [] assert result[1]["display_artists"] == [] def test_preserves_display_artists_from_product_query( self, mocker: MockerFixture, session, label_profiles ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) product_results = [ { "product": {"product_id": 12345}, "contributors": [ { "contributor": {"uuid": "550e8400-e29b-41d4-a716-446655440000"}, "participated_as": "performer", "product": {"product_id": 12345}, "release_artist_id": 1001, } ], } ] display_artist_results = [ { "product": {"product_id": 12345}, "display_artists": [ { "contributor": {"uuid": "11111111-2222-4333-8444-555555555555"}, "role": "primary", } ], } ] mocker.patch( "contributor.logic.contributors.pp_queries.get_by_product_ids", return_value=product_results, ) mocker.patch( "contributor.logic.contributors.display_artist_queries.get_by_product_ids", return_value=display_artist_results, ) result = get_product_contributors( session, product_ids=[12345], profiles=label_profiles ) assert result == [ { "product": {"product_id": 12345}, "contributors": [ { "contributor": {"uuid": "550e8400-e29b-41d4-a716-446655440000"}, "participated_as": "performer", "product": {"product_id": 12345}, "release_artist_id": 1001, } ], "display_artists": [ { "contributor": {"uuid": "11111111-2222-4333-8444-555555555555"}, "role": "primary", } ], } ] def test_returns_empty_list_when_none_found( self, mocker: MockerFixture, session, label_profiles ) -> None: mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) mocker.patch( "contributor.logic.contributors.pp_queries.get_by_product_ids", return_value=[], ) result = get_product_contributors( session, product_ids=[999], profiles=label_profiles ) assert result == [] class TestRenameContributor: @pytest.fixture def contributor_model(self, contributor): return Contributor.model_validate(contributor) @pytest.fixture def mysql_session(self): return MagicMock() @pytest.fixture def renamed_contributor(self, contributor): return {**contributor, "name": "New Name"} def _patch_dependencies( self, mocker: MockerFixture, *, contributor_model=None, vendor_access=True, is_admin=False, rename_result=None, ): mocker.patch( "contributor.logic.contributors.get_contributor", return_value=contributor_model, ) mocker.patch( "contributor.logic.contributors.check_vendor_access", return_value=vendor_access, ) mocker.patch( "contributor.logic.contributors.rename_art_relations_contributor_records", ) mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=is_admin, ) mocker.patch( "contributor.logic.contributors.neo4j_queries.rename", return_value=rename_result, ) def test_returns_renamed_contributor( self, mocker: MockerFixture, session, mysql_session, contributor_uuid, contributor_model, renamed_contributor, label_profiles, identity_uuid, ) -> None: self._patch_dependencies( mocker, contributor_model=contributor_model, rename_result=renamed_contributor, ) result = rename_contributor( session, session=mysql_session, contributor_uuid=contributor_uuid, name="New Name", updated_by="ows-contributor/test-user", profiles=label_profiles, identity_uuid=identity_uuid, ) assert result == Contributor.model_validate(renamed_contributor) def test_commits_mysql_session_on_success( self, mocker: MockerFixture, session, mysql_session, contributor_uuid, contributor_model, renamed_contributor, label_profiles, identity_uuid, ) -> None: self._patch_dependencies( mocker, contributor_model=contributor_model, rename_result=renamed_contributor, ) rename_contributor( session, session=mysql_session, contributor_uuid=contributor_uuid, name="New Name", updated_by="ows-contributor/test-user", profiles=label_profiles, identity_uuid=identity_uuid, ) mysql_session.commit.assert_called_once() def test_raises_404_when_contributor_not_found( self, mocker: MockerFixture, session, mysql_session, contributor_uuid, label_profiles, identity_uuid, ) -> None: self._patch_dependencies(mocker, contributor_model=None) with pytest.raises(HTTPException) as exc_info: rename_contributor( session, session=mysql_session, contributor_uuid=contributor_uuid, name="New Name", updated_by="ows-contributor/test-user", profiles=label_profiles, identity_uuid=identity_uuid, ) assert exc_info.value.status_code == 404 def test_raises_400_when_no_vendor_id( self, mocker: MockerFixture, session, mysql_session, contributor_uuid, contributor, label_profiles, identity_uuid, ) -> None: no_vendor = {**contributor, "label": {**contributor["label"], "vendor_id": 0}} self._patch_dependencies( mocker, contributor_model=Contributor.model_validate(no_vendor), ) with pytest.raises(HTTPException) as exc_info: rename_contributor( session, session=mysql_session, contributor_uuid=contributor_uuid, name="New Name", updated_by="ows-contributor/test-user", profiles=label_profiles, identity_uuid=identity_uuid, ) assert exc_info.value.status_code == 400 def test_raises_403_when_no_vendor_access( self, mocker: MockerFixture, session, mysql_session, contributor_uuid, contributor_model, label_profiles, identity_uuid, ) -> None: self._patch_dependencies( mocker, contributor_model=contributor_model, vendor_access=False, ) with pytest.raises(HTTPException) as exc_info: rename_contributor( session, session=mysql_session, contributor_uuid=contributor_uuid, name="New Name", updated_by="ows-contributor/test-user", profiles=label_profiles, identity_uuid=identity_uuid, ) assert exc_info.value.status_code == 403 def test_raises_500_and_rolls_back_on_rename_failure( self, mocker: MockerFixture, session, mysql_session, contributor_uuid, contributor_model, label_profiles, identity_uuid, ) -> None: self._patch_dependencies( mocker, contributor_model=contributor_model, ) mocker.patch( "contributor.logic.contributors.rename_art_relations_contributor_records", side_effect=RuntimeError("db error"), ) with pytest.raises(HTTPException) as exc_info: rename_contributor( session, session=mysql_session, contributor_uuid=contributor_uuid, name="New Name", updated_by="ows-contributor/test-user", profiles=label_profiles, identity_uuid=identity_uuid, ) assert exc_info.value.status_code == 500 mysql_session.rollback.assert_called_once() def test_calls_art_relations_rename_with_correct_args( self, mocker: MockerFixture, session, mysql_session, contributor_uuid, contributor_model, renamed_contributor, label_profiles, identity_uuid, ) -> None: self._patch_dependencies( mocker, contributor_model=contributor_model, rename_result=renamed_contributor, ) art_rename = mocker.patch( "contributor.logic.contributors.rename_art_relations_contributor_records", ) rename_contributor( session, session=mysql_session, contributor_uuid=contributor_uuid, name="New Name", updated_by="ows-contributor/test-user", profiles=label_profiles, identity_uuid=identity_uuid, ) art_rename.assert_called_once_with( session=mysql_session, name="New Name", vendor_id=100, current_name="Test Artist", ) def test_calls_neo4j_rename_with_correct_args( self, mocker: MockerFixture, session, mysql_session, contributor_uuid, contributor_model, renamed_contributor, label_profiles, identity_uuid, ) -> None: self._patch_dependencies( mocker, contributor_model=contributor_model, rename_result=renamed_contributor, is_admin=True, ) neo4j_rename = mocker.patch( "contributor.logic.contributors.neo4j_queries.rename", return_value=renamed_contributor, ) rename_contributor( session, session=mysql_session, contributor_uuid=contributor_uuid, name="New Name", updated_by="ows-contributor/test-user", profiles=label_profiles, identity_uuid=identity_uuid, ) neo4j_rename.assert_called_once_with( session=session, uuid=contributor_uuid, name="New Name", updated_by="ows-contributor/test-user", profile_ids=[7123], is_admin=True, ) class TestMergeContributors: @pytest.fixture def neo4j_session(self): return MagicMock() @pytest.fixture def duplicate_uuid(self): return UUID("aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee") @pytest.fixture def contributor_obj(self, contributor_uuid): return Contributor( id=42, uuid=contributor_uuid, name="Test Artist", label=Label( uuid=UUID("cccc0000-dddd-4eee-8fff-aaaaaaaaaaaa"), vendor_id=100 ), ) @pytest.fixture def duplicate_obj(self, duplicate_uuid): return Contributor( id=99, uuid=duplicate_uuid, name="Duplicate Artist", label=Label( uuid=UUID("dddd0000-eeee-4fff-9aaa-bbbbbbbbbbbb"), vendor_id=200 ), ) @pytest.fixture def artist_info_fixture(self): return {"artist_id": 1, "name": "Test Artist", "vendor_id": 100} @pytest.fixture def duplicate_artist_info_fixture(self): return {"artist_id": 2, "name": "Duplicate Artist", "vendor_id": 200} def test_raises_when_contributor_not_found( self, mocker: MockerFixture, session, neo4j_session, contributor_uuid, duplicate_uuid, label_profiles, identity_uuid, ) -> None: mocker.patch( "contributor.logic.contributors.get_contributor", return_value=None, ) with pytest.raises(AssertionError): merge_contributors( session=session, neo4j_session=neo4j_session, contributor_uuid=contributor_uuid, duplicate_uuid=duplicate_uuid, profiles=label_profiles, updated_by="ows-contributor/test-user", identity_uuid=identity_uuid, ) def test_raises_400_when_contributor_has_no_vendor_id( self, mocker: MockerFixture, session, neo4j_session, contributor_uuid, duplicate_uuid, label_profiles, identity_uuid, ) -> None: no_vendor_contributor = Contributor( id=42, uuid=contributor_uuid, name="Test Artist", label=Label( uuid=UUID("cccc0000-dddd-4eee-8fff-aaaaaaaaaaaa"), vendor_id=None ), ) mocker.patch( "contributor.logic.contributors.get_contributor", return_value=no_vendor_contributor, ) with pytest.raises(HTTPException) as exc_info: merge_contributors( session=session, neo4j_session=neo4j_session, contributor_uuid=contributor_uuid, duplicate_uuid=duplicate_uuid, profiles=label_profiles, updated_by="ows-contributor/test-user", identity_uuid=identity_uuid, ) assert exc_info.value.status_code == 400 def test_raises_when_duplicate_contributor_not_found( self, mocker: MockerFixture, session, neo4j_session, contributor_uuid, duplicate_uuid, label_profiles, contributor_obj, identity_uuid, ) -> None: mocker.patch( "contributor.logic.contributors.get_contributor", side_effect=[contributor_obj, None], ) with pytest.raises(AssertionError): merge_contributors( session=session, neo4j_session=neo4j_session, contributor_uuid=contributor_uuid, duplicate_uuid=duplicate_uuid, profiles=label_profiles, updated_by="ows-contributor/test-user", identity_uuid=identity_uuid, ) def test_does_not_reassign_when_artist_info_not_found( self, mocker: MockerFixture, session, neo4j_session, contributor_uuid, duplicate_uuid, label_profiles, contributor_obj, duplicate_obj, identity_uuid, ) -> None: mocker.patch( "contributor.logic.contributors.get_contributor", side_effect=[contributor_obj, duplicate_obj, contributor_obj], ) mocker.patch( "contributor.logic.contributors.artist_info_queries.get", return_value=None, ) mocker.patch( "contributor.logic.contributors.rename_art_relations_contributor_records" ) mock_reassign_content = mocker.patch( "contributor.logic.contributors.reassign_contributor_content" ) mock_reassign_artist_info = mocker.patch( "contributor.logic.contributors.reassign_artist_info" ) mock_remove = mocker.patch( "contributor.logic.contributors.remove_artist_info_references" ) mocker.patch("contributor.logic.contributors.neo4j_queries.merge") mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) merge_contributors( session=session, neo4j_session=neo4j_session, contributor_uuid=contributor_uuid, duplicate_uuid=duplicate_uuid, profiles=label_profiles, updated_by="ows-contributor/test-user", identity_uuid=identity_uuid, ) mock_reassign_content.assert_not_called() mock_reassign_artist_info.assert_not_called() mock_remove.assert_not_called() def test_does_not_reassign_when_duplicate_artist_info_not_found( self, mocker: MockerFixture, session, neo4j_session, contributor_uuid, duplicate_uuid, label_profiles, contributor_obj, duplicate_obj, artist_info_fixture, identity_uuid, ) -> None: mocker.patch( "contributor.logic.contributors.get_contributor", side_effect=[contributor_obj, duplicate_obj, contributor_obj], ) mocker.patch( "contributor.logic.contributors.artist_info_queries.get", side_effect=[artist_info_fixture, None], ) mocker.patch( "contributor.logic.contributors.rename_art_relations_contributor_records" ) mock_reassign_content = mocker.patch( "contributor.logic.contributors.reassign_contributor_content" ) mock_reassign_artist_info = mocker.patch( "contributor.logic.contributors.reassign_artist_info" ) mock_remove = mocker.patch( "contributor.logic.contributors.remove_artist_info_references" ) mocker.patch("contributor.logic.contributors.neo4j_queries.merge") mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) merge_contributors( session=session, neo4j_session=neo4j_session, contributor_uuid=contributor_uuid, duplicate_uuid=duplicate_uuid, profiles=label_profiles, updated_by="ows-contributor/test-user", identity_uuid=identity_uuid, ) mock_reassign_content.assert_not_called() mock_reassign_artist_info.assert_not_called() mock_remove.assert_not_called() def test_successful_merge_calls_all_methods_and_returns_contributor( self, mocker: MockerFixture, session, neo4j_session, contributor_uuid, duplicate_uuid, label_profiles, contributor_obj, duplicate_obj, artist_info_fixture, duplicate_artist_info_fixture, identity_uuid, ) -> None: mocker.patch( "contributor.logic.contributors.get_contributor", side_effect=[contributor_obj, duplicate_obj, contributor_obj], ) mocker.patch( "contributor.logic.contributors.artist_info_queries.get", side_effect=[artist_info_fixture, duplicate_artist_info_fixture], ) mock_rename = mocker.patch( "contributor.logic.contributors.rename_art_relations_contributor_records" ) mocker.patch("contributor.logic.contributors.reassign_contributor_content") mock_reassign = mocker.patch( "contributor.logic.contributors.reassign_artist_info" ) mock_remove = mocker.patch( "contributor.logic.contributors.remove_artist_info_references" ) mock_neo4j_merge = mocker.patch( "contributor.logic.contributors.neo4j_queries.merge" ) mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) result = merge_contributors( session=session, neo4j_session=neo4j_session, contributor_uuid=contributor_uuid, duplicate_uuid=duplicate_uuid, profiles=label_profiles, updated_by="ows-contributor/test-user", identity_uuid=identity_uuid, ) mock_rename.assert_called_once_with( session=session, name=contributor_obj.name, vendor_id=contributor_obj.label.vendor_id, current_name=duplicate_obj.name, ) mock_reassign.assert_called_once_with( session=session, artist_name=contributor_obj.name, artist_info_id=artist_info_fixture["artist_id"], duplicate_artist_info_id=duplicate_artist_info_fixture["artist_id"], ) mock_remove.assert_called_once_with( session=session, artist_info_id=duplicate_artist_info_fixture["artist_id"], ) mock_neo4j_merge.assert_called_once() assert result == contributor_obj def test_successful_merge_reassigns_contributor_content( self, mocker: MockerFixture, session, neo4j_session, contributor_uuid, duplicate_uuid, label_profiles, contributor_obj, duplicate_obj, artist_info_fixture, duplicate_artist_info_fixture, identity_uuid, ) -> None: mocker.patch( "contributor.logic.contributors.get_contributor", side_effect=[contributor_obj, duplicate_obj, contributor_obj], ) mocker.patch( "contributor.logic.contributors.artist_info_queries.get", side_effect=[artist_info_fixture, duplicate_artist_info_fixture], ) mocker.patch( "contributor.logic.contributors.rename_art_relations_contributor_records" ) mock_reassign_content = mocker.patch( "contributor.logic.contributors.reassign_contributor_content" ) mocker.patch("contributor.logic.contributors.reassign_artist_info") mocker.patch("contributor.logic.contributors.remove_artist_info_references") mocker.patch("contributor.logic.contributors.neo4j_queries.merge") mocker.patch( "contributor.logic.contributors.neo4j_queries.has_full_catalog_access", return_value=False, ) merge_contributors( session=session, neo4j_session=neo4j_session, contributor_uuid=contributor_uuid, duplicate_uuid=duplicate_uuid, profiles=label_profiles, updated_by="ows-contributor/test-user", identity_uuid=identity_uuid, ) mock_reassign_content.assert_called_once_with( session=session, contributor_artist_info_id=artist_info_fixture["artist_id"], duplicate_artist_info_id=duplicate_artist_info_fixture["artist_id"], )