"""Test resource model class.""" import textwrap from copy import deepcopy from unittest.mock import ANY, call, patch import pytest import stringcase from owsresponse import response from pythonfeatures import pythonfeatures from permissions.connectors import neo4j from permissions.constants import constants, error from permissions.models import auth0, identity as identity_model, owsusers, resource from permissions.types import Identity from tests.unit.conftest import get_session_mock, get_transactional_session_mock def test_get_resources_of_type_for_profile_no_result(): """Test get_resources_of_type_for_profile when no result are returned.""" session_mock = get_session_mock([]) with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.get_resources_of_type_for_profile('ArtisProfile', 10011, ['ArtistInfo']) assert actual assert actual.message == [] args = session_mock.__enter__().run.call_args expected_query = textwrap.dedent(""" MATCH (p:Profile)-[r:HAS_ACCESS_TO|HAS_ADMIN_ACCESS_TO]->(x) WHERE p.profileType = $profile_type AND p.profileId = $profile_id AND (x:ArtistInfo) AND NOT x:ParentCompany with p, r, x OPTIONAL MATCH (x)<-[:OWNS]-(v:Vendor) RETURN x as resource, apoc.coll.toSet(apoc.coll.flatten(collect(r.roles)+collect(p.roles))) as roles, v.id as vendorId, TRIM(v.name) as vendorName """) assert args[0][0] == expected_query def test_get_resources_of_type_for_profile_multiple(make_graph_node): """Test get_resources_of_type_for_profile when it finds multiple nodes.""" artist1 = make_graph_node(node_id=100, labels={'ArtistInfo'}) artist2 = make_graph_node(node_id=200, labels={'ArtistInfo'}) session_mock = get_session_mock( [{'resource': artist1, 'roles': ['foo']}, {'resource': artist2, 'roles': ['bar']}] ) expected = [ {'type': 'ArtistInfo', 'id': 100, 'roles': ['foo']}, # noqa {'type': 'ArtistInfo', 'id': 200, 'roles': ['bar']}, # noqa ] with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.get_resources_of_type_for_profile('ArtisProfile', 10011, ['ArtistInfo']) assert actual assert actual.message == expected args = session_mock.__enter__().run.call_args expected_query = textwrap.dedent(""" MATCH (p:Profile)-[r:HAS_ACCESS_TO|HAS_ADMIN_ACCESS_TO]->(x) WHERE p.profileType = $profile_type AND p.profileId = $profile_id AND (x:ArtistInfo) AND NOT x:ParentCompany with p, r, x OPTIONAL MATCH (x)<-[:OWNS]-(v:Vendor) RETURN x as resource, apoc.coll.toSet(apoc.coll.flatten(collect(r.roles)+collect(p.roles))) as roles, v.id as vendorId, TRIM(v.name) as vendorName """) assert args[0][0] == expected_query def test_get_resources_of_type_for_profile_resource_types(make_graph_node): """Test get_resources_of_type_for_profile for multiple resource types.""" artist1 = make_graph_node(node_id=100, labels={'Vendor', 'Label'}) artist2 = make_graph_node(node_id=200, labels={'ArtistInfo'}) subaccount1 = make_graph_node(node_id=300, labels={'SubAccount'}) collab1 = make_graph_node(node_id=400, labels={'Collaborator'}) session_mock = get_session_mock( [ {'resource': artist1, 'roles': ['foo']}, {'resource': artist2, 'roles': ['bar']}, {'resource': subaccount1, 'roles': ['bar']}, {'resource': collab1, 'roles': ['baz']}, ] ) expected = [ {'type': 'Vendor', 'id': 100, 'roles': ['foo']}, {'type': 'ArtistInfo', 'id': 200, 'roles': ['bar']}, {'type': 'Subaccount', 'id': 300, 'roles': ['bar']}, {'type': 'Collaborator', 'id': 400, 'roles': ['baz']}, ] with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.get_resources_of_type_for_profile( 'InsightsProfile', 10011, ['Vendor', 'SubAccount', 'Collaborator'] ) assert actual assert actual.message == expected args = session_mock.__enter__().run.call_args expected_query = textwrap.dedent(""" MATCH (p:Profile)-[r:HAS_ACCESS_TO|HAS_ADMIN_ACCESS_TO]->(x) WHERE p.profileType = $profile_type AND p.profileId = $profile_id AND (x:Vendor OR x:SubAccount OR x:Collaborator) AND NOT x:ParentCompany with p, r, x OPTIONAL MATCH (x)<-[:OWNS]-(v:Vendor) RETURN x as resource, apoc.coll.toSet(apoc.coll.flatten(collect(r.roles)+collect(p.roles))) as roles, v.id as vendorId, TRIM(v.name) as vendorName """) assert args[0][0] == expected_query def test_get_artists_for_labelparticipants(make_graph_node): """Test get_artists_for_labelparticipants.""" artist1 = make_graph_node(node_id=100, data={'name': 'a1'}, labels={'ArtistInfo'}) artist2 = make_graph_node(node_id=200, data={'name': 'a2'}, labels={'ArtistInfo'}) session_mock = get_session_mock( [{'resource': artist1, 'roles': ['foo']}, {'resource': artist2, 'roles': ['bar']}] ) expected = [ {'type': 'ArtistInfo', 'name': 'a1', 'id': 100, 'roles': ['foo']}, # noqa {'type': 'ArtistInfo', 'name': 'a2', 'id': 200, 'roles': ['bar']}, # noqa ] with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.get_artists_for_labelparticipants('InsightsProfile', 10011) assert actual assert actual.message == expected args = session_mock.__enter__().run.call_args assert args[0] == ( 'MATCH (p:Profile)-[r:HAS_ACCESS_TO]->(lp:LabelParticipant)\n -[:CREATED_FROM]->(a:ArtistInfo)\n WHERE\n p.profileType = $profile_type AND p.profileId = $profile_id\n RETURN\n a as resource,\n apoc.coll.toSet(apoc.coll.flatten(collect(r.roles)+collect(p.roles))) as roles\n ', ) # noqa def test_soft_delete_profile_to_resource_relationship(): """Test soft_delete_profile_to_resource_relationship.""" schema = { 'resource_type': 'ArtistInfo', 'resource_id': 100, 'profile_type': 'ArtistProfile', 'profile_id': 500, } expected = deepcopy(schema) session_mock = get_session_mock([{'input': 'input test', 'output': 'output test'}]) with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.soft_delete_profile_to_resource_relationship(schema) assert actual.status == 204 assert not actual.message session_mock.__enter__().run.assert_called_once_with(ANY, **expected) # check parts of the query. args = session_mock.__enter__().run.call_args assert 'MATCH (p:Profile)-[rel:HAS_ACCESS_TO]->' in args[0][0] assert '(r:ArtistInfo)' in args[0][0] assert 'p.profileType = $profile_type AND' in args[0][0] def test_soft_delete_profile_to_resource_relationship_no_result(make_graph_node): """Test soft_delete_profile_to_resource_relationship, no relation exist.""" schema = { 'resource_type': 'ArtistInfo', 'resource_id': 100, 'profile_type': 'ArtistProfile', 'profile_id': 500, } expected = {'code': 'internal_error', 'message': error.ERROR_MESSAGE_DELETE_RELATIONSHIP_FAILED} session_mock = get_session_mock([]) with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.soft_delete_profile_to_resource_relationship(schema) assert not actual assert actual.errors == expected session_mock.__enter__().run.assert_called_once_with(ANY, **schema) # check parts of the query. args = session_mock.__enter__().run.call_args assert 'MATCH (p:Profile)-[rel:HAS_ACCESS_TO]->' in args[0][0] assert '(r:ArtistInfo)' in args[0][0] def test_get_node(make_graph_node): """Test get_node.""" artist1 = make_graph_node(node_id=100, labels={'ArtistInfo'}) session_mock = get_session_mock([{'resource': artist1}]) expected = {'type': 'ArtistInfo', 'id': 100} with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.get_node('artistinfo', 100) assert actual assert actual.message == expected def test_get_node_invalid_type(make_graph_node): """Test get_node with invalid type.""" artist1 = make_graph_node(node_id=100, labels={'ArtistInfo'}) session_mock = get_session_mock([{'resource': artist1}]) expected = {'code': 'internal_error', 'message': 'Invalid Resource type'} with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.get_node('foo', 100) assert not actual assert actual.errors == expected def test_get_node_no_result(): """Test get_node when no matching node found.""" session_mock = get_session_mock([{'resource': None}]) expected = {'code': 'not_found_error', 'message': 'Resource not found.'} with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.get_node('artistinfo', 100) assert not actual assert actual.errors == expected @patch('permissions.models.resource.g') def test_get_resources_brand(mock_g, mocker, app_context): """Test get_resource_brand.""" resource_input = [ {'resource_type': 'Vendor', 'uuid': '123abc'}, {'resource_type': 'LabelParticipant', 'uuid': '456def'}, ] expected = [ { 'resource_type': 'Vendor', 'uuid': '123abc', 'brand': constants.AWAL_BRAND, 'vendor_id': 12345, }, { 'resource_type': 'LabelParticipant', 'uuid': '456def', 'brand': constants.THEORCHARD_BRAND, 'vendor_id': None, }, ] session_mock = get_session_mock( [ {'result': [{'companyBrand': constants.AWAL_BRAND, 'vendorId': 12345}]}, {'result': [{'companyBrand': constants.THEORCHARD_BRAND}]}, ] ) with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.get_resources_brand(resource_input) assert actual assert actual == expected args = session_mock.__enter__().run.call_args assert args[0] == ( """UNWIND $resources as res RETURN CASE res.resource_type WHEN 'LabelParticipant' THEN [(l:LabelParticipant{uuid:res.uuid})<-[:HAS_LABEL_PARTICIPANT]-(v:Vendor) <-[:HAS_LABEL]-(cb:CompanyBrand)| {companyBrand: cb.name}] WHEN 'Vendor' THEN [(v:Vendor{uuid:res.uuid})<-[:HAS_LABEL]-(cb:CompanyBrand) | {companyBrand: cb.name, vendorId: v.vendorId}] WHEN 'Subaccount' THEN [(s:Subaccount{uuid:res.uuid})<-[:OWNS]-(v:Vendor)<-[:HAS_LABEL]-(cb:CompanyBrand) | {companyBrand: cb.name, vendorId: v.vendorId}] WHEN 'Collaborator' THEN [(c:Collaborator{uuid:res.uuid})<-[:OWNS]-(v:Vendor)<-[:HAS_LABEL]-(cb:CompanyBrand) | {companyBrand: cb.name}] END AS result;""", ) # noqa @patch('permissions.models.resource.g') def test_get_resource_brand_no_result(mock_g, mocker, app_context): """Test get_resource_brand when no matching node found.""" session_mock = get_session_mock([{'result': []}]) expected = { 'code': 'not_found_error', 'message': 'Missing resource information. Could not assign resource access.', } with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.get_resources_brand({'resource_type': 'Vendor', 'uuid': '123abc'}) assert not actual assert actual.errors == expected def test_create_node(make_graph_node): """Test create_node.""" artist1 = make_graph_node(node_id=500, labels={'ArtistInfo'}) session_mock = get_session_mock([{'resource': artist1}]) expected = {'type': 'ArtistInfo', 'id': 500} with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.create_node('artistinfo', 500) assert actual assert actual.message == expected def test_create_node_invalid_type(make_graph_node): """Test create_node with invalid type.""" artist1 = make_graph_node(node_id=100, labels={'ArtistInfo'}) session_mock = get_session_mock([{'resource': artist1}]) expected = {'code': 'internal_error', 'message': 'Invalid Resource type'} with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.create_node('dummy', 100) assert not actual assert actual.errors == expected def test_delete_node(): """Test delete_node success.""" session_mock = get_session_mock([]) expected = None with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.delete_node('ArtistInfo', 500) assert actual assert actual.message == expected session_mock.__enter__().run.assert_called_once_with(ANY, node_id=500) # check parts of the query. args = session_mock.__enter__().run.call_args assert 'Match (a:ArtistInfo)' in args[0][0] assert 'WHERE a.id = $node_id' in args[0][0] assert 'DETACH DELETE a' in args[0][0] @patch('permissions.models.resource.g') def test_create_profile_to_resource_relationship(_, app_context): """Test create_profile_to_resource_relationship.""" schema = { 'resource_type': 'ArtistInfo', 'resource_id': 100, 'profile_type': 'ArtistProfile', 'profile_id': 500, 'roles': ['test'], } expected = deepcopy(schema) session_mock = get_session_mock([{'rel': 'foo'}]) with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.create_profile_to_resource_relationship(schema, 'admin-id') assert actual assert actual.message == expected assert session_mock.__enter__().run.call_count == 2 session_mock.__enter__().run.assert_called_with(ANY, **expected) # check parts of the query. args = session_mock.__enter__().run.call_args assert 'MERGE (p)-[rel:HAS_ACCESS_TO]->(r)' in args[0][0] assert 'MATCH (p:Profile)' in args[0][0] assert 'MATCH(r:ArtistInfo)' in args[0][0] assert 'p.profileType = $profile_type AND' in args[0][0] def test_edit_full_catalog_access(make_graph_node): """Test edit_full_catalog_access.""" access = True profile_id = 500 profile_type = 'InsightsProfile' profile1 = make_graph_node( node_id=100, labels={'Profile'}, data={ 'profileId': profile_id, 'profileType': profile_type, 'uuid': 'some-uuid', 'profileName': 'some name', }, ) session_mock = get_session_mock([{'p': profile1}]) with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.edit_full_catalog_access(profile_id, profile_type, access) assert actual assert session_mock.__enter__().run.call_count == 1 session_mock.__enter__().run.assert_called_with( ANY, profile_type=profile_type, profile_id=profile_id, access=access ) args = session_mock.__enter__().run.call_args assert args[0] == ( 'MATCH (p:Profile {profileType: $profile_type, profileId: $profile_id})\n SET p.fullCatalogAccess = $access RETURN p', ) # noqa def test_create_profile_to_resource_relationship_no_result(): """Test create_profile_to_resource_relationship, no relation created.""" schema = { 'resource_type': 'ArtistInfo', 'resource_id': 100, 'profile_type': 'ArtistProfile', 'profile_id': 500, 'roles': ['test'], } expected = {'code': 'internal_error', 'message': error.ERROR_MESSAGE_CREATE_RELATIONSHIP_FAILED} session_mock = get_session_mock([]) with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.create_profile_to_resource_relationship(schema, 'admin-id') assert not actual assert actual.errors == expected assert session_mock.__enter__().run.call_count == 2 session_mock.__enter__().run.assert_called_with(ANY, **schema) # check parts of the query. args = session_mock.__enter__().run.call_args assert 'MERGE (p)-[rel:HAS_ACCESS_TO]->(r)' in args[0][0] assert 'MATCH (p:Profile)' in args[0][0] assert ' MATCH(r:ArtistInfo)' in args[0][0] @pytest.mark.parametrize( ('expected', 'query'), [ ( { 'lastName': 'Last', 'firstName': 'First', 'name': 'First Last', 'googleUserId': 'google-123', 'active': 'Y', 'id': '12345abc', 'auth0UserId': 'auth-123', 'email': 'test@theorchard.com', 'defaultBrand': 'theorchard', 'pending': False, }, [ """MATCH (i:Identity)-[:HAS_PROFILE]->(ap:Profile)-\n [:HAS_ADMIN_ACCESS_TO]->(r)-[*0..1]->(x)<-[:HAS_ACCESS_TO]-(up:Profile)<-[:HAS_PROFILE]-(others:Identity)\n WHERE ap.profileType = 'SettingsProfile' AND i.id = $identity_id AND others.id <> $identity_id AND x.uuid IN $label_participants AND "LabelParticipant" IN labels(x)\n WITH x, ap, collect(others) as rows\n \n OPTIONAL MATCH (x)-[:OWNS]->(s:SubAccount)<-[:HAS_ACCESS_TO]-\n (up2:Profile)<-[:HAS_PROFILE]-(others2:Identity)\n WHERE up2.profileType = ap.profileType\n WITH rows, collect(others2) as allOthers\n WITH rows + allOthers as allRows\n UNWIND allRows as row\n RETURN DISTINCT row SKIP $offset LIMIT $limit """ # noqa ], ), ], ) def test_get_all_user_profiles_for_admin_success(expected, query, make_graph_node): """Test get_all_user_profiles_for_admin.""" data = [{'row': expected, 'total': 1}] session_mock = get_session_mock(data) with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.get_all_user_profiles_for_admin('12345abc', label_participants=[123, 456]) assert actual assert 'data' in actual.message assert 'total' in actual.message assert actual.message['data'] == [expected] args = session_mock.__enter__().run.call_args for query_string in query: assert query_string in args[0][0] @pytest.mark.parametrize( ( 'expected', 'query', 'label_participants', 'resource_access', 'parent_vendor_filter', 'include_subaccount_users', ), [ ( { 'lastName': 'Last', 'firstName': 'First', 'name': 'First Last', 'googleUserId': 'google-123', 'active': 'Y', 'id': '12345abc', 'auth0UserId': 'auth-123', 'email': 'test@theorchard.com', 'defaultBrand': 'theorchard', 'pending': False, }, [ """MATCH (i:Identity)-[:HAS_PROFILE]->(ap:Profile)-\n [:HAS_ADMIN_ACCESS_TO]->(r)-[*0..1]->(x)<-[:HAS_ACCESS_TO]-(up:Profile)<-[:HAS_PROFILE]-(others:Identity)\n WHERE ap.profileType = 'SettingsProfile' AND i.id = $identity_id AND others.id <> $identity_id AND up.profileType in $profile_types AND x.uuid IN $label_participants AND "LabelParticipant" IN labels(x)\n WITH x, ap, collect(others) as rows\n \n OPTIONAL MATCH (x)-[:OWNS]->(s:SubAccount)<-[:HAS_ACCESS_TO]-\n (up2:Profile)<-[:HAS_PROFILE]-(others2:Identity)\n WHERE up2.profileType = ap.profileType\n WITH rows, collect(others2) as allOthers\n WITH rows + allOthers as allRows\n UNWIND allRows as row\n RETURN DISTINCT row SKIP $offset LIMIT $limit """ # noqa ], [123, 456], [], None, True, ), ( { 'lastName': 'Last', 'firstName': 'First', 'name': 'First Last', 'googleUserId': 'google-123', 'active': 'Y', 'id': '12345abc', 'auth0UserId': 'auth-123', 'email': 'test@theorchard.com', 'defaultBrand': 'theorchard', 'pending': False, }, [ """MATCH (i:Identity)-[:HAS_PROFILE]->(ap:Profile)-\n [:HAS_ADMIN_ACCESS_TO]->(r)-[*0..1]->(x)<-[:HAS_ACCESS_TO]-(up:Profile)<-[:HAS_PROFILE]-(others:Identity)\n WHERE ap.profileType = 'SettingsProfile' AND i.id = $identity_id AND others.id <> $identity_id AND up.profileType in $profile_types AND (x:LabelParticipant OR x:Vendor OR x:Subaccount OR x:Collaborator) AND x.uuid IN $resource_access\n WITH x, ap, collect(others) as rows\n \n OPTIONAL MATCH (x)-[:OWNS]->(s:SubAccount)<-[:HAS_ACCESS_TO]-\n (up2:Profile)<-[:HAS_PROFILE]-(others2:Identity)\n WHERE up2.profileType = ap.profileType\n WITH rows, collect(others2) as allOthers\n WITH rows + allOthers as allRows\n UNWIND allRows as row\n RETURN DISTINCT row SKIP $offset LIMIT $limit """ # noqa ], [], ['uuid1', 'uuid2'], None, True, ), ( { 'lastName': 'Last', 'firstName': 'First', 'name': 'First Last', 'googleUserId': 'google-123', 'active': 'Y', 'id': '12345abc', 'auth0UserId': 'auth-123', 'email': 'test@theorchard.com', 'defaultBrand': 'theorchard', 'pending': False, }, [ """MATCH (i:Identity)-[:HAS_PROFILE]->(ap:Profile)-\n [:HAS_ADMIN_ACCESS_TO]->(r)-[*0..1]->(x)<-[:HAS_ACCESS_TO]-(up:Profile)<-[:HAS_PROFILE]-(others:Identity)\n WHERE ap.profileType = 'SettingsProfile' AND i.id = $identity_id AND others.id <> $identity_id AND up.profileType in $profile_types AND (x:LabelParticipant OR x:Vendor OR x:Subaccount OR x:Collaborator) AND x.uuid IN $resource_access\n WITH x, ap, collect(others) as rows\n \n WITH rows as allRows\n UNWIND allRows as row\n RETURN DISTINCT row SKIP $offset LIMIT $limit """ # noqa ], [], ['uuid1', 'uuid2'], None, False, ), ( { 'lastName': 'Last', 'firstName': 'First', 'name': 'First Last', 'googleUserId': 'google-123', 'active': 'Y', 'id': '12345abc', 'auth0UserId': 'auth-123', 'email': 'test@theorchard.com', 'defaultBrand': 'theorchard', 'pending': False, }, [ """MATCH (i:Identity)-[:HAS_PROFILE]->(ap:Profile)-\n [:HAS_ADMIN_ACCESS_TO]->(r)-[*0..1]->(x)<-[:HAS_ACCESS_TO]-(up:Profile)<-[:HAS_PROFILE]-(others:Identity)\n WHERE ap.profileType = 'SettingsProfile' AND i.id = $identity_id AND others.id <> $identity_id AND up.profileType in $profile_types AND x.uuid IN $label_participants AND "LabelParticipant" IN labels(x) AND r:Vendor AND r.uuid IN $parent_vendor_filter\n WITH x, ap, collect(others) as rows\n \n OPTIONAL MATCH (x)-[:OWNS]->(s:SubAccount)<-[:HAS_ACCESS_TO]-\n (up2:Profile)<-[:HAS_PROFILE]-(others2:Identity)\n WHERE up2.profileType = ap.profileType\n WITH rows, collect(others2) as allOthers\n WITH rows + allOthers as allRows\n UNWIND allRows as row\n RETURN DISTINCT row SKIP $offset LIMIT $limit """ # noqa ], [123, 456], [], ['c46005a7-ad3a-434b-b3ba-3d8a1c44c0a5'], True, ), ], ) def test_get_user_profiles_for_admin_by_profile_type_success( expected, query, label_participants, resource_access, parent_vendor_filter, make_graph_node, include_subaccount_users, ): """Test get_all_user_profiles_for_admin by profile_type.""" data = [{'row': expected, 'total': 1}] session_mock = get_session_mock(data) with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.get_all_user_profiles_for_admin( '12345abc', ['LabelProfile', 'InsightsProfile'], label_participants=label_participants, resource_access=resource_access, parent_vendor_filter=parent_vendor_filter, include_subaccount_users=include_subaccount_users, ) assert actual assert 'data' in actual.message assert 'total' in actual.message assert actual.message['data'] == [expected] args = session_mock.__enter__().run.call_args for query_string in query: assert query_string in args[0][0] def test_get_resources_from_identity_by_profile_type_no_result(): """Test get_all_user_profiles_for_admin by_profile_type, no result.""" data = [{'total': 0}] session_mock = get_session_mock(data) with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.get_all_user_profiles_for_admin('12345abc', 'LabelProfile') assert actual assert 'data' in actual.message assert 'total' in actual.message assert actual.message['data'] == [] @pytest.mark.parametrize( ('active', 'pending', 'expected_where'), [ (None, None, 'i.id = $identity_id AND others.id <> $identity_id'), ( 'Y', None, 'i.id = $identity_id AND others.id <> $identity_id AND others.active = $active', # noqa ), ( 'Y', 'N', 'i.id = $identity_id AND others.id <> $identity_id AND NOT others.id = others.auth0UserId AND others.active = $active', # noqa ), ( 'Y', 'Y', 'i.id = $identity_id AND others.id <> $identity_id AND others.id = others.auth0UserId AND others.active = $active', # noqa ), ( 'N', 'Y', 'i.id = $identity_id AND others.id <> $identity_id AND others.id = others.auth0UserId AND others.active = $active', # noqa ), ( 'N', 'N', 'i.id = $identity_id AND others.id <> $identity_id AND NOT others.id = others.auth0UserId AND others.active = $active', # noqa ), ( 'N', None, 'i.id = $identity_id AND others.id <> $identity_id AND others.active = $active', # noqa ), ( None, 'N', 'i.id = $identity_id AND others.id <> $identity_id AND NOT others.id = others.auth0UserId', # noqa ), ( None, 'Y', 'i.id = $identity_id AND others.id <> $identity_id AND others.id = others.auth0UserId', # noqa ), ], ) def test_filters_get_all_user_profiles_for_admin(active, pending, expected_where): """Test get_all_user_profiles query for optional filters.""" data = [ { 'row': { 'lastName': 'Last', 'firstName': 'First', 'name': 'First Last', 'googleUserId': 'google-123', 'active': 'Y', 'id': '12345abc', 'auth0UserId': 'auth-id-mock', 'email': 'test@theorchard.com', 'defaultBrand': 'theorchard', }, 'total': 1, } ] if pending == 'Y': data[0]['row']['auth0UserId'] = '12345abc' session_mock = get_session_mock(data) with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.get_all_user_profiles_for_admin( '12345abc', None, None, None, None, None, active, pending, None, None, True ) assert actual args = session_mock.__enter__().run.call_args assert expected_where in args[0][0] @pytest.mark.parametrize( ('active', 'pending', 'expected_where'), [ (None, None, 'true AND NOT EXISTS((ps)-[:HAS_ADMIN_ACCESS_TO]->(v))'), ( 'Y', None, "MATCH (i:Identity)-[:HAS_PROFILE]->(ps:Profile {profileType:'SettingsProfile'}), (v:Vendor {id: '*'}) WITH i, ps, v WHERE true AND i.active = $active AND NOT EXISTS((ps)-[:HAS_ADMIN_ACCESS_TO]->(v)) RETURN DISTINCT i as identity SKIP $offset LIMIT $limit", # noqa ), ( 'Y', 'N', "MATCH (i:Identity)-[:HAS_PROFILE]->(ps:Profile {profileType:'SettingsProfile'}), (v:Vendor {id: '*'}) WITH i, ps, v WHERE true AND NOT i.id = i.auth0UserId AND i.active = $active AND NOT EXISTS((ps)-[:HAS_ADMIN_ACCESS_TO]->(v)) RETURN DISTINCT i as identity SKIP $offset LIMIT $limit", # noqa ), ( 'Y', 'Y', "MATCH (i:Identity)-[:HAS_PROFILE]->(ps:Profile {profileType:'SettingsProfile'}), (v:Vendor {id: '*'}) WITH i, ps, v WHERE true AND i.id = i.auth0UserId AND i.active = $active AND NOT EXISTS((ps)-[:HAS_ADMIN_ACCESS_TO]->(v)) RETURN DISTINCT i as identity SKIP $offset LIMIT $limit", # noqa ), ( 'N', 'Y', "MATCH (i:Identity)-[:HAS_PROFILE]->(ps:Profile {profileType:'SettingsProfile'}), (v:Vendor {id: '*'}) WITH i, ps, v WHERE true AND i.id = i.auth0UserId AND i.active = $active AND NOT EXISTS((ps)-[:HAS_ADMIN_ACCESS_TO]->(v)) RETURN DISTINCT i as identity SKIP $offset LIMIT $limit", # noqa ), ( 'N', 'N', "MATCH (i:Identity)-[:HAS_PROFILE]->(ps:Profile {profileType:'SettingsProfile'}), (v:Vendor {id: '*'}) WITH i, ps, v WHERE true AND NOT i.id = i.auth0UserId AND i.active = $active AND NOT EXISTS((ps)-[:HAS_ADMIN_ACCESS_TO]->(v)) RETURN DISTINCT i as identity SKIP $offset LIMIT $limit", # noqa ), ( 'N', None, "MATCH (i:Identity)-[:HAS_PROFILE]->(ps:Profile {profileType:'SettingsProfile'}), (v:Vendor {id: '*'}) WITH i, ps, v WHERE true AND i.active = $active AND NOT EXISTS((ps)-[:HAS_ADMIN_ACCESS_TO]->(v)) RETURN DISTINCT i as identity SKIP $offset LIMIT $limit", # noqa ), ( None, 'N', 'true AND NOT i.id = i.auth0UserId AND NOT EXISTS((ps)-[:HAS_ADMIN_ACCESS_TO]->(v))', ), ( None, 'Y', 'MATCH (i:Identity)-[:HAS_PROFILE]->(ps:Profile ' "{profileType:'SettingsProfile'}), (v:Vendor {id: '*'}) WITH i, ps, v WHERE " 'true AND i.id = i.auth0UserId AND NOT EXISTS((ps)-[:HAS_ADMIN_ACCESS_TO]->(v)) ' 'RETURN DISTINCT i as identity SKIP $offset LIMIT $limit', ), ], ) def test_filters_get_all_profiles_with_artist_access(active, pending, expected_where): """Test get_all_profiles_with_artist_access query for optional filters.""" data = [ { 'row': { 'lastName': 'Last', 'firstName': 'First', 'name': 'First Last', 'googleUserId': 'google-123', 'active': 'Y', 'id': '12345abc', 'auth0UserId': 'auth-id-mock', 'auth0UserCreatedBy': 'invitation', 'updatedOn': '2021-09-16T06:40:05.595000000Z', 'createdAt': '2021-09-16T06:40:05.595000000Z', 'email': 'test@theorchard.com', 'defaultBrand': 'theorchard', }, 'total': 1, } ] if pending == 'Y': data[0]['row']['auth0UserId'] = '12345abc' session_mock = get_session_mock(data) with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.get_all_profiles_with_artist_access( '12345abc', None, None, None, None, None, active, pending, None, False ) assert actual args = session_mock.__enter__().run.call_args assert expected_where in args[0][0] @pytest.mark.parametrize( ('resource_type', 'query'), [ ( None, 'MATCH (p:Profile)-[r:HAS_ACCESS_TO]->(x)\n WHERE\n \n p.profileType = $profile_type AND\n p.profileId = $profile_id\n WITH x,r\n OPTIONAL MATCH (x)<-[:OWNS]-(v:Vendor)\n RETURN\n x as resource,\n apoc.coll.toSet(apoc.coll.flatten(collect(r.roles))) as roles,\n v.id as vendor_id\n ', # noqa ), ( 'ArtistInfo', 'MATCH (p:Profile)-[r:HAS_ACCESS_TO]->(x)\n WHERE\n x:ArtistInfo AND \n p.profileType = $profile_type AND\n p.profileId = $profile_id\n WITH x,r\n OPTIONAL MATCH (x)<-[:OWNS]-(v:Vendor)\n RETURN\n x as resource,\n apoc.coll.toSet(apoc.coll.flatten(collect(r.roles))) as roles,\n v.id as vendor_id\n ', # noqa ), ], ) def test_get_resources_for_profile(resource_type, query, make_graph_node): """Test get_resources_for_profile nodes.""" profile_type = 'InsightsProfile' profile_id = 1001 artist1 = make_graph_node(node_id=100, labels={'ArtistInfo'}) artist2 = make_graph_node(node_id=200, labels={'ArtistInfo'}) release1 = make_graph_node(node_id=202, labels={'Release'}) vendor1 = make_graph_node(node_id=7123, labels={'Vendor', 'Label'}) collab1 = make_graph_node(node_id=300, labels={'Collaborator'}) session_mock = get_session_mock( [ {'resource': artist1, 'roles': ['role1']}, {'resource': artist2}, {'resource': vendor1}, {'resource': release1, 'roles': ['role1']}, {'resource': collab1, 'roles': ['role1']}, ] ) expected = [ {'type': 'ArtistInfo', 'id': 100, 'roles': ['role1']}, {'type': 'ArtistInfo', 'id': 200, 'roles': None}, {'type': 'Vendor', 'id': 7123, 'roles': None}, {'type': 'Collaborator', 'id': 300, 'roles': ['role1']}, ] with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.get_resources_for_profile(profile_type, profile_id, resource_type) assert actual assert actual.message == expected session_mock.__enter__().run.assert_called_once_with( query, profile_id=profile_id, profile_type=profile_type ) @pytest.mark.parametrize( ('profile_type', 'resource_data', 'expected'), [ [ 'SettingsProfile', { 'node_id': '*', 'labels': {'Label', 'Vendor', 'Orchard'}, 'data': {'name': 'All Orchard Labels', 'id': '*'}, }, response.Response( { 'data': [ { 'id': '*', 'name': 'All Orchard Labels', 'profile_access': {'roles': ['administrator']}, 'type': 'Vendor', } ], 'total': 1, } ), ], [ 'InsightsProfile', { 'node_id': '*', 'labels': {'Label', 'Vendor', 'Orchard'}, 'data': {'name': 'All Orchard Labels', 'id': '*'}, }, response.Response( { 'data': [ { 'id': '*', 'name': 'All Orchard Labels', 'profile_access': {'roles': ['administrator']}, 'type': 'Vendor', } ], 'total': 1, } ), ], [ 'SettingsProfile', None, response.create_not_found_response(error.ERROR_MESSAGE_FORBIDDEN_USER), ], [ 'InsightsProfile', None, response.create_not_found_response(error.ERROR_MESSAGE_FORBIDDEN_USER), ], ], ) def test_get_vend_star_for_admin(make_graph_node, profile_type, resource_data, expected): """Test get_vend_star_for_admin.""" identity_id = 'admin-uuid-mock' profile_id = 1234 if resource_data: vend_mock = make_graph_node( resource_data['node_id'], resource_data['labels'], resource_data['data'] ) session_mock = get_session_mock([{'resource': vend_mock}]) else: session_mock = get_session_mock() with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.get_vend_star_for_admin(identity_id, profile_type, profile_id) assert actual.status == expected.status assert actual.message == expected.message @pytest.mark.parametrize( ( 'admin_profile_type', 'user_profile_type', 'resource_data', 'rel_type', 'songwhip_enabled', 'expected', ), [ [ 'SettingsProfile', 'SettingsProfile', { 'node_id': '*', 'labels': {'Label', 'Vendor', 'Orchard'}, 'data': {'name': 'All Orchard Labels', 'id': '*'}, }, 'HAS_ADMIN_ACCESS_TO', True, response.Response( { 'data': [ { 'id': '*', 'name': 'All Orchard Labels', 'profile_access': {'roles': ['administrator']}, 'type': 'Vendor', } ], 'total': 1, } ), ], [ 'SettingsProfile', 'InsightsProfile', { 'node_id': '*', 'labels': {'Label', 'Vendor', 'Orchard'}, 'data': {'name': 'All Orchard Labels', 'id': '*'}, }, 'HAS_ACCESS_TO', True, response.Response( { 'data': [ { 'id': '*', 'name': 'All Orchard Labels', 'profile_access': {'roles': ['analytics']}, 'type': 'Vendor', } ], 'total': 1, } ), ], [ 'SettingsProfile', 'SongwhipProfile', { 'node_id': '*', 'labels': {'Label', 'Vendor', 'Orchard'}, 'data': {'name': 'All Orchard Labels', 'id': '*'}, }, 'HAS_ACCESS_TO', True, response.Response( { 'data': [ { 'id': '*', 'name': 'All Orchard Labels', 'profile_access': {'roles': ['songwhip']}, 'type': 'Vendor', } ], 'total': 1, } ), ], [ 'SettingsProfile', 'SongwhipProfile', { 'node_id': '*', 'labels': {'Label', 'Vendor', 'Orchard'}, 'data': {'name': 'All Orchard Labels', 'id': '*'}, }, 'HAS_ACCESS_TO', False, response.create_not_found_response(error.ERROR_MESSAGE_FORBIDDEN_USER), ], [ 'SettingsProfile', 'NotAllowedProfile', { 'node_id': '*', 'labels': {'Label', 'Vendor', 'Orchard'}, 'data': {'name': 'All Orchard Labels', 'id': '*'}, }, 'HAS_ACCESS_TO', True, response.create_not_found_response(error.ERROR_MESSAGE_FORBIDDEN_USER), ], [ 'SettingsProfile', 'SettingsProfile', None, None, True, response.create_not_found_response(error.ERROR_MESSAGE_FORBIDDEN_USER), ], [ 'LabelProfile', 'SettingsProfile', None, None, True, response.create_error_response( error.ERROR_CODE_AUTHORIZATION, error.ERROR_MESSAGE_INVALID_PROFILE ), ], ], ) def test_get_user_vend_star( make_graph_node, admin_profile_type, user_profile_type, resource_data, rel_type, songwhip_enabled, expected, ): """Test get_user_vend_star.""" identity_id = 'non-admin-uuid-mock' admin_context = { 'identity_id': 'admin-uuid', 'profile_type': admin_profile_type, 'profile_id': 12, } allowed_profile_types = ['SettingsProfile', 'InsightsProfile'] if songwhip_enabled: allowed_profile_types.append('SongwhipProfile') if resource_data and user_profile_type in allowed_profile_types: vend_mock = make_graph_node( resource_data['node_id'], resource_data['labels'], resource_data['data'] ) session_mock = get_session_mock( [ { 'resource': vend_mock, 'resource_relationship': rel_type, 'profile_type': user_profile_type, } ] ) else: session_mock = get_session_mock() with patch.object(neo4j, 'db_session', return_value=session_mock), patch.object( pythonfeatures, 'get_single_feature_by_attributes', return_value=response.Response(message=('enabled' if songwhip_enabled else 'control')), ): actual = resource.get_user_vend_star(admin_context, identity_id) assert actual.status == expected.status assert actual.message == expected.message assert actual.errors == expected.errors if resource_data and user_profile_type in allowed_profile_types: assert actual.message assert actual.message['data'] assert actual.message['data'] == expected.message['data'] else: assert actual.errors assert actual.errors['code'] assert actual.errors == expected.errors @pytest.mark.parametrize('active', [True, False]) def test_get_user_resources_for_admin(make_graph_node, active): """Test get_user_resources_for_admin.""" artist1 = make_graph_node(node_id=100, labels={'ArtistInfo'}, data={'name': 'Artist1'}) artist2 = make_graph_node(node_id=200, labels={'ArtistInfo'}, data={'name': 'Artist2'}) session_mock = get_session_mock( [ {'x': artist1, 'roles': ['foo'], 'updatedOn': '2020-01-01', 'total': 2}, {'x': artist2, 'roles': ['bar'], 'total': 2}, ] ) expected = { 'data': [ { 'id': 100, 'name': 'Artist1', 'type': 'ArtistInfo', 'profile_access': {'roles': ['foo'], 'updatedOn': '2020-01-01'}, }, { 'id': 200, 'name': 'Artist2', 'type': 'ArtistInfo', 'profile_access': {'roles': ['bar'], 'updatedOn': None}, }, ], 'total': 2, } admin_context = { 'identity_id': 'admin-uuid', 'profile_type': 'InsightsProfile', 'profile_id': 12, } identity_id = 'user-uuid' resource_type = 'ArtistInfo' if active: access_type = 'HAS_ACCESS_TO|HAS_ADMIN_ACCESS_TO' else: access_type = 'DELETED_HAS_ACCESS_TO' with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.get_user_resources_for_admin( identity_id, admin_context['profile_type'], resource_type, 50, 0, active ) assert actual assert actual.message == expected args = session_mock.__enter__().run.call_args assert args[1] == dict( userIdentityId=identity_id, profileTypes=admin_context['profile_type'], offset=0, limit=50, active=active, ) assert args[0] == ( f'MATCH (user:Identity)-[:HAS_PROFILE]->(up:Profile)-[rel:{access_type}]->(x:ArtistInfo)\n WHERE user.id = $userIdentityId AND up.profileType IN $profileTypes\n OPTIONAL MATCH (x)<-[:OWNS]-(v:Vendor)\n \n RETURN DISTINCT x, v.id as vendorId, TRIM(v.name) as vendorName,\n apoc.coll.toSet(apoc.coll.flatten(collect(\n CASE\n WHEN up.profileType = "InsightsProfile" AND (x:Vendor OR x:Subaccount)\n THEN ["analytics"]\n WHEN up.profileType = "SettingsProfile" AND type(rel)= "HAS_ADMIN_ACCESS_TO"\n THEN ["administrator"]\n ELSE up.roles\n END))) as roles,\n apoc.coll.max(collect(up.updatedOn)) as updatedOn\n SKIP $offset\n LIMIT $limit\n ', ) # noqa def test_get_users_resources_by_type(make_graph_node): """Test get_users_resources_by_type.""" artist1 = make_graph_node(node_id=100, labels={'ArtistInfo'}, data={'name': 'Artist1'}) artist2 = make_graph_node(node_id=200, labels={'ArtistInfo'}, data={'name': 'Artist2'}) session_mock = get_session_mock( [ {'x': artist1, 'roles': ['foo'], 'updatedOn': '2020-01-01', 'total': 2}, {'x': artist2, 'roles': ['bar'], 'total': 2}, ] ) expected = { 'data': [ { 'id': 100, 'name': 'Artist1', 'type': 'ArtistInfo', 'profile_access': {'roles': ['foo'], 'updatedOn': '2020-01-01'}, }, { 'id': 200, 'name': 'Artist2', 'type': 'ArtistInfo', 'profile_access': {'roles': ['bar'], 'updatedOn': None}, }, ], 'total': 2, } admin_context = { 'identity_id': 'admin-uuid', 'profile_type': 'InsightsProfile', 'profile_id': 12, } identity_id = 'user-uuid' resource_type = 'ArtistInfo' with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.get_users_resources_by_type(admin_context, identity_id, resource_type) assert actual assert actual.message == expected args = session_mock.__enter__().run.call_args assert args[1] == dict( adminIdentityId=admin_context['identity_id'], adminProfileType=admin_context['profile_type'], adminProfileId=admin_context['profile_id'], userIdentityId=identity_id, offset=0, limit=50, ) assert args[0] == ( 'MATCH (admin:Identity)-[:HAS_PROFILE]->(ap:Profile)-[:HAS_ACCESS_TO]->\n (r)-[*0..1]->(x:ArtistInfo)<-[:HAS_ACCESS_TO]-(up:Profile)<-[:HAS_PROFILE]-(user:Identity)\n WHERE\n admin.id = $adminIdentityId\n AND \'administrator\' IN ap.roles\n AND ap.profileType = $adminProfileType\n AND ap.profileId = $adminProfileId\n AND user.id = $userIdentityId\n AND up.profileType = $adminProfileType\n WITH x, up\n OPTIONAL MATCH (x)<-[:OWNS]-(v:Vendor)\n \n RETURN DISTINCT x, v.id as vendorId, TRIM(v.name) as vendorName,\n apoc.coll.toSet(apoc.coll.flatten(collect(\n CASE WHEN up.profileType = "InsightsProfile" and (x:Vendor OR x:Subaccount)\n THEN [] ELSE up.roles\n END))) as roles,\n apoc.coll.max(collect(up.updatedOn)) as updatedOn\n SKIP $offset\n LIMIT $limit\n ', ) # noqa def test_get_users_resources_for_settings_by_type(make_graph_node): """Test get_users_resources_for_settings_by_type.""" artist1 = make_graph_node(node_id=100, labels={'ArtistInfo'}, data={'name': 'Artist1'}) artist2 = make_graph_node(node_id=200, labels={'ArtistInfo'}, data={'name': 'Artist2'}) session_mock = get_session_mock( [ {'x': artist1, 'roles': ['foo'], 'updatedOn': '2020-01-01', 'total': 2}, {'x': artist2, 'roles': ['bar'], 'total': 2}, ] ) expected = { 'data': [ { 'id': 100, 'name': 'Artist1', 'type': 'ArtistInfo', 'profile_access': {'roles': ['foo'], 'updatedOn': '2020-01-01'}, }, { 'id': 200, 'name': 'Artist2', 'type': 'ArtistInfo', 'profile_access': {'roles': ['bar'], 'updatedOn': None}, }, ], 'total': 2, } admin_context = { 'identity_id': 'admin-uuid', 'profile_type': 'SettingsProfile', 'profile_id': 12, } identity_id = 'user-uuid' resource_type = 'ArtistInfo' with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.get_users_resources_for_settings_by_type( admin_context, identity_id, constants.SETTINGS_SUPPORT_MAPPING['profileTypes'], resource_type, ) assert actual assert actual.message == expected args = session_mock.__enter__().run.call_args assert args[1] == dict( adminIdentityId=admin_context['identity_id'], adminProfileType=admin_context['profile_type'], adminProfileId=admin_context['profile_id'], profileTypes=constants.SETTINGS_SUPPORT_MAPPING['profileTypes'], userIdentityId=identity_id, offset=0, limit=50, ) assert args[0] == ( "MATCH (admin:Identity)-[:HAS_PROFILE]->(sp:Profile {profileType: 'SettingsProfile'})\n -[:HAS_ADMIN_ACCESS_TO]->(r)-[*0..1]->(x:ArtistInfo)<-[:HAS_ACCESS_TO]-(up:Profile)<-[:HAS_PROFILE]-(user:Identity)\n USING JOIN ON r\n WHERE\n admin.id = $adminIdentityId AND sp.profileId = $adminProfileId\n AND user.id = $userIdentityId\n AND up.profileType IN $profileTypes\n WITH x,up, user\n OPTIONAL MATCH (x)<-[:OWNS]-(v:Vendor)\n \n RETURN DISTINCT x, v.id as vendorId, TRIM(v.name) as vendorName,\n apoc.coll.toSet(apoc.coll.flatten(collect(\n CASE\n WHEN up.profileType = \"InsightsProfile\" and (x:Vendor OR x:Subaccount) THEN []\n WHEN (x)<-[:HAS_ADMIN_ACCESS_TO]-(:Profile {profileType: 'SettingsProfile'})\n <-[:HAS_PROFILE]-(user) THEN ['administrator']\n ELSE up.roles\n END))) as roles,\n apoc.coll.max(collect(up.updatedOn)) as updatedOn\n SKIP $offset\n LIMIT $limit\n ", ) # noqa @pytest.mark.parametrize( ('profile', 'expected', 'query'), [ ( {'identity': {'id': 1, 'name': 'abc', 'pending': False}, 'total': 1}, {'data': [{'id': 1, 'name': 'abc', 'pending': False}], 'total': 1}, ( "MATCH (i:Identity)-[:HAS_PROFILE]->(up:Profile), (v:Vendor {id: '*'}) WITH i, up, v WHERE true AND NOT EXISTS((i)-[:HAS_PROFILE]->(:Profile {profileType:'SettingsProfile'})-[:HAS_ADMIN_ACCESS_TO]->(v)) AND up.profileType in $profile_types RETURN DISTINCT i as identity SKIP $offset LIMIT $limit", ), # noqa ), ], ) def test_get_all_profiles_with_artist_access(profile, expected, query, make_graph_node): """Test get_all_profiles_with_artist_access.""" session_mock = get_session_mock([profile]) identity_id = 'user-uuid' profile_types = ['InsightsProfile'] with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.get_all_profiles_with_artist_access(identity_id, profile_types) assert actual assert actual.message == expected args = session_mock.__enter__().run.call_args assert args[1] == dict( profile_types=profile_types, active=None, admin_id=identity_id, resource_access=None, search_term=None, label_participants=None, limit=50, offset=0, settings_support_profiles=[ 'AbacusProfile', 'InsightsProfile', 'SettingsProfile', 'LabelProfile', 'MoneyhubProfile', 'CollaboratorsProfile', 'DocumentsProfile', 'SongwhipProfile', 'AudienceProfile', ], ) assert args[0] == query @pytest.mark.parametrize( ('term', 'profile_types', 'label_participants', 'resource_access', 'query'), [ ( None, None, None, None, ( 'MATCH (i:Identity) WHERE true RETURN DISTINCT i as identity SKIP $offset LIMIT $limit', ), # noqa ), ( None, ['InsightsProfile'], None, None, ( 'MATCH (i:Identity)-[:HAS_PROFILE]->(up:Profile) WHERE true AND up.profileType in $profile_types RETURN DISTINCT i as identity SKIP $offset LIMIT $limit', ), # noqa ), ( 'michael', ['InsightsProfile'], None, None, ( 'MATCH (i:Identity)-[:HAS_PROFILE]->(up:Profile) WHERE true AND up.profileType in $profile_types AND (toLower(i.email) CONTAINS $search_term OR toLower(i.name) CONTAINS $search_term OR toLower(i.firstName) CONTAINS $search_term OR toLower(i.lastName) CONTAINS $search_term OR toLower(i.firstName + " " + i.lastName) CONTAINS $search_term OR toLower(i.lastName + " " + i.firstName) CONTAINS $search_term ) RETURN DISTINCT i as identity SKIP $offset LIMIT $limit', ), # noqa ), ( None, ['InsightsProfile'], ['17197be2-5b74-4a1f-8e6a-cee7740e40c8'], None, ( 'MATCH (i:Identity)-[:HAS_PROFILE]->(up:Profile)-[:HAS_ACCESS_TO]->(x) WHERE true AND up.profileType in $profile_types AND x:LabelParticipant AND x.uuid IN $label_participants RETURN DISTINCT i as identity SKIP $offset LIMIT $limit', ), # noqa ), ( None, ['InsightsProfile'], None, ['fff741c2-6def-4493-bfdf-c2bcb1128e02'], ( 'MATCH (i:Identity)-[:HAS_PROFILE]->(up:Profile)-[:HAS_ACCESS_TO]->(x) WHERE true AND up.profileType in $profile_types AND (x:LabelParticipant OR x:Vendor OR x:Subaccount OR x:Collaborator) AND x.uuid IN $resource_access RETURN DISTINCT i as identity SKIP $offset LIMIT $limit', ), # noqa ), ( 'michael', ['InsightsProfile'], ['17197be2-5b74-4a1f-8e6a-cee7740e40c8'], ['fff741c2-6def-4493-bfdf-c2bcb1128e02'], ( 'MATCH (i:Identity)-[:HAS_PROFILE]->(up:Profile)-[:HAS_ACCESS_TO]->(x) WHERE true AND up.profileType in $profile_types AND (toLower(i.email) CONTAINS $search_term OR toLower(i.name) CONTAINS $search_term OR toLower(i.firstName) CONTAINS $search_term OR toLower(i.lastName) CONTAINS $search_term OR toLower(i.firstName + " " + i.lastName) CONTAINS $search_term OR toLower(i.lastName + " " + i.firstName) CONTAINS $search_term ) AND x:LabelParticipant AND x.uuid IN $label_participants AND (x:LabelParticipant OR x:Vendor OR x:Subaccount OR x:Collaborator) AND x.uuid IN $resource_access RETURN DISTINCT i as identity SKIP $offset LIMIT $limit', ), # noqa ), ( None, None, ['17197be2-5b74-4a1f-8e6a-cee7740e40c8'], ['fff741c2-6def-4493-bfdf-c2bcb1128e02'], ( 'MATCH (i:Identity)-[:HAS_PROFILE]->(up:Profile)-[:HAS_ACCESS_TO]->(x) WHERE true AND up.profileType in $settings_support_profiles AND x:LabelParticipant AND x.uuid IN $label_participants AND (x:LabelParticipant OR x:Vendor OR x:Subaccount OR x:Collaborator) AND x.uuid IN $resource_access RETURN DISTINCT i as identity SKIP $offset LIMIT $limit', ), # noqa ), ], ) def test_get_all_profiles_with_artist_access_feature_on( term, profile_types, label_participants, resource_access, query, make_graph_node ): """Test get_all_profiles_with_artist_access.""" profile = { 'identity': { 'id': 1, 'name': 'abc', }, 'total': 1, } session_mock = get_session_mock([profile]) identity_id = 'user-uuid' with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.get_all_profiles_with_artist_access( identity_id, profile_types, search_term=term, label_participants=label_participants, resource_access=resource_access, feature_flag=True, ) assert actual assert actual.message == {'data': [{'id': 1, 'name': 'abc', 'pending': False}], 'total': 1} args = session_mock.__enter__().run.call_args assert args[1] == dict( profile_types=profile_types, active=None, admin_id=identity_id, resource_access=resource_access, search_term=term, label_participants=label_participants, limit=50, offset=0, settings_support_profiles=[ 'AbacusProfile', 'InsightsProfile', 'SettingsProfile', 'LabelProfile', 'MoneyhubProfile', 'CollaboratorsProfile', 'DocumentsProfile', 'SongwhipProfile', 'AudienceProfile', ], ) assert args[0] == query def test_get_all_profiles_search(): """Test get_all_profiles_with_artist_access search.""" profile = { 'identity': { 'id': 1, 'name': 'abc', }, 'total': 1, } expected = {'data': [{'id': 1, 'name': 'abc', 'pending': False}], 'total': 1} session_mock = get_session_mock([profile]) identity_id = 'user-uuid' profile_types = ['InsightsProfile'] term = 'alice' with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.get_all_profiles_with_artist_access( identity_id, profile_types, search_term=term ) assert actual assert actual.message == expected args = session_mock.__enter__().run.call_args assert args[1] == dict( profile_types=profile_types, active=None, admin_id=identity_id, resource_access=None, search_term=term, label_participants=None, limit=50, offset=0, settings_support_profiles=[ 'AbacusProfile', 'InsightsProfile', 'SettingsProfile', 'LabelProfile', 'MoneyhubProfile', 'CollaboratorsProfile', 'DocumentsProfile', 'SongwhipProfile', 'AudienceProfile', ], ) assert ( 'MATCH (i:Identity)-[:HAS_PROFILE]->(up:Profile), (v:Vendor {id: \'*\'}) WITH i, up, v WHERE true AND NOT EXISTS((i)-[:HAS_PROFILE]->(:Profile {profileType:\'SettingsProfile\'})-[:HAS_ADMIN_ACCESS_TO]->(v)) AND up.profileType in $profile_types AND (toLower(i.email) CONTAINS $search_term OR toLower(i.name) CONTAINS $search_term OR toLower(i.firstName) CONTAINS $search_term OR toLower(i.lastName) CONTAINS $search_term OR toLower(i.firstName + " " + i.lastName) CONTAINS $search_term OR toLower(i.lastName + " " + i.firstName) CONTAINS $search_term ) RETURN DISTINCT i as identity SKIP $offset LIMIT $limit' in args[0][0] ) # noqa assert ( "NOT EXISTS((i)-[:HAS_PROFILE]->(:Profile {profileType:'SettingsProfile'})-[:HAS_ADMIN_ACCESS_TO]->(v))" in args[0][0] ) # noqa assert ( '(toLower(i.email) CONTAINS $search_term OR toLower(i.name) CONTAINS $search_term OR toLower(i.firstName) CONTAINS $search_term OR toLower(i.lastName) CONTAINS $search_term OR toLower(i.firstName + " " + i.lastName) CONTAINS $search_term OR toLower(i.lastName + " " + i.firstName) CONTAINS $search_term' in args[0][0] ) # noqa def test_get_artists_for_lp_by_profile_uuid(make_graph_node): """Test get_artists_for_lp_by_profile_uuid.""" artist1 = make_graph_node(node_id=100, labels={'ArtistInfo'}, data={'name': 'Artist1'}) artist2 = make_graph_node(node_id=200, labels={'ArtistInfo'}, data={'name': 'Artist2'}) session_mock = get_session_mock( [{'resource': artist1, 'roles': ['foo']}, {'resource': artist2, 'roles': ['bar']}] ) profile_uuid = 'some-profile-uuid' with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.get_artists_for_lp_by_profile_uuid(profile_uuid) assert actual == [ {'id': 100, 'name': 'Artist1', 'roles': ['foo'], 'type': 'ArtistInfo'}, {'id': 200, 'name': 'Artist2', 'roles': ['bar'], 'type': 'ArtistInfo'}, ] args = session_mock.__enter__().run.call_args assert args[1] == dict(profile_uuid=profile_uuid) assert args[0] == ( 'MATCH (p:Profile)-[r:HAS_ACCESS_TO]->(lp:LabelParticipant)\n -[:CREATED_FROM]->(a:ArtistInfo)\n WHERE\n p.uuid = $profile_uuid\n RETURN\n a as resource,\n apoc.coll.toSet(apoc.coll.flatten(collect(r.roles)+collect(p.roles))) as roles\n ', ) # noqa @pytest.mark.parametrize( ('resource_types', 'cypher'), [ [ ['Vendor', 'SubAccount'], textwrap.dedent(""" MATCH (p:Profile)-[r:HAS_ACCESS_TO|HAS_ADMIN_ACCESS_TO]->(x) WHERE p.uuid = $profile_uuid AND (x:Vendor OR x:SubAccount) AND NOT x:ParentCompany with p, r, x OPTIONAL MATCH (x)<-[:OWNS]-(v:Vendor) RETURN x as resource, apoc.coll.toSet(apoc.coll.flatten(collect(r.roles)+collect(p.roles))) as roles, v.id as vendorId, TRIM(v.name) as vendorName """), ], [ ['ArtistInfo'], textwrap.dedent(""" MATCH (p:Profile)-[r:HAS_ACCESS_TO|HAS_ADMIN_ACCESS_TO]->(x) WHERE p.uuid = $profile_uuid AND (x:ArtistInfo) AND NOT x:ParentCompany with p, r, x OPTIONAL MATCH (x)<-[:OWNS]-(v:Vendor) RETURN x as resource, apoc.coll.toSet(apoc.coll.flatten(collect(r.roles)+collect(p.roles))) as roles, v.id as vendorId, TRIM(v.name) as vendorName """), ], [ ['Collaborator'], textwrap.dedent(""" MATCH (p:Profile)-[r:HAS_ACCESS_TO|HAS_ADMIN_ACCESS_TO]->(x) WHERE p.uuid = $profile_uuid AND (x:Collaborator) AND NOT x:ParentCompany with p, r, x OPTIONAL MATCH (x)<-[:OWNS]-(v:Vendor) RETURN x as resource, apoc.coll.toSet(apoc.coll.flatten(collect(r.roles)+collect(p.roles))) as roles, v.id as vendorId, TRIM(v.name) as vendorName """), ], [ [], textwrap.dedent(""" MATCH (p:Profile)-[r:HAS_ACCESS_TO|HAS_ADMIN_ACCESS_TO]->(x) WHERE p.uuid = $profile_uuid AND NOT x:ParentCompany with p, r, x OPTIONAL MATCH (x)<-[:OWNS]-(v:Vendor) RETURN x as resource, apoc.coll.toSet(apoc.coll.flatten(collect(r.roles)+collect(p.roles))) as roles, v.id as vendorId, TRIM(v.name) as vendorName """), ], ], ) def test_get_resources_of_type_for_profile_uuid(resource_types, cypher, make_graph_node): """Test get_resources_of_type_for_profile_uuid.""" profile_uuid = 'some-profile-uuid' artist1 = make_graph_node(node_id=100, labels={'Vendor', 'Label'}) artist2 = make_graph_node(node_id=200, labels={'ArtistInfo'}) subaccount1 = make_graph_node(node_id=300, labels={'SubAccount'}) collab1 = make_graph_node(node_id=400, labels={'Collaborator'}) session_mock = get_session_mock( [ {'resource': artist1, 'roles': ['foo']}, {'resource': artist2, 'roles': ['bar']}, {'resource': subaccount1, 'roles': ['bar']}, {'resource': collab1, 'roles': ['bar']}, ] ) with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.get_resources_of_type_for_profile_uuid(profile_uuid, resource_types) assert actual args = session_mock.__enter__().run.call_args assert args[0][0] == cypher @pytest.mark.parametrize( ('limit', 'offset'), [(constants.DEFAULT_LIMIT, constants.DEFAULT_OFFSET), (1, 2), (2, 1)] ) def test_get_resources_for_identity_uuid(make_graph_node, limit, offset): """Test get_resources_for_identity_uuid.""" identity_uuid = 'some-uuid' vendor_1 = make_graph_node( node_id=500, labels={'Vendor', 'Label'}, data={'vendorId': 16055, 'id': 16055, 'uuid': 'uuid-500'}, ) subaccount_1 = make_graph_node( node_id=501, labels={'SubAccount'}, data={'id': 62776, 'uuid': 'uuid-501'} ) subaccount_2 = make_graph_node( node_id=502, labels={'Subaccount'}, data={'id': 62778, 'uuid': 'uuid-502'} ) company_brand_1 = make_graph_node( node_id=503, labels={'CompanyBrand'}, data={'uuid': 'uuid-503'} ) label_participant_1 = make_graph_node( node_id=504, labels={'LabelParticipant'}, data={'uuid': 'uuid-504'} ) collaborator_1 = make_graph_node( node_id=505, labels={'Collaborator'}, data={'uuid': 'uuid-505'} ) session_mock = get_session_mock( [ {'resource': vendor_1}, {'resource': subaccount_1}, {'resource': subaccount_2}, {'resource': company_brand_1}, {'resource': label_participant_1}, {'resource': collaborator_1}, ] ) # Mock the return value of `single` which is used for the count response session_mock.__enter__.return_value.execute.return_value.single.return_value = [5] with patch.object(neo4j, 'db_session', return_value=session_mock): actual, actual_count = resource.get_resources_for_identity_uuid( identity_uuid, limit, offset ) # Two calls - one to get total records, one to get paginated list assert session_mock.__enter__().run.call_count == 2 # Assert args for getting total records assert session_mock.__enter__().run.call_args_list[0] == call( 'MATCH (i:Identity)-[hp:HAS_PROFILE]->(p:Profile)\n -[r:HAS_ADMIN_ACCESS_TO]->(x)\n WHERE\n p.profileType = "SettingsProfile" AND\n i.id = $identity_uuid AND\n (x:Subaccount OR x:SubAccount OR x:Vendor OR x:CompanyBrand OR x:LabelParticipant OR x:Collaborator)\n RETURN\n COUNT(x) AS total\n ', # noqa: E501 identity_uuid=identity_uuid, ) # Assert args for getting data assert session_mock.__enter__().run.call_args_list[1] == call( 'MATCH (i:Identity)-[hp:HAS_PROFILE]->(p:Profile)\n -[r:HAS_ADMIN_ACCESS_TO]->(x)\n WHERE\n p.profileType = "SettingsProfile" AND\n i.id = $identity_uuid AND\n (x:Subaccount OR x:SubAccount OR x:Vendor OR x:CompanyBrand OR x:LabelParticipant OR x:Collaborator)\n RETURN\n x AS resource\n ORDER BY resource.uuid\n SKIP $offset LIMIT $limit\n ', # noqa: E501 identity_uuid=identity_uuid, limit=limit, offset=offset, ) assert actual == [ {'vendorId': 16055, 'id': 500, 'uuid': 'uuid-500', 'type': 'Vendor'}, {'id': 501, 'type': 'SubAccount', 'uuid': 'uuid-501'}, {'id': 502, 'type': 'Subaccount', 'uuid': 'uuid-502'}, {'id': 503, 'uuid': 'uuid-503', 'type': 'CompanyBrand'}, {'id': 504, 'type': 'LabelParticipant', 'uuid': 'uuid-504'}, {'id': 505, 'type': 'Collaborator', 'uuid': 'uuid-505'}, ] assert actual_count == 5 @pytest.mark.parametrize( ('mock_identity', 'expected', 'db_count'), [ [ [{'id': 'user-uuid', 'resource_count': 2, 'auth0_user_id': 'auth|test123678test'}], {'id': 'user-uuid', 'resource_count': 2, 'auth0_user_id': 'auth|test123678test'}, 1, ], [ # no resources left, so set active=N for Identity [{'id': 'user-uuid', 'resource_count': 0, 'identity': {'active': 'N'}}], {'id': 'user-uuid', 'resource_count': 0, 'active': 'N', 'identity': {'active': 'N'}}, 2, ], ], ) def test_deactivate_user(mock_identity, expected, db_count): """Test deactivate_user.""" session_mock = get_transactional_session_mock(mock_identity) admin_context = { 'identity_id': 'admin-uuid', 'profile_type': 'InsightsProfile', 'profile_id': 12, } identity_id = 'user-uuid' profiles = [ {'id': 12, 'profileType': 'InsightsProfile'}, {'id': 22, 'profileType': 'LabelProfile'}, {'id': 32, 'profileType': 'SettingsProfile'}, ] # mock_identity = [{'id': identity_id, 'resource_count': 2, # 'auth0_user_id': 'auth|test123678test'}] with patch.object(neo4j, '_get_neo4j_session', return_value=session_mock), patch.object( resource, '_deactivate_resources_common_with_admin', return_value=profiles ), patch.object(resource, '_get_identity_resource_count', side_effect=mock_identity): actual = resource.deactivate_user( admin_context, identity_id, constants.SETTINGS_SUPPORT_MAPPING['profileTypes'] ) assert actual assert actual.message == expected assert resource._get_identity_resource_count.call_count == 1 assert resource._deactivate_resources_common_with_admin.call_count == 1 IDENTITY_FIELDS = { 'id': 'user-uuid', 'active': 'Y', 'auth0UserId': 'somehexvalue', 'firstName': 'Bruno', 'lastName': 'Dog', 'name': 'Bruno Dog', 'email': 'bruno@emailfor.dogs', 'userTypes': ['label'], 'defaultBrand': 'theorchard', } @pytest.mark.parametrize( ('auth0_result', 'expected'), [ [ # user is successfully activated in neo4j and auth0 response.Response({'some': 'auth0 metadata'}), response.Response( message={ **{stringcase.snakecase(k): v for k, v in IDENTITY_FIELDS.items()}, 'auth0_result': {'some': 'auth0 metadata'}, } ), ], [ # user is successfully activated in neo4j but auth0 update fails. response.create_error_response('some-error', 'some auth0 error'), response.create_error_response(error.CODE_FAILED_ACTIVATE_USER, 'some error'), ], ], ) @patch.object(owsusers, 'get_auth0_user_id_by_email') @patch.object(identity_model, 'update_identity_active_status') @patch('permissions.models.resource.g') def test_activate_user( g_mock, identity_model_mock, owsusers_mock, auth0_result, expected, app_context ): """Test activate_user.""" admin_id = 'admin-uuid' identity_id = 'user-uuid' mock_identity = Identity(**{stringcase.snakecase(k): v for k, v in IDENTITY_FIELDS.items()}) identity_model_mock.return_value = mock_identity owsusers_mock.return_value = IDENTITY_FIELDS['auth0UserId'] session_mock = get_transactional_session_mock([IDENTITY_FIELDS]) with patch.object(neo4j, '_get_neo4j_session', return_value=session_mock), patch.object( auth0, 'activate_deactivate_user', side_effect=[auth0_result] ): actual = resource.activate_user(admin_id=admin_id, identity_id=identity_id) assert actual.status == expected.status if expected: assert actual.message == expected.message owsusers_mock.assert_called_once_with(IDENTITY_FIELDS['email']) if auth0_result and not auth0_result.status: g_mock.log.error.assert_called_once_with( 'Failed to activate user in auth0', identity_id=identity_id, auth0_user_id=IDENTITY_FIELDS['auth0UserId'], errors=auth0_result.errors, ) @patch.object(owsusers, 'get_auth0_user_id_by_email') @patch.object(identity_model, 'update_identity_active_status') def test_activate_user_without_auth0_user_id(identity_model_mock, owsusers_mock): """Test activate_user when auth0 user does not exist.""" neo4j_result = deepcopy(IDENTITY_FIELDS) # user is successfully activated in neo4j but does not exist in auth0 neo4j_result['auth0UserId'] = None identity_id = 'user-uuid' mock_identity = Identity(**{stringcase.snakecase(k): v for k, v in neo4j_result.items()}) identity_model_mock.return_value = mock_identity owsusers_mock.return_value = None session_mock = get_transactional_session_mock([neo4j_result]) expected = response.Response( message={ **{stringcase.snakecase(k): v for k, v in IDENTITY_FIELDS.items()}, 'auth0_user_id': None, 'auth0_result': None, } ) with patch.object(neo4j, '_get_neo4j_session', return_value=session_mock), patch.object( auth0, 'activate_deactivate_user', side_effect=[None] ): actual = resource.activate_user(admin_id='whatever', identity_id=identity_id) assert actual.status == expected.status assert actual.message == expected.message owsusers_mock.assert_called_once_with(IDENTITY_FIELDS['email']) auth0.activate_deactivate_user.assert_not_called() @patch.object(owsusers, 'get_auth0_user_id_by_email') @patch.object(auth0, 'activate_deactivate_user') @patch.object(neo4j, '_get_neo4j_session') @patch.object(identity_model, 'update_identity_active_status') def test_activate_user_pending_identity(identity_model_mock, neo4j_mock, auth0_mock, owsusers_mock): """Test activate_user with a pending identity - should still check auth0 by email.""" neo4j_result = deepcopy(IDENTITY_FIELDS) # Make identity pending by setting id == auth0UserId neo4j_result['id'] = 'pending-uuid' neo4j_result['auth0UserId'] = 'pending-uuid' identity_id = 'pending-uuid' mock_identity = Identity(**{stringcase.snakecase(k): v for k, v in neo4j_result.items()}) identity_model_mock.return_value = mock_identity owsusers_mock.return_value = 'auth0|actualuserid' auth0_user_data = { 'user_id': 'auth0|actualuserid', 'email': neo4j_result['email'], 'blocked': False, } auth0_mock.return_value = response.Response(auth0_user_data) session_mock = get_transactional_session_mock([neo4j_result]) neo4j_mock.return_value = session_mock expected = response.Response( message={ **{stringcase.snakecase(k): v for k, v in neo4j_result.items()}, 'auth0_result': auth0_user_data, } ) actual = resource.activate_user(admin_id='admin-uuid', identity_id=identity_id) assert actual.status == expected.status assert actual.message == expected.message owsusers_mock.assert_called_once_with(neo4j_result['email']) auth0_mock.assert_called_once_with('auth0|actualuserid', True) @patch('permissions.models.resource.g') def test_create_profile_uuid_to_resource_relationship(_, app_context): """Test create_profile_uuid_to_resource_relationship.""" schema = { 'resource_type': 'ArtistInfo', 'resource_id': 100, 'profile_uuid': 'some-uuid', 'roles': ['test'], } expected = deepcopy(schema) session_mock = get_session_mock([{'rel': 'foo'}]) with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.create_profile_to_resource_relationship(schema, 'admin-id') assert actual assert actual.message == expected assert session_mock.__enter__().run.call_count == 2 session_mock.__enter__().run.assert_called_with(ANY, **expected) # check parts of the query. args = session_mock.__enter__().run.call_args assert args[0] == ( 'MATCH (p:Profile)\n MATCH(r:ArtistInfo)\n WHERE\n p.profileType = $profile_type AND\n p.profileId = $profile_id AND r.id = $resource_id\n MERGE (p)-[rel:HAS_ACCESS_TO]->(r)\n SET rel.roles = $roles, rel.createdAt = datetime()\n \n RETURN rel', ) # noqa def test_soft_delete_profile_uuid_to_resource_relationship(): """Test soft_delete_profile_uuid_to_resource_relationship.""" schema = { 'resource_type': 'ArtistInfo', 'resource_id': 100, 'profile_uuid': 'some-uuid', 'roles': ['test'], } expected = deepcopy(schema) session_mock = get_session_mock([{'input': 'foo', 'output': 'bar'}]) with patch.object(neo4j, 'db_session', return_value=session_mock): actual = resource.soft_delete_profile_uuid_to_resource_relationship(schema) assert actual assert session_mock.__enter__().run.call_count == 1 session_mock.__enter__().run.assert_called_with(ANY, **expected) # check parts of the query. args = session_mock.__enter__().run.call_args assert args[0] == ( "MATCH (p:Profile)-[rel:HAS_ACCESS_TO]->\n (r:ArtistInfo)\n WHERE p.uuid = $profile_uuid AND r.id = $resource_id\n SET rel.dateDeleted = localdatetime()\n WITH rel\n CALL apoc.refactor.setType(rel, 'DELETED_HAS_ACCESS_TO')\n YIELD input, output\n RETURN input, output", ) # noqa