"""Test for resource logic.""" from unittest.mock import MagicMock, patch import pytest from owsresponse import response from application import app from permissions.connectors import mysql, neo4j, redis from permissions.constants import constants, error from permissions.logic import identity as identity_logic, resource from permissions.models import ( auth0 as auth0_model, label as label_model, profile as profile_model, resource as model, ) from permissions.utils.cache_utils import get_resources_key from permissions.validations.schemas.resource_to_profile import ResourceToProfileSchema from tests.unit.conftest import get_transactional_session_mock @pytest.mark.parametrize( ('resource_type', 'status', 'expected'), [ ('all', 200, response.Response([{'foo': 'bar'}])), ('dummy', 200, response.Response([])), ], ) @patch.object(model, 'get_artists_for_labelparticipants', return_value=response.Response([])) def test_get_resources_for_profile_diff_resource_type(monkeypatch, resource_type, status, expected): """Test get_resources_for_profile success.""" db_result = response.Response([{'foo': 'bar'}]) with patch.object(model, 'get_resources_of_type_for_profile', return_value=db_result): actual = resource.get_resources_for_profile(constants.LABELPROFILE, 123, resource_type) assert actual.status == status if status == 200: assert actual.message == expected.message else: assert actual.errors == expected.errors @pytest.mark.parametrize( ('profile_type', 'status', 'expected'), [ ('label', 200, response.Response([{'foo': 'bar'}])), ('artist', 200, response.Response([{'foo': 'bar'}])), ('artistprofile', 200, response.Response([{'foo': 'bar'}])), ('ArtistProfile', 200, response.Response([{'foo': 'bar'}])), ('Artistprofile', 200, response.Response([{'foo': 'bar'}])), ('LABELPROFILE', 200, response.Response([{'foo': 'bar'}])), ('dummy', 500, response.create_fatal_response('Invalid Profile type')), ('artist_profile', 500, response.create_fatal_response('Invalid Profile type')), # noqa ('artist-profile', 500, response.create_fatal_response('Invalid Profile type')), # noqa ], ) def test_get_resources_for_profile_diff_profile_types(monkeypatch, profile_type, status, expected): """Test get_resources_for_profile with different profile types.""" neo4j_result = response.Response([{'foo': 'bar'}]) artist_result = response.Response({}) with patch.object( model, 'get_resources_of_type_for_profile', return_value=neo4j_result ), patch.object(model, 'get_artists_for_labelparticipants', return_value=artist_result): actual = resource.get_resources_for_profile(profile_type, 123, constants.RESOURCE_TYPE_ALL) assert actual.status == status if status == 200: assert actual.message == expected.message else: assert actual.errors == expected.errors # note: These are not fixtures as you cannot use them in parametrize(). LABEL_RESOURCE_RESPONSE = response.Response( [ {'type': constants.VENDOR_RESOURCE_TYPE, 'id': 123}, {'type': constants.SUBACCOUNT_RESOURCE_TYPE, 'id': 456}, ] ) ARTIST_INFO_RESOURCE_RESPONSE = response.Response( [{'type': constants.ARTIST_INFO_RESOURCE_TYPE, 'id': 12245}] ) @pytest.mark.parametrize( ('profile_type', 'resource_type', 'expected'), [ (constants.LABELPROFILE, constants.RESOURCE_TYPE_LABEL, LABEL_RESOURCE_RESPONSE), ( constants.LABELPROFILE, constants.RESOURCE_TYPE_ALL, response.Response( [ {'type': constants.VENDOR_RESOURCE_TYPE, 'id': 123}, {'type': constants.SUBACCOUNT_RESOURCE_TYPE, 'id': 456}, {'type': constants.ARTIST_INFO_RESOURCE_TYPE, 'id': 12245}, ] ), ), (constants.INSIGHTSPROFILE, constants.RESOURCE_TYPE_LABEL, LABEL_RESOURCE_RESPONSE), (constants.PUBLISHINGPROFILE, constants.RESOURCE_TYPE_LABEL, LABEL_RESOURCE_RESPONSE), (constants.MONEYHUBPROFILE, constants.RESOURCE_TYPE_LABEL, LABEL_RESOURCE_RESPONSE), (constants.COLLABORATORSPROFILE, constants.RESOURCE_TYPE_LABEL, LABEL_RESOURCE_RESPONSE), (constants.SONGWHIPPROFILE, constants.RESOURCE_TYPE_LABEL, LABEL_RESOURCE_RESPONSE), (constants.DOCUMENTSPROFILE, constants.RESOURCE_TYPE_LABEL, LABEL_RESOURCE_RESPONSE), (constants.AUDIENCEPROFILE, constants.RESOURCE_TYPE_LABEL, LABEL_RESOURCE_RESPONSE), ], ) def test_get_resources_for_profile_labels(monkeypatch, profile_type, resource_type, expected): """Test get_resources_for_profile returning label resources.""" db_result = LABEL_RESOURCE_RESPONSE neo4j_result = ARTIST_INFO_RESOURCE_RESPONSE with patch.object( model, 'get_resources_of_type_for_profile', return_value=db_result ), patch.object(model, 'get_artists_for_labelparticipants', return_value=neo4j_result): actual = resource.get_resources_for_profile(profile_type, 123, resource_type) assert actual.message == expected.message @pytest.mark.parametrize( 'expected', [ response.Response({'foo': 'bar'}), response.create_not_found_response('Resource not found.'), response.create_fatal_response('Invalid Resource type'), ], ) def test_get_resources(expected): """Test get_resources.""" with patch.object(model, 'get_node', return_value=expected): actual = resource.get_resources('ArtistProfile', 123) assert actual.status == expected.status if expected.status == 200: assert actual.message == expected.message else: assert actual.errors == expected.errors @pytest.mark.parametrize( 'data,expected,edit_full_catalog_calls,expected_full_catalog_access', [ [ { 'resource_type': 'Vendor', 'resource_id': '7123', 'profile_type': 'InsightsProfile', 'profile_id': 100, 'roles': ['analytics'], }, response.Response({'foo': 'bar'}, status=201), 0, False, ], [ { 'resource_type': 'Vendor', 'resource_id': '7123', 'profile_type': 'InsightsProfile', 'profile_id': 100, 'roles': ['analytics'], }, response.create_fatal_response(error.ERROR_MESSAGE_CREATE_RELATIONSHIP_FAILED), 0, False, ], [ { 'resource_type': 'Vendor', 'resource_id': '*', 'profile_type': 'InsightsProfile', 'profile_id': 100, 'roles': ['analytics'], }, response.Response({'foo': 'bar'}, status=201), 1, True, ], [ { 'resource_type': 'Vendor', 'resource_id': '*', 'profile_type': 'InsightsProfile', 'profile_id': 100, 'roles': ['analytics'], }, response.create_fatal_response(error.ERROR_MESSAGE_CREATE_RELATIONSHIP_FAILED), 0, True, ], [ { 'resource_type': 'Vendor', 'resource_id': '7123', 'profile_uuid': 'some-uuid', 'profile_id': 0, 'profile_type': 'InsightsProfile', 'roles': ['analytics'], }, response.Response({'foo': 'bar'}, status=201), 0, False, ], [ { 'resource_type': 'Vendor', 'resource_id': '7123', 'profile_uuid': 'some-uuid', 'profile_id': 0, 'profile_type': 'InsightsProfile', 'roles': ['analytics'], }, response.create_fatal_response(error.ERROR_MESSAGE_CREATE_RELATIONSHIP_FAILED), 0, False, ], [ { 'resource_type': 'Vendor', 'resource_id': '*', 'profile_uuid': 'some-uuid', 'profile_id': 0, 'profile_type': 'InsightsProfile', 'roles': ['analytics'], }, response.Response({'foo': 'bar'}, status=201), 1, True, ], [ { 'resource_type': 'Vendor', 'resource_id': '*', 'profile_uuid': 'some-uuid', 'profile_id': 0, 'profile_type': 'LabelProfile', 'roles': ['analytics'], }, response.Response({'foo': 'bar'}, status=201), 0, False, ], [ { 'resource_type': 'Vendor', 'resource_id': '*', 'profile_uuid': 'some-uuid', 'profile_id': 0, 'profile_type': 'SettingsProfile', 'roles': ['analytics'], }, response.Response({'foo': 'bar'}, status=201), 0, True, ], [ { 'resource_type': 'Vendor', 'resource_id': '3344', 'profile_uuid': 'some-uuid', 'profile_id': 0, 'profile_type': 'SettingsProfile', 'roles': ['analytics'], }, response.Response({'foo': 'bar'}, status=201), 0, False, ], [ { 'resource_type': 'Vendor', 'resource_id': '*', 'profile_uuid': 'some-uuid', 'profile_id': 0, 'profile_type': 'SongwhipProfile', 'roles': ['analytics'], }, response.Response({'foo': 'bar'}, status=201), 0, True, ], [ { 'resource_type': 'Vendor', 'resource_id': '5566', 'profile_uuid': 'some-uuid', 'profile_id': 0, 'profile_type': 'SongwhipProfile', 'roles': ['analytics'], }, response.Response({'foo': 'bar'}, status=201), 0, False, ], ], ) @patch('permissions.logic.resource.g') def test_add_resource_to_profile( _, data, expected, edit_full_catalog_calls, expected_full_catalog_access, app_context ): """Test add_resource_to_profile.""" with patch.object( model, 'create_profile_to_resource_relationship', return_value=expected ), patch.object(model, 'create_profile_uuid_to_resource_relationship', return_value=expected): schema = ResourceToProfileSchema().load(data) redis_key = get_resources_key(schema['profile_type'], schema['profile_id']) redis.set(redis_key, 'test') admin_id = 'the-id-of-admin' actual = resource.add_resource_to_profile(schema, admin_id) assert actual.status == expected.status if expected.status == 201: assert actual.message == expected.message else: assert actual.errors == expected.errors if schema.get('profile_uuid'): assert model.create_profile_uuid_to_resource_relationship.call_count == 1 assert model.create_profile_to_resource_relationship.call_count == 0 model.create_profile_uuid_to_resource_relationship.assert_called_with( add_schema=schema, set_full_catalog_access=expected_full_catalog_access, admin_id=admin_id, ) else: assert model.create_profile_uuid_to_resource_relationship.call_count == 0 assert model.create_profile_to_resource_relationship.call_count == 1 model.create_profile_to_resource_relationship.assert_called_with( add_schema=schema, set_full_catalog_access=expected_full_catalog_access, admin_id=admin_id, ) assert redis.get(redis_key) is None @pytest.mark.parametrize( ('resource_type', 'expected'), [ (constants.RESOURCE_TYPE_LABEL, response.Response([{'Resource': 'One'}])), ( constants.RESOURCE_TYPE_ALL, response.Response([{'Resource': 'One'}, {'ArtistInfo': 'One'}]), ), ( constants.ARTIST_INFO_RESOURCE_TYPE, response.Response([{'Resource': 'One'}, {'ArtistInfo': 'One'}]), ), (constants.VENDOR_RESOURCE_TYPE, response.Response([{'Resource': 'One'}])), (constants.LABELPARTICIPANT_RESOURCE_TYPE, response.Response([{'Resource': 'One'}])), ], ) def test_get_resources_for_profile_uuid(monkeypatch, resource_type, expected): """Test get_resources_for_profile_uuid.""" neo4j_result = [{'Resource': 'One'}] artist_result = [{'ArtistInfo': 'One'}] profile_uuid = 'some-uuid' with patch.object( model, 'get_resources_of_type_for_profile_uuid', return_value=neo4j_result ), patch.object(model, 'get_artists_for_lp_by_profile_uuid', return_value=artist_result): actual = resource.get_resources_for_profile_uuid(profile_uuid, resource_type) assert actual assert actual.message == expected.message @pytest.mark.parametrize( ('neo4j_result', 'response_message'), [ ( ( [ { 'name': 'resource one name', 'vendorId': 12345, 'id': 12345, 'uuid': 'vendor-one-uuid', 'type': 'Vendor', } ], 1, ), [{'id': 12345, 'uuid': 'vendor-one-uuid', 'type': 'Vendor'}], ), ( ( [ { 'isDeleted': False, 'name': 'subaccount one name', 'id': 11111, 'uuid': 'subaccount-one-uuid', 'type': 'SubAccount', } ], 1, ), [{'id': 11111, 'uuid': 'subaccount-one-uuid', 'type': 'Subaccount'}], ), ( ( [ { 'isDeleted': False, 'name': 'subaccount two name', 'id': 11112, 'uuid': 'subaccount-two-uuid', 'type': 'Subaccount', } ], 1, ), [{'id': 11112, 'uuid': 'subaccount-two-uuid', 'type': 'Subaccount'}], ), ( ( [ { 'name': 'resource company brand name', 'uuid': 'company-brand-one-uuid', 'type': 'CompanyBrand', } ], 1, ), [ { 'id': 'company-brand-one-uuid', 'uuid': 'company-brand-one-uuid', 'type': 'CompanyBrand', } ], ), ], ) def test_get_resources_for_identity_uuid(monkeypatch, neo4j_result, response_message): """Test get_resources_for_identity_uuid.""" identity_uuid = 'test-identity-uuid' limit = constants.DEFAULT_LIMIT offset = constants.DEFAULT_OFFSET with patch.object( model, 'get_resources_for_identity_uuid', return_value=neo4j_result ) as mock_model_get_resources_for_identity_uuid: actual = resource.get_resources_for_identity_uuid(identity_uuid, limit, offset) assert actual assert actual.message['items'] == response_message assert actual.message['pagination'] == { 'type': constants.PAGINATION_TYPE_STANDARD, 'total_records': neo4j_result[1], } mock_model_get_resources_for_identity_uuid.assert_called_with( 'test-identity-uuid', constants.DEFAULT_LIMIT, constants.DEFAULT_OFFSET ) @patch('permissions.models.resource.get_resources_of_type_for_profile_uuid') def test_check_label_access_for_profile(mock_lookup): """Test check_label_access_for_profile.""" mock_lookup.return_value = [ {'type': 'Vendor', 'vendorId': 1}, {'type': 'SubAccount', 'vendorId': 2}, ] assert resource.check_label_access_for_profile('abc', 1) == 200 assert resource.check_label_access_for_profile('abc', 2) == 403 # test Vendor * access mock_lookup.return_value.append({'type': 'Vendor', 'vendorId': '*'}) assert resource.check_label_access_for_profile('abc', 1) == 200 assert resource.check_label_access_for_profile('abc', 2) == 200 assert resource.check_label_access_for_profile('abc', 0) == 200 @pytest.mark.parametrize( 'data,expected,edit_full_catalog_calls', [ [ { 'resource_type': 'Vendor', 'resource_id': '7123', 'profile_type': 'InsightsProfile', 'profile_id': 100, 'roles': ['analytics'], }, response.Response({'foo': 'bar'}, status=201), 0, ], [ { 'resource_type': 'Vendor', 'resource_id': '7123', 'profile_type': 'InsightsProfile', 'profile_id': 100, 'roles': ['analytics'], }, response.create_fatal_response(error.ERROR_MESSAGE_DELETE_RELATIONSHIP_FAILED), 0, ], [ { 'resource_type': 'Vendor', 'resource_id': '*', 'profile_type': 'InsightsProfile', 'profile_id': 100, 'roles': ['analytics'], }, response.Response({'foo': 'bar'}, status=201), 1, ], [ { 'resource_type': 'Vendor', 'resource_id': '*', 'profile_type': 'InsightsProfile', 'profile_id': 100, 'roles': ['analytics'], }, response.create_fatal_response(error.ERROR_MESSAGE_DELETE_RELATIONSHIP_FAILED), 0, ], [ { 'resource_type': 'Vendor', 'resource_id': '7123', 'profile_uuid': 'some-uuid', 'profile_type': 'InsightsProfile', 'profile_id': 0, 'roles': ['analytics'], }, response.Response({'foo': 'bar'}, status=201), 0, ], [ { 'resource_type': 'Vendor', 'resource_id': '7123', 'profile_uuid': 'some-uuid', 'profile_type': 'InsightsProfile', 'profile_id': 0, 'roles': ['analytics'], }, response.create_fatal_response(error.ERROR_MESSAGE_DELETE_RELATIONSHIP_FAILED), 0, ], [ { 'resource_type': 'Vendor', 'resource_id': '*', 'profile_uuid': 'some-uuid', 'profile_type': 'InsightsProfile', 'profile_id': 0, 'roles': ['analytics'], }, response.Response({'foo': 'bar'}, status=201), 1, ], ], ) def test_delete_resource_to_profile(data, expected, edit_full_catalog_calls): """Test delete_resource_to_profile.""" with patch.object( model, 'soft_delete_profile_to_resource_relationship', return_value=expected ), patch.object( model, 'soft_delete_profile_uuid_to_resource_relationship', return_value=expected ): schema = ResourceToProfileSchema().load(data) redis_key = get_resources_key(schema['profile_type'], schema['profile_id']) redis.set(redis_key, 'test') actual = resource.delete_resource_to_profile(schema) assert actual.status == expected.status if expected.status == 201: assert actual.message == expected.message else: assert actual.errors == expected.errors if schema.get('profile_uuid'): assert model.soft_delete_profile_uuid_to_resource_relationship.call_count == 1 assert model.soft_delete_profile_to_resource_relationship.call_count == 0 else: assert model.soft_delete_profile_uuid_to_resource_relationship.call_count == 0 assert model.soft_delete_profile_to_resource_relationship.call_count == 1 assert redis.get(redis_key) is None @pytest.mark.parametrize( ('neo4j', 'labelparticipants', 'expected'), [ ( # when we only get data from neo4j access, no db data or labelparticipants. [{'id': 1, 'type': 'label'}, {'id': 1, 'type': 'artist'}], response.Response([]), [{'id': 1, 'type': 'label'}, {'id': 1, 'type': 'artist'}], ), ( # when we only get data from neo4j access and some labelparticipants, no db data. [{'id': 1, 'type': 'label'}, {'id': 1, 'type': 'artist'}], response.Response([{'id': 456, 'type': 'artist'}]), [ {'id': 1, 'type': 'label'}, {'id': 1, 'type': 'artist'}, {'id': 456, 'type': 'artist'}, ], ), ( # No dupe records. [{'id': 1, 'type': 'label'}, {'id': 1, 'type': 'artist'}], response.Response([]), [{'id': 1, 'type': 'label'}, {'id': 1, 'type': 'artist'}], ), ( [{'id': 1, 'type': 'label'}, {'id': 1, 'type': 'artist'}], response.Response([{'id': 3, 'type': 'artist'}, {'id': 4, 'type': 'artist'}]), [ {'id': 1, 'type': 'label'}, {'id': 1, 'type': 'artist'}, {'id': 3, 'type': 'artist'}, {'id': 4, 'type': 'artist'}, ], ), ], ) def test_get_resources_for_all_resource_types(monkeypatch, neo4j, labelparticipants, expected): """Test get_resources_for_profile with both neo4j and db data.""" neo4j_result = response.Response(neo4j) with patch.object( model, 'get_resources_of_type_for_profile', return_value=neo4j_result ), patch.object(model, 'get_artists_for_labelparticipants', return_value=labelparticipants): # noqa actual = resource.get_resources_for_profile( constants.INSIGHTSPROFILE, 123, constants.RESOURCE_TYPE_ALL ) assert actual assert actual.message == expected @pytest.mark.parametrize( ('get_result', 'create_result', 'expected'), [ [ response.create_not_found_response(), response.Response(message={'foo', 'bar'}), response.Response(message={'foo', 'bar'}), ], [ response.Response(status=200), None, response.create_error_response( error.ERROR_CODE_VALIDATION_ERROR, 'Resource already exist.' ), ], [ response.create_not_found_response(), response.create_fatal_response(error.ERROR_MESSAGE_DELETE_RELATIONSHIP_FAILED), response.create_fatal_response(error.ERROR_MESSAGE_DELETE_RELATIONSHIP_FAILED), ], ], ) def test_create_resource(get_result, create_result, expected): """Test create_resource.""" with patch.object(model, 'create_node', return_value=create_result), patch.object( model, 'get_node', return_value=get_result ): actual = resource.create_resource('ArtistInfo', 500, {'name': 'test'}) assert actual.status == expected.status if expected.status == 201: model.create_node.assert_called_once_with('ArtistInfo', 500, {'name': 'test'}) @pytest.mark.parametrize( ('get_result', 'expected'), [ [response.create_not_found_response(), response.Response(status=204)], [response.Response(status=200), response.Response(status=204)], ], ) def test_delete_resource(get_result, expected): """Test delete_resource.""" delete_result = response.Response(status=204) with patch.object(model, 'delete_node', return_value=delete_result), patch.object( model, 'get_node', return_value=get_result ): actual = resource.delete_resource('ArtistInfo', 500) assert actual.status == expected.status if get_result.status == 200: model.delete_node.assert_called_once_with('ArtistInfo', 500) else: model.delete_node.assert_not_called() @pytest.mark.parametrize( ('profile_types', 'label_participants', 'get_profiles_result', 'all_profiles', 'admin_result'), [ # When this is not super admin user. ( None, [123, 456], response.Response([{'foo': 'bar'}]), None, response.create_not_found_response(), ), ( ['LabelProfile', 'InsightsProfile'], None, response.Response([{'foo': 'bar'}]), None, response.create_not_found_response(), ), ( ['InsightsProfile'], [123, 456], response.Response([{'foo': 'bar'}]), None, response.create_not_found_response(), ), # When this is a super admin user. ( None, [123, 456], response.Response([{'foo': 'bar'}]), response.Response([{'admin': 'response'}]), response.Response(), ), ( ['LabelProfile', 'InsightsProfile'], None, response.Response([{'foo': 'bar'}]), response.Response([{'admin': 'response'}]), response.Response(), ), ( ['InsightsProfile'], [123, 456], response.Response([{'foo': 'bar'}]), response.Response([{'admin': 'response'}]), response.Response(), ), ], ) def test_get_user_profiles_for_admin( monkeypatch, profile_types, label_participants, get_profiles_result, all_profiles, admin_result ): """Test get_user_profiles_for_admin.""" admin_context = { 'identity_id': 'admin-uuid', 'profile_type': 'InsightsProfile', 'profile_id': 12, } with patch.object( model, 'get_all_user_profiles_for_admin', return_value=get_profiles_result ), patch.object( profile_model, 'check_vendor_star_access', return_value=admin_result ), patch.object(model, 'get_all_profiles_with_artist_access', return_value=all_profiles): request_ctx = app.test_request_context() request_ctx.push() actual = resource.get_user_profiles_for_admin( admin_context, profile_types, label_participants=label_participants ) assert actual.status == get_profiles_result.status if admin_result: model.get_all_profiles_with_artist_access.assert_called_once_with( admin_context['identity_id'], profile_types, 50, 0, None, label_participants, None, None, None, False, ) assert actual.message == all_profiles.message else: model.get_all_user_profiles_for_admin.assert_called_once_with( admin_context['identity_id'], profile_types, 50, 0, None, label_participants, None, None, None, None, True, False, ) assert actual.message == get_profiles_result.message @pytest.mark.parametrize( ( 'resource_type', 'admin_result', 'get_resources_result', 'get_admin_result', 'active', 'expected', ), [ # when the admin in not super admin. ( # valid resource and model response 'LabelParticipant', response.create_not_found_response(), response.Response(status=200), None, True, response.Response(status=200), ), ( # valid resource but error in model response 'LabelParticipant', response.create_not_found_response(), response.create_fatal_response('Some Neo4j error'), None, False, response.create_fatal_response('Some Neo4j error'), ), # when the admin in a super admin. ( 'LabelParticipant', response.Response(status=200), None, response.Response([{'foo': 'bar'}]), True, response.Response(status=200), ), ( 'LabelParticipant', response.Response(status=200), None, response.create_fatal_response('Some Neo4j error'), False, response.create_fatal_response('Some Neo4j error'), ), ], ) def test_get_user_resources_for_admin( monkeypatch, resource_type, admin_result, get_resources_result, get_admin_result, active, expected, ): """Test get_user_resources_for_admin.""" admin_context = { 'identity_id': 'admin-uuid', 'profile_type': 'InsightsProfile', 'profile_id': 12, } identity_id = 'user-uuid' with patch.object( model, 'get_users_resources_by_type', return_value=get_resources_result ), patch.object( model, 'get_user_resources_for_admin', return_value=get_admin_result ), patch.object(profile_model, 'check_vendor_star_access', return_value=admin_result): actual = resource.get_user_resources_for_admin(admin_context, identity_id, resource_type) if expected: assert actual.status == expected.status if get_resources_result: # not super admin model.get_users_resources_by_type.assert_called_once_with( admin_context, identity_id, resource_type, constants.DEFAULT_LIMIT, 0, active, ) if get_admin_result: # super admin model.get_user_resources_for_admin.assert_called_once_with( identity_id, constants.SETTINGS_SUPPORT_MAPPING['profileTypes'], resource_type, constants.DEFAULT_LIMIT, 0, active, ) else: assert actual.status == expected.status assert actual.errors == expected.errors @pytest.mark.parametrize( ('identity_id', 'admin_result', 'resource_type', 'admin_profile_type', 'expected'), [ # when the admin in not super admin. ( 'noadmin-uuid', response.create_not_found_response(), 'Vendor', 'SettingsProfile', response.Response(message='get_users_resources_for_settings_by_type called'), ), ( 'admin-uuid', response.create_not_found_response(), 'Vendor', 'InsightsProfile', response.Response(message='get_user_resources_for_admin called'), ), # when the admin is a super admin. ( 'noadmin-uuid', response.Response(), 'Vendor', 'InsightsProfile', response.Response(message='get_user_resources_for_admin called'), ), ( 'admin-uuid', response.Response(), 'Vendor', 'InsightsProfile', response.Response(message='get_user_resources_for_admin called'), ), ( 'admin-uuid', response.Response(), 'VendorStar', 'SettingsProfile', response.Response(message='get_vend_star_for_admin called'), ), ( 'noadmin-uuid', response.Response(), 'VendorStar', 'SettingsProfile', response.Response(message='get_user_vend_star called'), ), ], ) def test_get_user_resources_for_admin_cases( monkeypatch, identity_id, admin_result, resource_type, admin_profile_type, expected ): """Test get_user_resources_for_admin.""" admin_context = { 'identity_id': 'admin-uuid', 'profile_type': admin_profile_type, 'profile_id': 12, } resource_response = response.Response('get_users_resources_by_type called') user_vend_star = response.Response('get_user_vend_star called') admin_response = response.Response('get_user_resources_for_admin called') admin_response_settings = response.Response('get_users_resources_for_settings_by_type called') admin_vend_star = response.Response('get_vend_star_for_admin called') with patch.object( model, 'get_users_resources_by_type', return_value=resource_response ), patch.object( model, 'get_user_resources_for_admin', return_value=admin_response ), patch.object(model, 'get_vend_star_for_admin', return_value=admin_vend_star), patch.object( model, 'get_user_vend_star', return_value=user_vend_star ), patch.object( profile_model, 'check_vendor_star_access', return_value=admin_result ), patch.object( model, 'get_users_resources_for_settings_by_type', return_value=admin_response_settings ): actual = resource.get_user_resources_for_admin(admin_context, identity_id, resource_type) assert actual.status == expected.status assert actual.message == expected.message @pytest.mark.parametrize( ('resource_type', 'admin_result', 'resources', 'get_admin_result', 'expected'), [ # when the admin in not super admin. ( # valid resource and model response 'LabelParticipant', response.create_not_found_response(), response.Response(status=200), None, response.Response(status=200), ), ( # valid resource but error in model response 'LabelParticipant', response.create_not_found_response(), response.create_fatal_response('Some Neo4j error'), None, response.create_fatal_response('Some Neo4j error'), ), # when the admin in a super admin. ( 'LabelParticipant', response.Response(status=200, message={'manage_profile_types': ['InsightsProfile']}), None, response.Response([{'foo': 'bar'}]), response.Response(status=200), ), ( 'VendorStar', response.Response(status=200), response.Response([{'foo': 'bar'}]), None, response.Response(status=200), ), ( 'LabelParticipant', response.Response(status=200, message={'manage_profile_types': ['InsightsProfile']}), None, response.create_fatal_response('Some Neo4j error'), response.create_fatal_response('Some Neo4j error'), ), ], ) def test_get_user_resources_for_admin_settings_profile( monkeypatch, resource_type, admin_result, resources, get_admin_result, expected ): """Test get_user_resources_for_admin.""" admin_context = { 'identity_id': 'admin-uuid', 'profile_type': 'SettingsProfile', 'profile_id': 12, } identity_id = 'user-uuid' with patch.object( model, 'get_users_resources_for_settings_by_type', return_value=resources ), patch.object( model, 'get_user_resources_for_admin', return_value=get_admin_result ), patch.object(model, 'get_user_vend_star', return_value=resources), patch.object( profile_model, 'check_vendor_star_access', return_value=admin_result ): actual = resource.get_user_resources_for_admin(admin_context, identity_id, resource_type) if expected: assert actual.status == expected.status if resources: if resource_type != constants.VENDOR_STAR_RESOURCE_TYPE: # not super admin model.get_users_resources_for_settings_by_type.assert_called_once_with( admin_context, identity_id, constants.SETTINGS_SUPPORT_MAPPING['profileTypes'], resource_type, constants.DEFAULT_LIMIT, 0, True, ) else: model.get_user_vend_star.assert_called_once_with(admin_context, identity_id) if get_admin_result: # super admin model.get_user_resources_for_admin.assert_called_once_with( identity_id, constants.SETTINGS_SUPPORT_MAPPING['profileTypes'], resource_type, constants.DEFAULT_LIMIT, 0, True, ) else: assert actual.status == expected.status assert actual.errors == expected.errors @pytest.mark.parametrize( ('identity_result', 'deactivate_result', 'expected'), [ ( # Identity found response.Response(message={'id': 'user-uuid'}), response.Response({'foo': 'bar'}), response.Response({'foo': 'bar'}), ), ( # Identity already deactivated response.Response(message={'id': 'user-uuid', 'active': 'N'}), response.Response({'foo': 'bar'}), response.Response(message={'id': 'user-uuid', 'active': 'N'}, status=202), ), ( # Identity found but model returned error response.Response(message={'id': 'user-uuid'}), response.create_fatal_response('Some Neo4j error'), response.create_fatal_response('Some Neo4j error'), ), ], ) @patch('permissions.logic.resource.g') def test_deactivate_resources_common_with_admin( _, identity_result, deactivate_result, expected, app_context ): """Test deactivate_resources_common_with_admin.""" admin_context = { 'identity_id': 'admin-uuid', 'profile_type': 'InsightsProfile', 'profile_id': 12, } identity_id = 'user-uuid' with patch.object(model, 'deactivate_user', return_value=deactivate_result), patch.object( identity_logic, 'get_identity_by_id', return_value=identity_result ): actual = resource.deactivate_resources_common_with_admin(admin_context, identity_id) assert actual.status == expected.status if expected.status == 200: model.deactivate_user.assert_called_once_with( admin_context, identity_id, constants.SETTINGS_SUPPORT_MAPPING['deleteProfileTypes'] ) elif expected: assert model.deactivate_user.call_count == 0 elif not expected: assert actual.errors == expected.errors @pytest.mark.parametrize( ('mock_identity', 'mock_profile', 'mock_auth0', 'mock_id', 'expected'), [ ( # Identity found [{'id': 'user-uuid', 'resource_count': 0}], [{'id': 'id', 'type': 'LabelProfile'}], response.Response( message={ 'name': 'test', 'id': 'abcd-234', 'email': 'one@theorchard.io', 'auth0_user_id': 'id', 'first_name': 'first', 'last_name': 'last', 'user_types': None, } ), {'auth0_user_id': 'id'}, { 'auth0_result': { 'name': 'test', 'id': 'abcd-234', 'email': 'one@theorchard.io', 'auth0_user_id': 'id', 'first_name': 'first', 'last_name': 'last', 'user_types': None, }, 'auth0_user_id': 'id', 'id': 'user-uuid', 'resource_count': 0, }, ), ], ) @patch('permissions.logic.resource.g') def test_label_update_vend_contact_called_within_deactivate( _, mock_identity, mock_profile, mock_auth0, mock_id, expected, app_context ): """Test label_update_vend_contact_called_within_deactivate.""" session_mock = get_transactional_session_mock(mock_identity) admin_context = { 'identity_id': 'admin-uuid', 'profile_type': 'InsightsProfile', 'profile_id': 12, } identity_id = 'user-uuid' mysql_session_mock = MagicMock() mysql_session_mock.execute.return_value = {'success': 'update'} with patch.object(neo4j, '_get_neo4j_session', return_value=session_mock), patch.object( identity_logic, 'get_identity_by_id' ), patch.object(label_model, 'update_vend_contact'), patch.object( auth0_model, 'activate_deactivate_user', return_value=mock_auth0 ), patch.object(model, '_deactivate_identity', return_value=mock_id), patch.object( model, '_deactivate_resources_common_with_admin', return_value=mock_profile ), patch.object(mysql, 'db_session', return_value=mysql_session_mock), patch.object( model, '_get_identity_resource_count', side_effect=mock_identity ): actual = resource.deactivate_resources_common_with_admin(admin_context, identity_id) assert actual assert actual.message == expected label_model.update_vend_contact.assert_called_once()