"""Test for profile.""" from copy import deepcopy from unittest.mock import MagicMock, Mock, patch import pytest from owsresponse import response from pythonfeatures import pythonfeatures from pythonfeatures.constants import split as split_constants from permissions.constants import application, constants, error from permissions.logic import identity as identity_logic, profile from permissions.logic.profile import should_add_moneyhub_subaccount from permissions.models import auth0, label, profile as model, resource as resource_model @patch('permissions.logic.profile.g') @patch('pythonfeatures.pythonfeatures.get_single_feature_by_attributes') def test_map_resources_to_profile_types(mock_features, mock_g, app_context): """Test _map_resources_to_profile_types.""" mock_g.return_value.request_context.side_effect = 'fake_id' mock_response = Mock() mock_response.message = split_constants.FEATURE_ENABLED mock_features.return_value = mock_response resources = [ { 'resource_type': 'Vendor', 'roles': ['accounting', 'analytics', 'marketing', 'audience'], 'uuid': 'u1', }, { 'resource_type': 'Subaccount', 'roles': ['catalog', 'marketing', 'audience'], 'uuid': 'u2', }, {'resource_type': 'LabelParticipant', 'roles': ['analytics'], 'uuid': 'u3'}, {'resource_type': 'Vendor', 'roles': ['analytics'], 'uuid': 'u4'}, {'resource_type': 'LabelParticipant', 'roles': ['accounting'], 'uuid': 'u5'}, {'resource_type': 'Vendor', 'roles': ['accounting'], 'uuid': 'u6'}, {'resource_type': 'Vendor', 'roles': ['songwhip'], 'uuid': 'u7'}, {'resource_type': 'Subaccount', 'roles': ['songwhip'], 'uuid': 'u8'}, {'resource_type': 'Vendor', 'roles': ['songwhip'], 'uuid': 'u9'}, {'resource_type': 'Collaborator', 'roles': ['administrator'], 'uuid': 'u10'}, ] resources_with_brand = [ { 'resource_type': 'Vendor', 'roles': ['accounting', 'analytics', 'marketing', 'audience'], 'uuid': 'u1', 'brand': constants.THEORCHARD_BRAND, 'vendor_id': 123, }, { 'resource_type': 'Subaccount', 'roles': ['catalog', 'marketing', 'audience'], 'uuid': 'u2', 'brand': constants.THEORCHARD_BRAND, 'vendor_id': None, }, { 'resource_type': 'LabelParticipant', 'roles': ['analytics'], 'uuid': 'u3', 'brand': constants.AWAL_BRAND, 'vendor_id': None, }, { 'resource_type': 'Vendor', 'roles': ['analytics', 'marketing', 'audience'], 'uuid': 'u4', 'brand': constants.AWAL_BRAND, 'vendor_id': 789, }, { 'resource_type': 'LabelParticipant', 'roles': ['accounting'], 'uuid': 'u5', 'brand': constants.KNR_BRAND, 'vendor_id': None, }, { 'resource_type': 'Vendor', 'roles': ['accounting'], 'uuid': 'u6', 'brand': constants.KNR_BRAND, 'vendor_id': 987, }, { 'resource_type': 'Vendor', 'roles': ['songwhip'], 'uuid': 'u7', 'brand': constants.THEORCHARD_BRAND, 'vendor_id': 123, }, { 'resource_type': 'Subaccount', 'roles': ['songwhip'], 'uuid': 'u8', 'brand': constants.THEORCHARD_BRAND, 'vendor_id': None, }, { 'resource_type': 'Vendor', 'roles': ['songwhip'], 'uuid': 'u9', 'brand': constants.AWAL_BRAND, 'vendor_id': 789, }, ] with patch.object( resource_model, 'get_resources_brand', return_value=resources_with_brand ), patch.object(label, 'get_migrated_to_abacus', return_value=None): actual = profile._map_resources_to_profile_types(resources) assert actual assert actual.message == { 'all_uuids': ['u1', 'u2', 'u3', 'u4', 'u5', 'u6', 'u7', 'u8', 'u9'], 'profile_roles': { 'CollaboratorsProfile': [ 'accounting', 'analytics', 'marketing', 'audience', 'royalties', ], 'InsightsProfile': [ 'accounting', 'analytics', 'marketing', 'audience', 'analytics', 'analytics', 'marketing', 'audience', ], 'SongwhipProfile': ['songwhip', 'songwhip', 'songwhip'], 'AudienceProfile': [ 'accounting', 'analytics', 'marketing', 'audience', 'catalog', 'marketing', 'audience', 'analytics', 'marketing', 'audience', ], 'MoneyhubProfile': ['accounting', 'accounting'], }, 'resources': [ { 'explicit_profile_types': [ 'AudienceProfile', 'CollaboratorsProfile', 'InsightsProfile', ], 'implicit_profile_types': ['LabelProfile'], 'uuid': 'u1', 'vendor_id': 123, 'resource_type': 'Vendor', 'roles': ['accounting', 'analytics', 'marketing', 'audience'], 'brand': constants.THEORCHARD_BRAND, }, { 'explicit_profile_types': ['AudienceProfile'], 'implicit_profile_types': ['LabelProfile'], 'uuid': 'u2', 'vendor_id': None, 'resource_type': 'Subaccount', 'roles': ['catalog', 'marketing', 'audience'], 'brand': constants.THEORCHARD_BRAND, }, { 'explicit_profile_types': ['InsightsProfile'], 'implicit_profile_types': [], 'uuid': 'u3', 'vendor_id': None, 'resource_type': 'LabelParticipant', 'roles': ['analytics'], 'brand': constants.AWAL_BRAND, }, { 'explicit_profile_types': ['AudienceProfile', 'InsightsProfile'], 'implicit_profile_types': [], 'uuid': 'u4', 'vendor_id': 789, 'resource_type': 'Vendor', 'roles': ['analytics', 'marketing', 'audience'], 'brand': constants.AWAL_BRAND, }, { 'explicit_profile_types': ['MoneyhubProfile'], 'implicit_profile_types': [], 'uuid': 'u5', 'vendor_id': None, 'resource_type': 'LabelParticipant', 'roles': ['accounting'], 'brand': constants.KNR_BRAND, }, { 'explicit_profile_types': ['MoneyhubProfile'], 'implicit_profile_types': ['MoneyhubProfile'], 'uuid': 'u6', 'vendor_id': 987, 'resource_type': 'Vendor', 'roles': ['accounting'], 'brand': constants.KNR_BRAND, }, { 'brand': constants.THEORCHARD_BRAND, 'explicit_profile_types': ['SongwhipProfile'], 'implicit_profile_types': [], 'resource_type': 'Vendor', 'roles': ['songwhip'], 'uuid': 'u7', 'vendor_id': 123, }, { 'brand': constants.THEORCHARD_BRAND, 'explicit_profile_types': ['SongwhipProfile'], 'implicit_profile_types': [], 'resource_type': 'Subaccount', 'roles': ['songwhip'], 'uuid': 'u8', 'vendor_id': None, }, { 'brand': constants.AWAL_BRAND, 'explicit_profile_types': ['SongwhipProfile'], 'implicit_profile_types': [], 'resource_type': 'Vendor', 'roles': ['songwhip'], 'uuid': 'u9', 'vendor_id': 789, }, ], 'aggregate_brand': None, } @patch('permissions.logic.profile.g') @patch('pythonfeatures.pythonfeatures.get_single_feature_by_attributes') def test_map_resources_to_profile_types_with_admin_role(mock_features, mock_g, app_context): """Test _map_resources_to_profile_types.""" mock_g.return_value.request_context.side_effect = 'fake_id' mock_response = Mock() mock_response.message = split_constants.FEATURE_ENABLED mock_features.return_value = mock_response resources = [ {'resource_type': 'Vendor', 'roles': ['administrator'], 'uuid': 'u1'}, {'resource_type': 'Vendor', 'roles': ['administrator'], 'uuid': 'u2'}, {'resource_type': 'Collaborator', 'roles': ['administrator'], 'uuid': 'u3'}, ] resources_with_brand = [ { 'resource_type': 'Vendor', 'roles': ['administrator'], 'uuid': 'u1', 'brand': constants.THEORCHARD_BRAND, 'vendor_id': 1, }, { 'resource_type': 'Vendor', 'roles': ['administrator'], 'uuid': 'u2', 'brand': constants.THEORCHARD_BRAND, 'vendor_id': 2, }, { 'resource_type': 'Collaborator', 'roles': ['administrator'], 'uuid': 'u3', 'brand': constants.THEORCHARD_BRAND, 'vendor_id': 3, }, ] with patch.object( resource_model, 'get_resources_brand', return_value=resources_with_brand ), patch.object(label, 'get_migrated_to_abacus', return_value=False): actual = profile._map_resources_to_profile_types(resources) assert actual assert actual.message == { 'all_uuids': ['u1', 'u2', 'u3'], 'profile_roles': { 'CollaboratorsProfile': [ 'administrator', 'royalties', 'administrator', 'royalties', ], 'InsightsProfile': ['administrator', 'administrator'], 'MoneyhubProfile': ['administrator'], 'DocumentsProfile': ['administrator', 'payee_management'], }, 'resources': [ { 'explicit_profile_types': ['CollaboratorsProfile', 'InsightsProfile'], 'implicit_profile_types': ['LabelProfile'], 'uuid': 'u1', 'resource_type': 'Vendor', 'roles': ['administrator'], 'brand': constants.THEORCHARD_BRAND, 'vendor_id': 1, }, { 'explicit_profile_types': ['CollaboratorsProfile', 'InsightsProfile'], 'implicit_profile_types': ['LabelProfile'], 'uuid': 'u2', 'resource_type': 'Vendor', 'roles': ['administrator'], 'brand': constants.THEORCHARD_BRAND, 'vendor_id': 2, }, { 'brand': constants.THEORCHARD_BRAND, 'explicit_profile_types': ['DocumentsProfile', 'MoneyhubProfile'], 'implicit_profile_types': [], 'resource_type': 'Collaborator', 'roles': ['administrator'], 'uuid': 'u3', 'vendor_id': 3, }, ], 'aggregate_brand': constants.THEORCHARD_BRAND, } @patch('permissions.logic.profile.g') @patch('pythonfeatures.pythonfeatures.get_single_feature_by_attributes') def test_map_resources_to_profile_types_with_subaccount_admin_role( mock_features, mock_g, app_context ): """Test _map_resources_to_profile with moneyhub subaccount.""" mock_g.return_value.request_context.side_effect = 'fake_id' mock_response = Mock() mock_response.message = split_constants.FEATURE_ENABLED mock_features.return_value = mock_response resources = [ {'resource_type': 'Subaccount', 'roles': ['administrator'], 'uuid': 'u4'}, ] resources_with_brand = [ { 'resource_type': 'Subaccount', 'roles': ['administrator'], 'uuid': 'u4', 'brand': constants.THEORCHARD_BRAND, 'vendor_id': 4, }, ] with patch.object( resource_model, 'get_resources_brand', return_value=resources_with_brand ), patch.object(label, 'get_migrated_to_abacus', return_value=True): actual = profile._map_resources_to_profile_types(resources) assert actual assert actual.message == { 'all_uuids': ['u4'], 'profile_roles': { 'InsightsProfile': ['administrator'], 'MoneyhubProfile': ['administrator'], }, 'resources': [ { 'explicit_profile_types': ['InsightsProfile', 'MoneyhubProfile'], 'implicit_profile_types': ['LabelProfile'], 'uuid': 'u4', 'resource_type': 'Subaccount', 'roles': ['administrator'], 'brand': constants.THEORCHARD_BRAND, 'vendor_id': 4, } ], 'aggregate_brand': constants.THEORCHARD_BRAND, } @patch('permissions.logic.profile.g') @patch('pythonfeatures.pythonfeatures.get_single_feature_by_attributes') def test_map_resources_to_profile_types_with_audience_role_migrated( mock_features, mock_g, app_context ): """Test _map_resources_to_profile with migrated vendor audience role.""" mock_g.return_value.request_context.side_effect = 'fake_id' mock_response = Mock() mock_response.message = split_constants.FEATURE_ENABLED mock_features.return_value = mock_response resources = [ {'resource_type': 'Vendor', 'roles': ['audience'], 'uuid': 'u4'}, ] resources_with_brand = [ { 'resource_type': 'Vendor', 'roles': ['audience'], 'uuid': 'u4', 'brand': constants.THEORCHARD_BRAND, 'vendor_id': 4, }, ] with patch.object( resource_model, 'get_resources_brand', return_value=resources_with_brand ), patch.object(label, 'get_migrated_to_abacus', return_value=True): actual = profile._map_resources_to_profile_types(resources) assert actual assert actual.message == { 'all_uuids': ['u4'], 'profile_roles': { 'AudienceProfile': ['audience'], }, 'resources': [ { 'explicit_profile_types': ['AudienceProfile'], 'implicit_profile_types': [], 'uuid': 'u4', 'resource_type': 'Vendor', 'roles': ['audience'], 'brand': constants.THEORCHARD_BRAND, 'vendor_id': 4, } ], 'aggregate_brand': constants.THEORCHARD_BRAND, } @pytest.mark.parametrize( ('errors', 'message', 'should_add_moneyhub_subaccount_response', 'feature_flag'), [ pytest.param( True, None, True, split_constants.FEATURE_DISABLED, id='Returns error response due to feature flag being off and Subaccount being sent.', ), pytest.param( False, { 'all_uuids': ['u1', 'u2'], 'profile_roles': { 'MoneyhubProfile': ['accounting', 'accounting'], }, 'resources': [ { 'explicit_profile_types': ['MoneyhubProfile'], 'implicit_profile_types': [], 'uuid': 'u1', 'resource_type': 'Subaccount', 'roles': ['accounting'], 'brand': constants.THEORCHARD_BRAND, 'vendor_id': 1, }, { 'explicit_profile_types': ['MoneyhubProfile'], 'implicit_profile_types': [], 'uuid': 'u2', 'resource_type': 'Subaccount', 'roles': ['accounting'], 'brand': constants.AWAL_BRAND, 'vendor_id': 2, }, ], 'aggregate_brand': None, }, True, split_constants.FEATURE_ENABLED, id='Return profiles due to feature flag being on and Subaccount being sent.', ), ], ) @patch('permissions.logic.profile.g') @patch('pythonfeatures.pythonfeatures.get_single_feature_by_attributes') @patch('permissions.logic.profile.should_add_moneyhub_subaccount') def test_map_resources_to_profile_types_with_accounting_role_subaccount( mock_subaccount_feature_enabled, mock_features, mock_g, errors, message, should_add_moneyhub_subaccount_response, feature_flag, app_context, ): """Test _map_resources_to_profile_types.""" mock_g.return_value.request_context.side_effect = 'fake_id' mock_response = Mock() mock_response.message = feature_flag mock_features.return_value = mock_response resources = [ {'resource_type': 'Subaccount', 'roles': ['accounting'], 'uuid': 'u1'}, {'resource_type': 'Subaccount', 'roles': ['accounting'], 'uuid': 'u2'}, ] resources_with_brand = [ { 'resource_type': 'Subaccount', 'roles': ['accounting'], 'uuid': 'u1', 'brand': constants.THEORCHARD_BRAND, 'vendor_id': 1, }, { 'resource_type': 'Subaccount', 'roles': ['accounting'], 'uuid': 'u2', 'brand': constants.AWAL_BRAND, 'vendor_id': 2, }, ] mock_subaccount_feature_enabled.return_value = should_add_moneyhub_subaccount_response with patch.object( resource_model, 'get_resources_brand', return_value=resources_with_brand ), patch.object(label, 'get_migrated_to_abacus', return_value=True): actual = profile._map_resources_to_profile_types(resources) assert bool(actual.errors) == errors assert actual.message == message @patch('permissions.logic.profile.g') @patch('pythonfeatures.pythonfeatures.get_single_feature_by_attributes') def test_map_resources_to_profile_types_awal(mock_features, mock_g, app_context): """Test _map_resources_to_profile_types with awal brand.""" mock_g.return_value.request_context.side_effect = 'fake_id' mock_response = Mock() mock_response.message = split_constants.FEATURE_ENABLED mock_features.return_value = mock_response resources = [ { 'resource_type': 'Vendor', 'roles': ['accounting', 'analytics', 'catalog', 'marketing', 'songwhip', 'audience'], 'id': 7123, 'uuid': 'u1', } ] resources_with_brand = [ { 'resource_type': 'Vendor', 'roles': ['accounting', 'analytics', 'catalog', 'marketing', 'songwhip', 'audience'], 'vendor_id': 7123, 'uuid': 'u1', 'brand': constants.AWAL_BRAND, } ] with patch.object( resource_model, 'get_resources_brand', return_value=resources_with_brand ), patch.object(label, 'get_migrated_to_abacus', return_value=False): actual = profile._map_resources_to_profile_types(resources, master_contact=True) assert actual assert actual.message == { 'all_uuids': ['u1'], 'profile_roles': { 'AudienceProfile': [ 'accounting', 'analytics', 'catalog', 'marketing', 'songwhip', 'audience', ], 'CollaboratorsProfile': [ 'accounting', 'analytics', 'catalog', 'marketing', 'songwhip', 'audience', 'royalties', ], 'InsightsProfile': [ 'accounting', 'analytics', 'catalog', 'marketing', 'songwhip', 'audience', ], 'MoneyhubProfile': [ 'accounting', 'analytics', 'catalog', 'marketing', 'songwhip', 'audience', ], 'SongwhipProfile': [ 'accounting', 'analytics', 'catalog', 'marketing', 'songwhip', 'audience', ], }, 'resources': [ { 'explicit_profile_types': [ 'AudienceProfile', 'CollaboratorsProfile', 'InsightsProfile', 'MoneyhubProfile', 'SongwhipProfile', ], 'implicit_profile_types': ['LabelProfile'], 'uuid': 'u1', 'vendor_id': 7123, 'resource_type': 'Vendor', 'roles': [ 'accounting', 'analytics', 'catalog', 'marketing', 'songwhip', 'audience', ], 'brand': constants.AWAL_BRAND, }, ], 'aggregate_brand': constants.AWAL_BRAND, } @patch('permissions.logic.profile.g') @patch('pythonfeatures.pythonfeatures.get_single_feature_by_attributes') def test_map_resources_to_profile_types_awal_with_admin_role(mock_features, mock_g, app_context): """Test _map_resources_to_profile_types.""" mock_g.return_value.request_context.side_effect = 'fake_id' mock_response = Mock() mock_response.message = split_constants.FEATURE_ENABLED mock_features.return_value = mock_response resources = [ {'resource_type': 'Vendor', 'roles': ['administrator'], 'uuid': 'u1'}, ] resources_with_brand = [ { 'resource_type': 'Vendor', 'roles': ['administrator'], 'uuid': 'u1', 'brand': constants.AWAL_BRAND, 'vendor_id': 1, } ] with patch.object( resource_model, 'get_resources_brand', return_value=resources_with_brand ), patch.object(label, 'get_migrated_to_abacus', return_value=None): actual = profile._map_resources_to_profile_types(resources) assert actual assert actual.message == { 'all_uuids': ['u1'], 'profile_roles': { 'CollaboratorsProfile': ['administrator', 'royalties'], 'DocumentsProfile': ['administrator'], 'InsightsProfile': ['administrator'], 'MoneyhubProfile': ['administrator'], }, 'resources': [ { 'explicit_profile_types': [ 'CollaboratorsProfile', 'DocumentsProfile', 'InsightsProfile', 'MoneyhubProfile', ], 'implicit_profile_types': ['LabelProfile'], 'uuid': 'u1', 'vendor_id': 1, 'resource_type': 'Vendor', 'roles': ['administrator'], 'brand': constants.AWAL_BRAND, } ], 'aggregate_brand': constants.AWAL_BRAND, } @patch('permissions.logic.profile.g') @patch('pythonfeatures.pythonfeatures.get_single_feature_by_attributes') def test_map_resources_to_profile_types_knr_with_admin_role(mock_features, mock_g, app_context): """Test _map_resources_to_profile_types.""" mock_g.return_value.request_context.side_effect = 'fake_id' mock_response = Mock() mock_response.message = split_constants.FEATURE_ENABLED mock_features.return_value = mock_response resources = [ {'resource_type': 'Vendor', 'roles': ['administrator'], 'uuid': 'u1'}, ] resources_with_brand = [ { 'resource_type': 'Vendor', 'roles': ['administrator'], 'uuid': 'u1', 'brand': constants.KNR_BRAND, 'vendor_id': 123, } ] with patch.object( resource_model, 'get_resources_brand', return_value=resources_with_brand ), patch.object(label, 'get_migrated_to_abacus', return_value=False): actual = profile._map_resources_to_profile_types(resources) assert actual assert actual.message == { 'all_uuids': ['u1'], 'profile_roles': { 'MoneyhubProfile': ['administrator'], }, 'resources': [ { 'explicit_profile_types': ['MoneyhubProfile'], 'implicit_profile_types': ['MoneyhubProfile'], 'uuid': 'u1', 'vendor_id': 123, 'resource_type': 'Vendor', 'roles': ['administrator'], 'brand': constants.KNR_BRAND, } ], 'aggregate_brand': constants.KNR_BRAND, } @patch('permissions.logic.profile.g') @patch('pythonfeatures.pythonfeatures.get_single_feature_by_attributes') def test_map_resources_to_profile_types_awal_with_admin_role_and_master_contact( mock_features, mock_g, app_context ): """Test _map_resources_to_profile_types.""" mock_g.return_value.request_context.side_effect = 'fake_id' mock_response = Mock() mock_response.message = split_constants.FEATURE_ENABLED mock_features.return_value = mock_response resources = [ {'resource_type': 'Vendor', 'roles': ['administrator'], 'uuid': 'u1'}, ] resources_with_brand = [ { 'resource_type': 'Vendor', 'roles': ['administrator'], 'uuid': 'u1', 'brand': constants.AWAL_BRAND, 'vendor_id': 123, } ] with patch.object( resource_model, 'get_resources_brand', return_value=resources_with_brand ), patch.object(label, 'get_migrated_to_abacus', return_value=None): actual = profile._map_resources_to_profile_types(resources, master_contact=True) assert actual assert actual.message == { 'all_uuids': ['u1'], 'profile_roles': { 'CollaboratorsProfile': ['administrator', 'royalties'], 'DocumentsProfile': ['administrator'], 'InsightsProfile': ['administrator'], 'MoneyhubProfile': ['administrator'], }, 'resources': [ { 'explicit_profile_types': [ 'CollaboratorsProfile', 'DocumentsProfile', 'InsightsProfile', 'MoneyhubProfile', ], 'implicit_profile_types': ['LabelProfile'], 'uuid': 'u1', 'vendor_id': 123, 'resource_type': 'Vendor', 'roles': ['administrator'], 'brand': constants.AWAL_BRAND, } ], 'aggregate_brand': constants.AWAL_BRAND, } # note: in following tests, the brand property has been added to the resource input list # for the sake of brevity. In production, this property is not passed from the front end, # but is added via a neo4j lookup in the _map_resources_to_profile_types method tested above. @patch('permissions.logic.profile.g') @patch('pythonfeatures.pythonfeatures.get_single_feature_by_attributes') def test_map_resources_to_profile_types_for_sme_brand(mock_features, mock_g, app_context): """Test _map_resources_to_profile_types and check aggregate_brand.""" mock_g.return_value.request_context.side_effect = 'fake_id' mock_response = Mock() mock_response.message = split_constants.FEATURE_ENABLED mock_features.return_value = mock_response resources = [ {'resource_type': 'Vendor', 'roles': ['administrator'], 'uuid': 'u1'}, ] resources_with_brand = [ { 'resource_type': 'Vendor', 'roles': ['administrator'], 'uuid': 'u1', 'brand': constants.SONY_BRAND, 'vendor_id': 345, } ] with patch.object( resource_model, 'get_resources_brand', return_value=resources_with_brand ), patch.object(label, 'get_migrated_to_abacus', return_value=None): actual = profile._map_resources_to_profile_types(resources, master_contact=True) assert actual assert actual.message == { 'all_uuids': ['u1'], 'profile_roles': { 'CollaboratorsProfile': ['administrator', 'royalties'], 'InsightsProfile': ['administrator'], }, 'resources': [ { 'explicit_profile_types': ['CollaboratorsProfile', 'InsightsProfile'], 'implicit_profile_types': ['LabelProfile'], 'uuid': 'u1', 'vendor_id': 345, 'resource_type': 'Vendor', 'roles': ['administrator'], 'brand': constants.SONY_BRAND, } ], 'aggregate_brand': constants.SONY_BRAND, } @patch('permissions.logic.profile.g', spec=['request_context', 'log']) @pytest.mark.parametrize( ('edit_super_admins_enabled', 'identity_email', 'expected'), [ pytest.param( response.Response(message='enabled'), 'foo@sonymusic-pde.com', response.Response(message={'profiles_affected': [], 'identity_created': False}), id=""" Successfully edited because user has vendor star and target identity is allowed vendor * """, ), pytest.param( response.Response(message='enabled'), 'foo@sonymusic-pde.com', response.Response( message={ 'profiles_affected': [], 'identities_affected': [{'id': '1234abcd'}], 'identity_created': True, }, ), id=""" Successfully created because user has vendor star and target identity is allowed vendor * """, ), pytest.param( response.Response(message='enabled'), 'foo@bar.com', response.create_error_response( error.ERROR_CODE_BAD_REQUEST, error.ERROR_MESSAGE_CANNOT_ASSIGN_RESOURCE, status=403, ), id='Fails because target identity is not allowed vendor *', ), pytest.param( response.Response(message='control'), 'doesnotmatter@sonymusic-pde.com', response.create_error_response( error.ERROR_CODE_AUTHORIZATION, error.ERROR_MESSAGE_FORBIDDEN_USER ), id='Fails because user does not have vendor * access', ), ], ) def test_create_identity_profiles_resources_super_admin( mock_g: MagicMock, edit_super_admins_enabled: response.Response, identity_email: str, expected: response.Response, ) -> None: """Test create_identity_profiles_resources handling of vendor *.""" resources = [ { 'resource_type': 'Vendor', 'roles': ['analytics'], 'uuid': constants.VENDOR_STAR_UUID, 'brand': constants.THEORCHARD_BRAND, 'vendor_id': '*', }, { 'resource_type': 'Subaccount', 'roles': ['catalog'], 'uuid': 8869, 'brand': constants.THEORCHARD_BRAND, 'vendor_id': 69, }, { 'resource_type': 'LabelParticipant', 'roles': ['analytics'], 'uuid': 8869, 'brand': constants.THEORCHARD_BRAND, 'vendor_id': 88, }, ] admin_context = { 'identity_id': 'admin-uuid', 'profile_type': 'SettingsProfile', 'profile_id': 12, } identity = {'name': 'foo@bar.com', 'email': identity_email, 'id': '1234abcd'} default_brand = response.Response(message={'default_brand': constants.THEORCHARD_BRAND}) mock_g.request_context.context_type = 'profile' mock_g.request_context.profile_type = 'SettingsProfile' with patch.object(resource_model, 'get_resources_brand', return_value=resources), patch.object( label, 'get_migrated_to_abacus', return_value=None ), patch.object(model, 'check_vendor_star_access', return_value=True), patch.object( auth0, 'update_user_metadata' ), patch.object(identity_logic, 'get_identity_by_id', return_value=default_brand), patch.object( pythonfeatures, 'get_single_feature', return_value=edit_super_admins_enabled ), patch.object(model, 'add_resources_to_identity', return_value=expected): actual = profile.create_identity_profiles_resources(admin_context, identity, resources) assert actual.message == expected.message assert actual.status == expected.status @patch('permissions.logic.profile.g') @pytest.mark.parametrize( ('admin_identity',), [ ( { 'email': 'admin@bar.com', 'id': 'admin-uuid', 'default_brand': constants.THEORCHARD_BRAND, }, ), ({'email': 'admin@bar.com', 'id': 'admin-uuid', 'default_brand': constants.AWAL_BRAND},), ({'email': 'admin@bar.com', 'id': 'admin-uuid'},), ], ) def test_create_identity_profiles_resources(mock_g, app_context, admin_identity): """Test create_identity_profiles_resources.""" resources = [ { 'resource_type': 'Vendor', 'roles': ['accounting', 'analytics'], 'uuid': 7123, 'brand': constants.THEORCHARD_BRAND, 'vendor_id': 1, }, { 'resource_type': 'Subaccount', 'roles': ['catalog'], 'uuid': 8869, 'brand': constants.THEORCHARD_BRAND, 'vendor_id': 2, }, { 'resource_type': 'LabelParticipant', 'roles': ['analytics'], 'uuid': 8869, 'brand': constants.THEORCHARD_BRAND, 'vendor_id': 3, }, ] admin_context = { 'identity_id': 'admin-uuid', 'profile_type': 'SettingsProfile', 'profile_id': 12, } identity = {'name': 'foo@bar.com', 'email': 'foo@bar.com', 'id': '1234abcd'} add_result = response.Response(message={'profiles_affected': [], 'identity_created': False}) edit_super_admins = response.Response(message='control') mock_g.request_context.context_type = 'profile' mock_g.request_context.profile_type = 'SettingsProfile' with patch.object(resource_model, 'get_resources_brand', return_value=resources), patch.object( label, 'get_migrated_to_abacus', return_value=None ), patch.object(model, 'check_vendor_star_access', return_value=True), patch.object( auth0, 'update_user_metadata' ), patch.object( identity_logic, 'get_identity_by_id', return_value=response.Response(admin_identity), ), patch.object( pythonfeatures, 'get_single_feature', return_value=edit_super_admins ), patch.object(model, 'add_resources_to_identity', return_value=add_result): actual = profile.create_identity_profiles_resources( admin_context=admin_context, identity=identity, resource_access=resources ) assert actual args = model.add_resources_to_identity.call_args assert args.kwargs['identity']['id'] == identity['id'] # identity param assert args.kwargs['identity']['user_types'] == ['label', 'artist'] assert ( args.kwargs['brand'] == admin_identity.get('default_brand') or constants.THEORCHARD_BRAND ) @patch('permissions.logic.profile.g') @pytest.mark.parametrize( ('has_access_result', 'admin_identity', 'expected'), [ ( response.Response(), { 'email': 'admin@bar.com', 'id': 'admin-uuid', 'default_brand': constants.THEORCHARD_BRAND, }, response.Response(message={'profiles_affected': [], 'identity_created': False}), ), ( response.Response(), { 'email': 'admin@bar.com', 'id': 'admin-uuid', 'default_brand': constants.THEORCHARD_BRAND, }, response.Response(message={'profiles_affected': [], 'identity_created': False}), ), ( response.Response(), {'email': 'admin@bar.com', 'id': 'admin-uuid'}, response.Response(message={'profiles_affected': [], 'identity_created': False}), ), ( response.create_fatal_response('Admin does not have access to all the resources'), {'email': 'admin@bar.com', 'id': 'admin-uuid'}, response.create_fatal_response('Admin does not have access to all the resources'), ), ], ) def test_create_identity_profiles_resources_regular_admin( mock_g, app_context, has_access_result, admin_identity, expected ): """Test create_identity_profiles_resources.""" resources = [ { 'resource_type': 'Vendor', 'roles': ['accounting', 'analytics'], 'uuid': 7123, 'brand': constants.THEORCHARD_BRAND, 'vendor_id': 1, }, { 'resource_type': 'Subaccount', 'roles': ['catalog'], 'uuid': 8869, 'brand': constants.THEORCHARD_BRAND, 'vendor_id': 2, }, { 'resource_type': 'LabelParticipant', 'roles': ['analytics'], 'uuid': 8869, 'brand': constants.THEORCHARD_BRAND, 'vendor_id': 3, }, ] admin_context = { 'identity_id': 'admin-uuid', 'profile_type': 'SettingsProfile', 'profile_id': 12, } identity = {'name': 'foo@bar.com', 'email': 'foo@bar.com', 'id': '1234abcd'} add_result = response.Response(message={'profiles_affected': [], 'identity_created': False}) edit_super_admins = response.Response(message='control') mock_g.request_context.context_type = 'profile' mock_g.request_context.profile_type = 'SettingsProfile' with patch.object(resource_model, 'get_resources_brand', return_value=resources), patch.object( label, 'get_migrated_to_abacus', return_value=None ), patch.object(model, 'check_vendor_star_access', return_value=False), patch.object( auth0, 'update_user_metadata' ), patch.object( identity_logic, 'get_identity_by_id', return_value=response.Response(admin_identity), ), patch.object( model, 'has_admin_access_to_resources', return_value=has_access_result ), patch.object( pythonfeatures, 'get_single_feature', return_value=edit_super_admins ), patch.object(model, 'add_resources_to_identity', return_value=add_result): actual = profile.create_identity_profiles_resources(admin_context, identity, resources) assert actual.status == expected.status assert model.has_admin_access_to_resources.called if actual: assert model.add_resources_to_identity.called args = model.add_resources_to_identity.call_args assert args.kwargs['overwrite_existing_access'] is True assert args.kwargs['brand'] == constants.THEORCHARD_BRAND assert args.kwargs['master_contact'] is False @patch('permissions.logic.profile.g') def test_create_identity_by_regular_awal_admin(mock_g, app_context): """Test create_identity_profiles_resources by an AWAL Admin.""" has_access_result = response.Response() admin_identity = { 'email': 'admin@bar.com', 'id': 'admin-uuid', 'default_brand': constants.AWAL_BRAND, } resources = [ { 'resource_type': 'Vendor', 'roles': ['accounting', 'analytics'], 'uuid': 7123, 'brand': constants.AWAL_BRAND, 'vendor_id': 23, }, { 'resource_type': 'LabelParticipant', 'roles': ['analytics'], 'uuid': 8869, 'brand': constants.AWAL_BRAND, 'vendor_id': 69, }, ] admin_context = { 'identity_id': 'admin-uuid', 'profile_type': 'SettingsProfile', 'profile_id': 12, } identity = {'name': 'foo@bar.com', 'email': 'foo@bar.com', 'id': '1234abcd'} add_result = response.Response(message={'profiles_affected': [], 'identity_created': False}) edit_super_admins = response.Response(message='control') mock_g.request_context.context_type = 'profile' mock_g.request_context.profile_type = 'SettingsProfile' with patch.object(resource_model, 'get_resources_brand', return_value=resources), patch.object( label, 'get_migrated_to_abacus', return_value=None ), patch.object(model, 'check_vendor_star_access', return_value=False), patch.object( auth0, 'update_user_metadata' ), patch.object( identity_logic, 'get_identity_by_id', return_value=response.Response(admin_identity), ), patch.object( model, 'has_admin_access_to_resources', return_value=has_access_result ), patch.object( pythonfeatures, 'get_single_feature', return_value=edit_super_admins ), patch.object(model, 'add_resources_to_identity', return_value=add_result): actual = profile.create_identity_profiles_resources(admin_context, identity, resources) assert actual.status assert model.has_admin_access_to_resources.called assert model.add_resources_to_identity.called args = model.add_resources_to_identity.call_args assert args.kwargs['admin_context'] == admin_context assert args.kwargs['can_access_vendor_star'] is False assert args.kwargs['set_email_verified'] is False assert args.kwargs['overwrite_existing_access'] is True assert args.kwargs['brand'] == constants.AWAL_BRAND assert args.kwargs['create_auth0_user'] is False assert args.kwargs['master_contact'] is False @patch('permissions.logic.profile.g') def test_create_identity_profiles_resources_artist_only(mock_g, app_context): """Test create_identity_profiles_resources.""" resources = [ { 'resource_type': 'LabelParticipant', 'roles': ['analytics'], 'uuid': 'uuid1', 'brand': constants.AWAL_BRAND, 'vendor_id': 1, }, { 'resource_type': 'LabelParticipant', 'roles': ['analytics'], 'uuid': 'uuid2', 'brand': constants.AWAL_BRAND, 'vendor_id': 2, }, ] admin_context = { 'identity_id': 'admin-uuid', 'profile_type': 'SettingsProfile', 'profile_id': 12, } identity = {'name': 'foo@bar.com', 'email': 'foo@bar.com', 'id': '1234abcd'} labels = response.Response([{'label': 'names'}]) expected_identity = deepcopy(identity) expected_identity['app_metadata'] = { 'should_send_welcome_as_password_reset': True, 'label_participants': labels.message, } add_result = response.Response(message={'profiles_affected': [], 'identity_created': False}) edit_super_admins = response.Response(message='control') mock_g.request_context.context_type = 'profile' mock_g.request_context.profile_type = 'SettingsProfile' with patch.object(resource_model, 'get_resources_brand', return_value=resources), patch.object( label, 'get_migrated_to_abacus', return_value=None ), patch.object(model, 'check_vendor_star_access', return_value=True), patch.object( model, 'get_label_for_label_participant', return_value=labels ), patch.object(auth0, 'update_user_metadata'), patch.object( pythonfeatures, 'get_single_feature', return_value=edit_super_admins ), patch.object(model, 'add_resources_to_identity', return_value=add_result): actual = profile.create_identity_profiles_resources( admin_context, expected_identity, resources, brand=constants.AWAL_BRAND ) assert actual args = model.add_resources_to_identity.call_args assert args.kwargs['identity']['id'] == identity['id'] # identity param assert args.kwargs['identity']['user_types'] == ['artist'] assert args.kwargs['brand'] == constants.AWAL_BRAND # brand param @patch('permissions.logic.profile.g') def test_create_identity_profiles_resources_invitation(mock_g, app_context): """Test create_identity_profiles_resources.""" resources = [ { 'resource_type': 'Vendor', 'roles': ['administrator'], 'uuid': 'uuid1', 'brand': constants.AWAL_BRAND, 'vendor_id': 123, } ] admin_context = { 'identity_id': 'admin-uuid', 'profile_type': 'SettingsProfile', 'profile_id': 12, } identity = {'name': 'foo@bar.com', 'email': 'foo@bar.com', 'id': '1234abcd'} add_result = response.Response(message={'profiles_affected': [], 'identity_created': False}) mock_g.request_context.context_type = 'profile' mock_g.request_context.profile_type = 'SettingsProfile' edit_super_admins = response.Response(message='control') with patch.object(resource_model, 'get_resources_brand', return_value=resources), patch.object( label, 'get_migrated_to_abacus', return_value=None ), patch.object(model, 'check_vendor_star_access', return_value=True), patch.object( auth0, 'update_user_metadata' ), patch.object(auth0, 'bulk_send_password_reset'), patch.object( pythonfeatures, 'get_single_feature', return_value=edit_super_admins ), patch.object(model, 'add_resources_to_identity', return_value=add_result): actual = profile.create_identity_profiles_resources( admin_context, identity, resources, brand=constants.AWAL_BRAND, create_auth0_user=False ) assert actual assert model.add_resources_to_identity.call_count == 1 args = model.add_resources_to_identity.call_args assert args.kwargs['identity']['id'] == identity['id'] # identity param assert args.kwargs['identity']['user_types'] == ['label'] assert args.kwargs['brand'] == constants.AWAL_BRAND # brand param assert auth0.update_user_metadata.call_count == 0 assert auth0.bulk_send_password_reset.call_count == 0 @patch('permissions.logic.profile.g') @pytest.mark.parametrize( ('resources', 'expected_identity_brand'), [ ( # Expected behavior: Read brand from resource and set in args without modifying # TestCase 1: Brand 'theorchard' [ { 'resource_type': 'Vendor', 'roles': ['accounting', 'analytics'], 'uuid': 7123, 'brand': 'theorchard', 'vendor_id': 23, } ], constants.THEORCHARD_BRAND, ), ( # TestCase 2: Brand 'orchard' [ { 'resource_type': 'Vendor', 'roles': ['accounting', 'analytics'], 'uuid': 7123, 'brand': constants.MASS_APPEAL_BRAND, 'vendor_id': 12, } ], constants.THEORCHARD_BRAND, ), ( # TestCase 3: Brand 'orchard' with Subaccount [ { 'resource_type': 'Subaccount', 'roles': ['accounting', 'analytics'], 'uuid': 7123, 'brand': constants.MASS_APPEAL_BRAND, 'vendor_id': 12, } ], constants.THEORCHARD_BRAND, ), ( # TestCase 4: Brand 'orchard' with Subaccount but flag off [ { 'resource_type': 'Subaccount', 'roles': ['analytics'], 'uuid': 7123, 'brand': constants.MASS_APPEAL_BRAND, 'vendor_id': 12, } ], constants.THEORCHARD_BRAND, ), ( # TestCase 5: Brand 'awal' [ { 'resource_type': 'Vendor', 'roles': ['accounting', 'analytics'], 'uuid': 7123, 'brand': constants.AWAL_BRAND, 'vendor_id': 71, } ], constants.AWAL_BRAND, ), ], ) @patch('pythonfeatures.pythonfeatures.get_single_feature_by_attributes') def test_create_identity_profiles_resources_brand_orchard_sets_brand( mock_features, mock_g, resources, expected_identity_brand, app_context ): """Test create_identity_profiles_resources sets brand to orchard when it is theorchard.""" has_access_result = response.Response() mock_feature = MagicMock() mock_feature.message = split_constants.FEATURE_ENABLED mock_features.return_value = mock_feature admin_identity = {'email': 'admin@bar.com', 'id': 'admin-uuid', 'default_brand': None} admin_context = { 'identity_id': 'admin-uuid', 'profile_type': 'SettingsProfile', 'profile_id': 12, } identity = {'name': 'foo@bar.com', 'email': 'foo@bar.com', 'id': '1234abcd'} add_result = response.Response(message={'profiles_affected': [], 'identity_created': False}) edit_super_admins = response.Response(message='control') mock_g.request_context.context_type = 'profile' mock_g.request_context.profile_type = 'SettingsProfile' with patch.object(resource_model, 'get_resources_brand', return_value=resources), patch.object( label, 'get_migrated_to_abacus', return_value=None ), patch.object(model, 'check_vendor_star_access', return_value=False), patch.object( auth0, 'update_user_metadata' ), patch.object( identity_logic, 'get_identity_by_id', return_value=response.Response(admin_identity), ), patch.object( model, 'has_admin_access_to_resources', return_value=has_access_result ), patch.object( pythonfeatures, 'get_single_feature', return_value=edit_super_admins ), patch.object(model, 'add_resources_to_identity', return_value=add_result): actual = profile.create_identity_profiles_resources(admin_context, identity, resources) assert actual.status assert model.has_admin_access_to_resources.called assert model.add_resources_to_identity.called args = model.add_resources_to_identity.call_args assert args.kwargs['brand'] assert args.kwargs['admin_context'] == admin_context assert args.kwargs['brand'] == expected_identity_brand assert mock_features.called @patch('permissions.logic.profile.g') def test_create_identity_profiles_resources_brand_orchard(mock_g, app_context): """Test create_identity_profiles_resources wont create auth0 identity when FF + theorchard.""" has_access_result = response.Response() admin_identity = { 'email': 'admin@bar.com', 'id': 'admin-uuid', 'default_brand': constants.THEORCHARD_BRAND, } resources = [ { 'resource_type': 'Vendor', 'roles': ['accounting', 'analytics'], 'uuid': 7123, 'brand': constants.THEORCHARD_BRAND, 'vendor_id': 12345, } ] admin_context = { 'identity_id': 'admin-uuid', 'profile_type': 'SettingsProfile', 'profile_id': 12, } identity = {'name': 'foo@bar.com', 'email': 'foo@bar.com', 'id': '1234abcd'} add_result = response.Response(message={'profiles_affected': [], 'identity_created': False}) mock_g.request_context.context_type = 'profile' mock_g.request_context.profile_type = 'SettingsProfile' with patch.object(resource_model, 'get_resources_brand', return_value=resources), patch.object( label, 'get_migrated_to_abacus', return_value=None ), patch.object(model, 'check_vendor_star_access', return_value=False), patch.object( auth0, 'update_user_metadata' ), patch.object(auth0, 'bulk_send_password_reset'), patch.object( identity_logic, 'get_identity_by_id', return_value=response.Response(admin_identity), ), patch.object( model, 'has_admin_access_to_resources', return_value=has_access_result ), patch.object( pythonfeatures, 'get_single_feature', return_value=response.Response(message='control'), # superadmin flag ), patch.object(model, 'add_resources_to_identity', return_value=add_result): actual = profile.create_identity_profiles_resources(admin_context, identity, resources) assert actual.status assert model.has_admin_access_to_resources.called assert model.add_resources_to_identity.called add_res_args = model.add_resources_to_identity.call_args assert add_res_args.kwargs['brand'] == 'theorchard' assert add_res_args.kwargs['create_auth0_user'] is False assert add_res_args.kwargs['set_email_verified'] is False assert auth0.bulk_send_password_reset.call_count == 0 @patch('permissions.logic.profile.g') def test_create_identity_profiles_resources_brand_mass_appeal(mock_g, app_context): """Test create_identity_profiles_resources sets brand to orchard when resource brand is ma.""" has_access_result = response.Response() admin_identity = { 'email': 'admin@bar.com', 'id': 'admin-uuid', 'default_brand': constants.THEORCHARD_BRAND, } resources = [ { 'resource_type': 'Vendor', 'roles': ['accounting', 'analytics'], 'uuid': 7123, 'brand': constants.MASS_APPEAL_BRAND, 'vendor_id': 12345, } ] admin_context = { 'identity_id': 'admin-uuid', 'profile_type': 'SettingsProfile', 'profile_id': 12, } identity = {'name': 'foo@bar.com', 'email': 'foo@bar.com', 'id': '1234abcd'} add_result = response.Response(message={'profiles_affected': [], 'identity_created': False}) mock_g.request_context.context_type = 'profile' mock_g.request_context.profile_type = 'SettingsProfile' with patch.object(resource_model, 'get_resources_brand', return_value=resources), patch.object( label, 'get_migrated_to_abacus', return_value=None ), patch.object(model, 'check_vendor_star_access', return_value=False), patch.object( auth0, 'update_user_metadata' ), patch.object(auth0, 'bulk_send_password_reset'), patch.object( identity_logic, 'get_identity_by_id', return_value=response.Response(admin_identity), ), patch.object( model, 'has_admin_access_to_resources', return_value=has_access_result ), patch.object( pythonfeatures, 'get_single_feature', return_value=response.Response(message='control'), # superadmin flag ), patch.object(model, 'add_resources_to_identity', return_value=add_result): actual = profile.create_identity_profiles_resources(admin_context, identity, resources) assert actual.status assert model.has_admin_access_to_resources.called assert model.add_resources_to_identity.called add_res_args = model.add_resources_to_identity.call_args assert add_res_args.kwargs['brand'] == 'theorchard' assert add_res_args.kwargs['create_auth0_user'] is False assert add_res_args.kwargs['set_email_verified'] is False assert auth0.bulk_send_password_reset.call_count == 0 @patch('permissions.logic.profile.g') @pytest.mark.parametrize( ('edit_super_admins_enabled', 'identity_email', 'expected'), [ pytest.param( response.Response(message='enabled'), 'foo@sonymusic.com', response.Response( message={ 'name': 'foo@bar.com', 'email': 'foo@sonymusic.com', 'id': 'uuid', 'user_types': ['label', 'artist'], 'all_brands': [constants.THEORCHARD_BRAND], } ), id='Succeeds because both user and the target identity are allowed vendor * access', ), pytest.param( response.Response(message='enabled'), 'foo@bar.com', response.create_error_response( error.ERROR_CODE_BAD_REQUEST, 'somebar', status=403, ), id='Fails because the target identity is not allowed vendor * access', ), pytest.param( response.Response(message='control'), 'doesntmatter@sonymusic.com', response.create_error_response( error.ERROR_CODE_AUTHORIZATION, error.ERROR_MESSAGE_FORBIDDEN_USER, ), id='Fails because user does not have vendor * access', ), ], ) def test_edit_identity_profiles_resources_super_admin( mock_g, app_context, edit_super_admins_enabled, identity_email, expected ): """Test edit_identity_profiles_resources handling of vendor *.""" resources = [ { 'resource_type': 'Vendor', 'roles': ['analytics'], 'uuid': constants.VENDOR_STAR_UUID, 'brand': constants.THEORCHARD_BRAND, 'vendor_id': '*', }, { 'resource_type': 'Subaccount', 'roles': ['catalog'], 'uuid': 8869, 'brand': constants.THEORCHARD_BRAND, 'vendor_id': 69, }, { 'resource_type': 'LabelParticipant', 'roles': ['analytics'], 'uuid': 8869, 'brand': constants.THEORCHARD_BRAND, 'vendor_id': 12, }, ] admin_context = { 'identity_id': 'admin-uuid', 'profile_type': 'SettingsProfile', 'profile_id': 12, } identity_id = 'uuid' identity = response.Response( message={'name': 'foo@bar.com', 'email': identity_email, 'id': identity_id} ) add_result = response.Response(message={'profiles_affected': [], 'identity_created': False}) mock_g.request_context.context_type = 'profile' mock_g.request_context.profile_type = 'SettingsProfile' with patch.object(resource_model, 'get_resources_brand', return_value=resources), patch.object( label, 'get_migrated_to_abacus', return_value=None ), patch.object(identity_logic, 'get_identity_by_id', return_value=identity), patch.object( model, 'check_vendor_star_access', return_value=True ), patch.object(auth0, 'update_user_metadata'), patch.object( model, 'add_resources_to_identity', return_value=add_result ), patch.object(pythonfeatures, 'get_single_feature', return_value=edit_super_admins_enabled): actual = profile.edit_identity_profiles_resources(admin_context, identity_id, resources) assert actual.message == expected.message assert actual.status == expected.status @patch('permissions.logic.profile.g') def test_edit_identity_profiles_resources(mock_g, app_context): """Test edit_identity_profiles_resources.""" overwrite_existing_access = False resources = [ { 'resource_type': 'LabelParticipant', 'roles': ['analytics'], 'uuid': 8869, 'brand': constants.THEORCHARD_BRAND, 'vendor_id': 69, } ] admin_context = { 'identity_id': 'admin-uuid', 'profile_type': 'InsightsProfile', 'profile_id': 12, } identity_id = 'uuid' identity = response.Response( message={'name': 'foo@bar.com', 'email': 'foo@bar.com', 'id': identity_id} ) add_result = response.Response(message={'profiles_affected': []}) edit_super_admins = response.Response(message='control') mock_g.request_context.context_type = 'profile' mock_g.request_context.profile_type = 'InsightsProfile' with patch.object(resource_model, 'get_resources_brand', return_value=resources), patch.object( label, 'get_migrated_to_abacus', return_value=None ), patch.object(identity_logic, 'get_identity_by_id', return_value=identity), patch.object( model, 'check_vendor_star_access', return_value=True ), patch.object(auth0, 'update_user_metadata'), patch.object( pythonfeatures, 'get_single_feature', return_value=edit_super_admins ), patch.object(model, 'add_resources_to_identity', return_value=add_result): actual = profile.edit_identity_profiles_resources( admin_context, identity_id, resources, overwrite_existing_access=overwrite_existing_access, ) assert actual assert model.add_resources_to_identity.call_count == 1 args = model.add_resources_to_identity.call_args # identity param assert args[0][0]['id'] == identity_id assert args[0][0]['user_types'] == ['artist'] # resource param assert args[0][1][0]['resource_type'] == resources[0]['resource_type'] assert args[0][1][0]['uuid'] == resources[0]['uuid'] # optional param assert args[1]['overwrite_existing_access'] == overwrite_existing_access @patch('permissions.logic.profile.g') def test_edit_identity_label_profiles(mock_g, app_context): """Test edit_identity_profiles_resources.""" resources = [ { 'resource_type': 'Vendor', 'roles': ['analytics'], 'uuid': 8869, 'brand': constants.THEORCHARD_BRAND, 'vendor_id': 69, }, { 'resource_type': 'Vendor', 'roles': ['analytics'], 'uuid': 7123, 'brand': constants.THEORCHARD_BRAND, 'vendor_id': 23, }, ] mock_g.request_context.context_type = 'profile' mock_g.request_context.profile_type = 'SettingsProfile' admin_context = { 'identity_id': 'admin-uuid', 'profile_type': 'SettingsProfile', 'profile_id': 12, } identity_id = 'uuid' identity_obj = { 'name': 'foo@bar.com', 'email': 'foo@bar.com', 'id': identity_id, 'auth0_user_id': 'eabcde', } identity = response.Response(message=identity_obj) add_result = response.Response( message={ 'profiles_affected': [ { 'profile_type': 'LabelProfile', 'profile_id': 88108, 'roles': ['analytics'], 'user_identity': identity_obj, }, { 'profile_type': 'LabelProfile', 'profile_id': 88109, 'roles': ['analytics'], 'user_identity': identity_obj, }, { 'profile_type': 'InsightsProfile', 'profile_id': 88110, 'roles': ['analytics'], 'user_identity': identity_obj, }, ] } ) featureflag = response.Response(message='control') with patch.object(resource_model, 'get_resources_brand', return_value=resources), patch.object( label, 'get_migrated_to_abacus', return_value=None ), patch.object(identity_logic, 'get_identity_by_id', return_value=identity), patch.object( model, 'check_vendor_star_access', return_value=True ), patch.object(auth0, 'update_user_metadata'), patch.object( label, 'update_auth0_primary' ), patch.object(pythonfeatures, 'get_single_feature', return_value=featureflag), patch.object( model, 'add_resources_to_identity', return_value=add_result ): actual = profile.edit_identity_profiles_resources(admin_context, identity_id, resources) assert actual assert model.add_resources_to_identity.call_count == 1 update_call_args = { identity_obj['auth0_user_id']: {'type': 'alw', 'vend_contact_id': 88109} } auth0.update_user_metadata.assert_called_once_with(update_call_args) label.update_auth0_primary.assert_called_once_with(update_call_args) args = model.add_resources_to_identity.call_args # identity param assert args[0][0]['id'] == identity_id assert args[0][0]['user_types'] == ['label'] # resource param assert args[0][1][0]['resource_type'] == resources[0]['resource_type'] assert args[0][1][0]['uuid'] == resources[0]['uuid'] @patch('permissions.logic.profile.g') def test_edit_identity_profiles_several_resources(mock_g, app_context): """Test edit_identity_profiles_several_resources.""" resources = [ { 'resource_type': 'LabelParticipant', 'roles': ['analytics'], 'uuid': 8869, 'brand': constants.THEORCHARD_BRAND, 'vendor_id': 69, }, { 'resource_type': 'Vendor', 'roles': ['analytics'], 'uuid': 8870, 'brand': constants.THEORCHARD_BRAND, 'vendor_id': 70, }, { 'resource_type': 'LabelParticipant', 'roles': ['analytics'], 'uuid': 8871, 'brand': constants.THEORCHARD_BRAND, 'vendor_id': 71, }, { 'resource_type': 'Vendor', 'roles': ['analytics'], 'uuid': 8872, 'brand': constants.THEORCHARD_BRAND, 'vendor_id': 72, }, ] admin_context = { 'identity_id': 'admin-uuid', 'profile_type': 'InsightsProfile', 'profile_id': 12, } identity_id = 'uuid' identity = response.Response( message={'name': 'foo@bar.com', 'email': 'foo@bar.com', 'id': identity_id} ) add_result = response.Response(message={'profiles_affected': []}) edit_super_admins = response.Response(message='control') mock_g.request_context.context_type = 'profile' mock_g.request_context.profile_type = 'InsightsProfile' with patch.object(resource_model, 'get_resources_brand', return_value=resources), patch.object( label, 'get_migrated_to_abacus', return_value=None ), patch.object(identity_logic, 'get_identity_by_id', return_value=identity), patch.object( model, 'check_vendor_star_access', return_value=True ), patch.object(auth0, 'update_user_metadata'), patch.object( pythonfeatures, 'get_single_feature', return_value=edit_super_admins ), patch.object(model, 'add_resources_to_identity', return_value=add_result): actual = profile.edit_identity_profiles_resources(admin_context, identity_id, resources) assert actual assert model.add_resources_to_identity.call_count == 1 args = model.add_resources_to_identity.call_args # identity param assert args[0][0]['id'] == identity_id assert args[0][0]['user_types'] == ['label', 'artist'] # resource param assert args[0][1][0]['resource_type'] == resources[0]['resource_type'] assert args[0][1][0]['uuid'] == resources[0]['uuid'] @patch('permissions.logic.profile.g') def test_single_resource_invite_assigns_brand_from_resource(mock_g, app_context): """Test create_identity_profiles_resources with AWAL brand by an Orchard Admin.""" has_access_result = response.Response() admin_identity = { 'email': 'admin@bar.com', 'id': 'admin-uuid', 'default_brand': constants.THEORCHARD_BRAND, } resources = [ { 'resource_type': 'Vendor', 'roles': ['accounting', 'analytics'], 'uuid': 7123, 'brand': constants.AWAL_BRAND, 'vendor_id': 12345, } ] admin_context = { 'identity_id': 'admin-uuid', 'profile_type': 'SettingsProfile', 'profile_id': 12, } identity = {'name': 'foo@bar.com', 'email': 'foo@bar.com', 'id': '1234abcd'} add_result = response.Response(message={'profiles_affected': [], 'identity_created': False}) edit_super_admins = response.Response(message='control') mock_g.request_context.context_type = 'profile' mock_g.request_context.profile_type = 'SettingsProfile' with patch.object(resource_model, 'get_resources_brand', return_value=resources), patch.object( label, 'get_migrated_to_abacus', return_value=None ), patch.object(model, 'check_vendor_star_access', return_value=False), patch.object( auth0, 'update_user_metadata' ), patch.object( identity_logic, 'get_identity_by_id', return_value=response.Response(admin_identity), ), patch.object( model, 'has_admin_access_to_resources', return_value=has_access_result ), patch.object( pythonfeatures, 'get_single_feature', return_value=edit_super_admins ), patch.object(model, 'add_resources_to_identity', return_value=add_result): actual = profile.create_identity_profiles_resources(admin_context, identity, resources) assert actual.status assert model.has_admin_access_to_resources.called assert model.add_resources_to_identity.called args = model.add_resources_to_identity.call_args assert args.kwargs['brand'] == constants.AWAL_BRAND assert args.kwargs['create_auth0_user'] is False def test_get_identities_by_profile() -> None: """Test get_identities_by_profile logic.""" expected = ['uuid1', 'uuid2'] with patch.object(model, 'get_identities_by_profile', return_value=expected): actual = profile.get_identities_by_profile('ThisProfile', 12345, 'profile-uuid') assert actual == expected model.get_identities_by_profile.assert_called_with( 'ThisProfile', 12345, profile_uuid='profile-uuid', ) @pytest.mark.parametrize( ('v2_roles', 'expected_output_dict'), [ pytest.param( # Handles duplicates [application.COLLABORATORS_BASE_ROLE, application.COLLABORATORS_BASE_ROLE], {constants.COLLABORATORSPROFILE: [constants.ROYALTIES_ROLE]}, id='Duplicates', ), pytest.param( # Handles label profile roles [application.WORKSTATION_ADVERTISING_ROLE, application.WORKSTATION_MARKETING_ROLE], {constants.LABELPROFILE: [constants.ADVERTISING_ROLE, constants.MARKETING_ROLE]}, id='LabelProfile roles only without admin', ), pytest.param( # Handles label profile roles [application.WORKSTATION_ADMIN_ROLE, application.WORKSTATION_CATALOG_ROLE], {constants.LABELPROFILE: [constants.ADMINISTRATOR_ROLE, constants.CATALOG_ROLE]}, id='LabelProfile roles only with admin', ), pytest.param( # Handles a combination of label and non-label profile roles [ application.WORKSTATION_ADVERTISING_ROLE, application.WORKSTATION_MARKETING_ROLE, application.INSIGHTS_BASE_ROLE, application.SONGWHIP_READ_ROLE, ], { constants.LABELPROFILE: [constants.ADVERTISING_ROLE, constants.MARKETING_ROLE], constants.INSIGHTSPROFILE: [constants.ANALYTICS_ROLE], constants.SONGWHIPPROFILE: [constants.SONGWHIP_ROLE], }, id='LabelProfile and non-LabelProfile roles', ), pytest.param( # Handles settings role [application.SETTINGS_BASE_ROLE], {constants.SETTINGSPROFILE: []}, id='SettingsProfile role only', ), pytest.param( # Handles fansifter role (fansifter_can_view_fan_data is mapped to # audience so older nodes with audience continue to work) [application.FANSIFTER_BASE_ROLE], {constants.AUDIENCEPROFILE: [constants.AUDIENCE_ROLE]}, id='Fansifter role only', ), ], ) def test_v2_roles_to_profiles_and_roles_dict( v2_roles: list[str], expected_output_dict: dict[str, list[str]] ): """Test v2_roles_to_profiles_and_roles_dict...""" result = profile.v2_roles_to_profiles_and_roles_dict(v2_roles) # Sort nested lists for comparison assert {k: sorted(v) for k, v in result.items()} == expected_output_dict @pytest.mark.parametrize( ('resource', 'role', 'expected'), [ pytest.param( {'fake_resource_type': 'Vendor'}, 'accounting', False, id='False due to resource_type key not found.', ), pytest.param( {'resource_type': 'Vendor'}, 'accounting', False, id='False due to resource_type not Subaccount', ), pytest.param( {'resource_type': 'Subaccount'}, 'analytics', False, id='False due to role not accounting.', ), pytest.param( {'resource_type': 'Subaccount'}, 'accounting', True, id='True since every condition has passed.', ), pytest.param( {'resource_type': 'Subaccount'}, 'administrator', True, id='True since every condition has passed.', ), ], ) def test_moneyhub_feature_flag_enabled(resource, role, expected, app_context): """Test moneyhub feature flag function.""" result = should_add_moneyhub_subaccount(resource, role) assert result == expected