"""Test logic for profiles.""" from copy import deepcopy from freezegun import freeze_time from owsresponse import response import pytest from pythonfeatures import pythonfeatures from pythonfeatures.constants import split as split_constants from users import constants from users.logic import profiles from users.models import identities as identities_model, profiles as profiles_model from users.utils import authorization @pytest.mark.parametrize( ('get_identity', 'profiles_by_type_result', 'exception', 'expected', 'call_args'), [ ( # identity node don't exist response.create_not_found_response('no identity'), None, True, 'no identity', None, ), ( # identity exist but no profile of this type. response.Response(), response.Response([]), False, {'created': 'profile'}, {'profile_id': 1346777, 'profile_type': 'InsightsProfile', 'roles': ['r_old']}, ), ( # identity and profile exists. response.Response(), response.Response([{'profile_id': 123, 'profile_type': 'abc'}]), False, {'updated': 'profile'}, (123, 'abc', {'roles': ['r_old']}), ), ( # identity and profile exists with roles. response.Response(), response.Response([{'profile_id': 123, 'profile_type': 'abc', 'roles': ['r1']}]), False, {'updated': 'profile'}, (123, 'abc', {'roles': ['r1', 'r_old']}), ), ], ) def test_create_profile_if_not_exist( get_identity, profiles_by_type_result, exception, expected, call_args, mocker ): """Test create_profile_if_not_exist.""" identity_id = '5b83678aad451875aad4aaef' profile = {'profile_id': 1346777, 'profile_type': 'InsightsProfile', 'roles': ['r_old']} mocker.patch.object(identities_model, 'get_identity', return_value=get_identity, autospec=True) mocker.patch.object( profiles_model, 'get_profiles_by_type', return_value=profiles_by_type_result, autospec=True ) mocker.patch.object( profiles_model, 'update_profile', return_value=response.Response({'updated': 'profile'}), autospec=True, ) mocker.patch.object( profiles_model, 'create_profile', return_value=response.Response({'created': 'profile'}), autospec=True, ) mocker.patch.object( profiles_model, 'link_identity_to_profile', return_value=response.Response({'linked': 'profile'}), autospec=True, ) try: actual = profiles.create_profile_if_not_exist(identity_id, profile) assert actual assert actual.message == expected identities_model.get_identity.assert_called_once() profiles_model.get_profiles_by_type.assert_called_once() if expected == {'updated': 'profile'}: profiles_model.update_profile.assert_called_once_with(identity_id, *call_args) else: profiles_model.create_profile.assert_called_once_with(call_args) profiles_model.link_identity_to_profile.assert_called_once() except Exception as err: assert exception assert err.args[0] == expected identities_model.get_identity.assert_called_once() if get_identity: profiles_model.get_profiles_by_type.assert_called_once() if expected == {'updated': 'profile'}: profiles_model.update_profile.assert_called_once() else: profiles_model.create_profile.assert_called_once() profiles_model.link_identity_to_profile.assert_called_once() def test_grant_access(mocker): """Test grant_access.""" auth0_id = 'auth0id' profile = {'profile_id': 1346777, 'profile_type': 'InsightsProfile', 'roles': ['analytics']} resource_type = constants.RESOURCE_VENDOR resource_id = 7123 mocker.patch.object( profiles_model, 'create_profile_to_resource_relationship', return_value=response.Response({'test': 'rel'}), autospec=True, ) mocker.patch.object( identities_model, 'get_identity', return_value=response.Response({'id': 'abcd1234'}), autospec=True, ) profiles.grant_access(auth0_id, profile, resource_type, resource_id) identities_model.get_identity.assert_called_once_with(auth0_id) profiles_model.create_profile_to_resource_relationship.assert_called_once() @pytest.mark.parametrize( ( 'constants_mock', 'db_result', ), [ ( ['ArtistProfile', 'InsightsProfile'], [{'profile_id': 1346777, 'profile_type': 'ArtistProfile'}], ), (['ArtistProfile', 'InsightsProfile'], []), (['LabelProfile'], [{'profile_id': 1346777, 'profile_type': 'ArtistProfile'}]), ( ['LabelProfile', 'ArtistProfile'], [ {'profile_id': 1346777, 'profile_type': 'ArtistProfile'}, {'profile_id': 23234234, 'profile_type': 'LabelProfile'}, ], ), ( ['LabelProfile', 'DistributionProfile'], [ {'profile_id': 1346777, 'profile_type': 'DistributionProfile'}, {'profile_id': 23234234, 'profile_type': 'LabelProfile'}, ], ), ], ) def test_get_profile_and_roles_from_graph(mocker, constants_mock, db_result): """Test get_profile_and_roles_from_graph.""" orchard_identity = '5b83678aad451875aad4aaef' app_name = 'orchard-go' expected = {} db_result_original = deepcopy(db_result) for profile in db_result_original: if profile.get('profile_type') in constants_mock: expected.update({profile.get('profile_id'): profile}) mocker.patch.dict( 'users.constants.APPLICATION_PROFILE_TYPE_MAPPING', {app_name: constants_mock} ) mocker.patch.object( profiles_model, 'get_profiles', return_value=response.Response(db_result), autospec=True ) actual = profiles.get_profile_and_roles_from_graph(orchard_identity, constants_mock) assert actual == expected def test_get_identity_by_email(mocker): """Test get_identity_by_email.""" email = 'test@theorchard.com' db_result = response.Response( {'name': 'New User', 'email': email, 'id': '5dd7d78c4ec49c0e2c4e4d1f'} ) mocker.patch.object( identities_model, 'get_identity_by_email', return_value=db_result, autospec=True ) actual = profiles.get_identity_by_email(email) assert actual.status == db_result.status assert actual.message == db_result.message def test_get_identity_for_admin_tx_with_super_admin_feature_flag(mocker): """Test get_identity_for_admin_tx when admin is in EDIT_SUPER_ADMINS feature flag.""" admin_context = { 'identity_id': 'admin-uuid', 'profile_type': 'SettingsProfile', 'profile_id': 12, } identity_id = 'target-user-uuid' mocker.patch.object( pythonfeatures, 'get_single_feature_by_attributes', return_value=response.Response(split_constants.FEATURE_ENABLED), ) mock_session = mocker.MagicMock() mock_identity = {'id': identity_id, 'name': 'Test User'} mock_session.execute_read.return_value = response.Response(mock_identity) mocker.patch('users.logic.profiles.get_session', return_value=mock_session) result = profiles.get_identity_for_admin_tx(admin_context, identity_id) assert result.message == mock_identity # Should call get_identity_tx because admin is super admin mock_session.execute_read.assert_called_once_with(identities_model.get_identity_tx, identity_id) def test_get_identity_for_admin_tx_with_pdp_authorization(mocker): """Test get_identity_for_admin_tx when admin is authorized via PDP.""" admin_context = { 'identity_id': 'admin-uuid', 'profile_type': 'SettingsProfile', 'profile_id': 12, } identity_id = 'target-user-uuid' mocker.patch.object( pythonfeatures, 'get_single_feature_by_attributes', return_value=response.Response(split_constants.FEATURE_DISABLED), ) mocker.patch.object( authorization, 'pdp_authorize_list_employees', return_value=True, ) mock_session = mocker.MagicMock() mock_identity = {'id': identity_id, 'name': 'Test User'} mock_session.execute_read.return_value = response.Response(mock_identity) mocker.patch('users.logic.profiles.get_session', return_value=mock_session) result = profiles.get_identity_for_admin_tx(admin_context, identity_id) assert result.message == mock_identity # Should call get_identity_tx because admin is authorized via PDP mock_session.execute_read.assert_called_once_with(identities_model.get_identity_tx, identity_id) def test_get_identity_for_admin_tx_without_authorization(mocker): """Test get_identity_for_admin_tx when admin can't access employees.""" admin_context = { 'identity_id': 'admin-uuid', 'profile_type': 'SettingsProfile', 'profile_id': 12, } identity_id = 'target-user-uuid' mocker.patch.object( pythonfeatures, 'get_single_feature_by_attributes', return_value=response.Response(split_constants.FEATURE_DISABLED), ) mocker.patch.object( authorization, 'pdp_authorize_list_employees', return_value=False, ) mock_session = mocker.MagicMock() mock_identity = {'id': identity_id, 'name': 'Test User'} mock_session.execute_read.return_value = response.Response(mock_identity) mocker.patch('users.logic.profiles.get_session', return_value=mock_session) result = profiles.get_identity_for_admin_tx(admin_context, identity_id) assert result.message == mock_identity # Should call get_identity_for_admin_tx because admin doesn't have super admin access mock_session.execute_read.assert_called_once_with( identities_model.get_identity_for_admin_tx, admin_context, identity_id=identity_id ) def test_get_identity_for_admin_by_email(mocker): """Test get_identity_for_admin_by_email.""" admin_context = { 'identity_id': 'admin-uuid', 'profile_type': 'SettingsProfile', 'profile_id': 12, } email = 'foo@bar.com' mocker.patch.object( identities_model, 'get_identity_for_admin', return_value=response.Response() ) actual = profiles.get_identity_for_admin_by_email(admin_context, email) assert actual identities_model.get_identity_for_admin.assert_called_once() def test_get_profiles_for_identity_invalid_app(mocker): """Test get_profiles_for_identity.""" db_result = [ {'profile_id': 1346777, 'profile_type': 'ArtistProfile'}, {'profile_id': 23234234, 'profile_type': 'LabelProfile'}, ] orchard_identity = '5b83678aad451875aad4aaef' app_name = 'test-app' mocker.patch.object( profiles_model, 'get_profiles', return_value=response.Response(db_result), autospec=True ) actual = profiles.get_profiles_for_identity(orchard_identity, app_name) assert not actual assert actual.errors == { 'code': 'invalid_app', 'message': 'Application test-app not supported.', } def test_get_profiles_for_identity_with_neo4j_no_db(mocker, monkeypatch): """Test get_profiles_for_identity. when art_relations has no label profiles, then result list contains profiles from graph db only. """ profiles_graph_result = {'1234': {'dummy': 'profile'}} orchard_identity = '5b83678aad451875aad4aaef' app_name = 'orchard-go' expected = list(profiles_graph_result.values()) mocker.patch.object( profiles, 'get_profile_and_roles_from_graph', return_value=profiles_graph_result, autospec=True, ) mocker.patch.object( identities_model, 'get_identity', return_value=response.create_not_found_response(), autospec=True, ) actual = profiles.get_profiles_for_identity(orchard_identity, app_name) assert actual assert actual.message == { 'items': expected, 'pagination': {'total_records': len(expected), 'type': 'none'}, } profiles.get_profile_and_roles_from_graph.assert_called_once() @pytest.mark.parametrize( ('app_name', 'expected_profile_types'), [ # 1. Workstation never gets InsightsProfile (constants.APPLICATION_WORKSTATION, [constants.LABEL_PROFILE]), (constants.APPLICATION_WORKSTATION, [constants.LABEL_PROFILE]), (constants.APPLICATION_WORKSTATION, [constants.LABEL_PROFILE]), # 2. settings app (constants.APPLICATION_SETTINGS, [constants.SETTINGS_PROFILE]), (constants.APPLICATION_SETTINGS, [constants.SETTINGS_PROFILE]), # 3. orchardgo (constants.APPLICATION_ORCHARDGO, [constants.INSIGHTS_PROFILE]), ], ) def test_get_profiles_for_identity(mocker, app_name, expected_profile_types): """Test get_profiles_for_identity.""" orchard_identity_id = '5b83678aad451875aad4aaef' mocker.patch.object( profiles, 'get_profile_and_roles_from_graph', return_value={'foo': 'bar'}, autospec=True ) result = profiles.get_profiles_for_identity(orchard_identity_id, app_name) assert result profiles.get_profile_and_roles_from_graph.assert_called_once_with( orchard_identity_id, expected_profile_types ) @pytest.mark.parametrize( ('app_name', 'profile_types'), [ (constants.APPLICATION_WORKSTATION, [constants.LABEL_PROFILE]), (constants.APPLICATION_ORCHARDGO, [constants.INSIGHTS_PROFILE]), (constants.APPLICATION_ORCHARDGO, [constants.INSIGHTS_PROFILE, constants.SETTINGS_PROFILE]), ], ) def test_get_profiles_for_identity_with_profile_types(mocker, app_name, profile_types): """Test get_profiles_for_identity passes profile types to get_profile_and_roles_from_graph.""" orchard_identity_id = '5b83678aad451875aad4aaef' mocker.patch.object( profiles, 'get_profile_and_roles_from_graph', return_value={'foo': 'bar'}, autospec=True ) result = profiles.get_profiles_for_identity(orchard_identity_id, app_name, profile_types) assert result profiles.get_profile_and_roles_from_graph.assert_called_once_with( orchard_identity_id, profile_types ) @pytest.mark.parametrize( ('app_name', 'expected_profile_types'), [ # 1. Workstation never gets InsightsProfile (constants.APPLICATION_WORKSTATION, [constants.LABEL_PROFILE]), (constants.APPLICATION_WORKSTATION, [constants.LABEL_PROFILE]), (constants.APPLICATION_WORKSTATION, [constants.LABEL_PROFILE]), # 2. settings app (constants.APPLICATION_SETTINGS, [constants.SETTINGS_PROFILE]), (constants.APPLICATION_SETTINGS, [constants.SETTINGS_PROFILE]), # 3. orchardgo (constants.APPLICATION_ORCHARDGO, [constants.INSIGHTS_PROFILE]), ], ) def test_get_profiles_for_identity_with_no_profile_types(mocker, app_name, expected_profile_types): """Test get_profiles_for_identity with no profile types.""" orchard_identity_id = '5b83678aad451875aad4aaef' profile_types = [] mocker.patch.object( profiles, 'get_profile_and_roles_from_graph', return_value={'foo': 'bar'}, autospec=True ) result = profiles.get_profiles_for_identity(orchard_identity_id, app_name, profile_types) assert result profiles.get_profile_and_roles_from_graph.assert_called_once_with( orchard_identity_id, expected_profile_types ) @pytest.mark.parametrize( ('neo4j_result', 'get_identity', 'expected'), [ ( # identity is not migrated {'4567': {'dummy': 'artistprofile'}}, response.create_not_found_response(), [ {'dummy': 'artistprofile'}, ], ), ( # identity is not migrated and no label profiles. {'4567': {'dummy': 'artistprofile'}}, response.create_not_found_response(), [ {'dummy': 'artistprofile'}, ], ), ( # identity is migrated but no label profile {'4567': {'dummy': 'artistprofile'}}, response.Response({'auth0_user_id': 'abc2343'}), [ {'dummy': 'artistprofile'}, ], ), ( # identity is migrated and has label profile {'4567': {'dummy': 'artistprofile'}}, response.Response({'auth0_user_id': 'abc2343'}), [ {'dummy': 'artistprofile'}, ], ), ], ) def test_get_profiles_for_identity_with_neo4j_and_db(neo4j_result, get_identity, expected, mocker): """Test get_profiles_for_identity. when art_relations has some label profiles, then result list contains profiles from graph db and label profiles. """ orchard_identity = '5b83678aad451875aad4aaef' app_name = 'orchard-go' mocker.patch.object( profiles, 'get_profile_and_roles_from_graph', return_value=neo4j_result, autospec=True ) mocker.patch.object(identities_model, 'get_identity', return_value=get_identity, autospec=True) mocker.patch.object( pythonfeatures, 'get_single_feature', return_value=response.Response(message='control'), autospec=True, ) actual = profiles.get_profiles_for_identity(orchard_identity, app_name) assert actual assert actual.message == { 'items': expected, 'pagination': {'total_records': len(expected), 'type': 'none'}, } profiles.get_profile_and_roles_from_graph.assert_called_once() @pytest.mark.parametrize( ( 'profile_data', 'get_identity_response', 'get_profile_response', 'create_profile_response', 'link_profile_response', 'expected_response_status', 'expected_response_message', ), [ # 201 created ( { 'profile_id': 555, 'profile_name': 'Test Profile', 'profile_type': 'ArtistProfile', 'roles': ['catalog', 'analytics'], }, response.Response(message={}), response.create_not_found_response(message=constants.ERROR_MESSAGE_PROFILE_NOT_FOUND), response.Response( status=201, message={ 'profile_id': 555, 'profile_name': 'Test Profile', 'profile_type': 'ArtistProfile', 'roles': ['catalog', 'analytics'], }, ), response.Response(), 201, { 'profile_id': 555, 'profile_name': 'Test Profile', 'profile_type': 'ArtistProfile', 'roles': ['catalog', 'analytics'], }, ), ( { 'profile_id': 556, 'profile_name': 'Test Profile', 'profile_type': 'ArtistProfile', 'roles': ['catalog', 'analytics'], }, response.Response(message={'default_brand': 'sme'}), response.create_not_found_response(message=constants.ERROR_MESSAGE_PROFILE_NOT_FOUND), response.Response( status=201, message={ 'brand': 'sme', 'profile_id': 556, 'profile_name': 'Test Profile', 'profile_type': 'ArtistProfile', 'roles': ['catalog', 'analytics'], }, ), response.Response(), 201, { 'brand': 'sme', 'profile_id': 556, 'profile_name': 'Test Profile', 'profile_type': 'ArtistProfile', 'roles': ['catalog', 'analytics'], }, ), # 201 created without profile id ( { 'profile_name': 'Test Profile', 'profile_type': 'ArtistProfile', 'roles': ['catalog', 'analytics'], }, response.Response({}), None, response.Response( status=201, message={ 'profile_id': 888888889, 'profile_name': 'Test Profile', 'profile_type': 'ArtistProfile', 'roles': ['catalog', 'analytics'], }, ), response.Response(), 201, { 'profile_id': 888888889, 'profile_name': 'Test Profile', 'profile_type': 'ArtistProfile', 'roles': ['catalog', 'analytics'], }, ), # 404 identitiy not found ( { 'profile_id': 555, 'profile_name': 'Test Profile', 'profile_type': 'ArtistProfile', 'roles': ['catalog', 'analytics'], }, response.create_not_found_response(message=constants.ERROR_MESSAGE_IDENTITY_NOT_FOUND), None, None, None, 404, constants.ERROR_MESSAGE_IDENTITY_NOT_FOUND, ), # 400 no profile id, but profile exists with next auto-increment ( { 'profile_name': 'Test Profile', 'profile_type': 'ArtistProfile', 'roles': ['catalog', 'analytics'], }, response.Response({}), None, response.create_error_response( code=constants.ERROR_CODE_VALIDATION_ERROR, message=constants.ERROR_MESSAGE_PROFILE_EXISTS, ), None, 400, constants.ERROR_MESSAGE_PROFILE_EXISTS, ), # 400 profile already exists ( { 'profile_id': 555, 'profile_name': 'Test Profile', 'profile_type': 'ArtistProfile', 'roles': ['catalog', 'analytics'], }, response.Response(), response.Response(), None, None, 400, 'Profile already exists', ), ], ) def test_create_profile_logic( mocker, profile_data, get_identity_response, get_profile_response, create_profile_response, link_profile_response, expected_response_status, expected_response_message, ): """Test test_create_profile_logic.""" identity_id = 'auth0_user_id' mock_g = mocker.MagicMock() mock_g.log = mocker.MagicMock() mocker.patch('users.logic.profiles.g', mock_g) mocker.patch.object( identities_model, 'get_identity', return_value=get_identity_response, autospec=True ) mocker.patch.object( profiles_model, 'get_by_profile_id_and_type', return_value=get_profile_response, autospec=True, ) mocker.patch.object( profiles_model, 'create_profile', return_value=create_profile_response, autospec=True ) mocker.patch.object( profiles_model, 'link_identity_to_profile', return_value=link_profile_response, autospec=True, ) result = profiles.create_profile(identity_id, profile_data) assert result.status == expected_response_status if result: assert result.message == expected_response_message else: assert result.errors.get('message') == expected_response_message if get_identity_response is not None: identities_model.get_identity.assert_called_with(identity_id) if get_profile_response is not None: profiles_model.get_by_profile_id_and_type.assert_called_with( profile_data.get('profile_id', 888888889), profile_data.get('profile_type') ) if create_profile_response is not None: profiles_model.create_profile.assert_called() assert profiles_model.create_profile.call_args[0][0] == profile_data if link_profile_response is not None: profiles_model.link_identity_to_profile.assert_called_with( identity_id, profile_data.get('profile_id', 888888889), profile_data.get('profile_type') ) # Verify logging for error cases if get_identity_response is not None and not get_identity_response: mock_g.log.warn.assert_called_once_with( 'Error creating profile; identity not found', resources={ 'identity_id': identity_id, 'profile_type': profile_data.get('profile_type'), }, ) elif create_profile_response is not None and not create_profile_response: mock_g.log.warn.assert_called_once_with( 'Error creating profile', resources={ 'identity_id': identity_id, 'profile_type': profile_data.get('profile_type'), }, ) else: mock_g.log.warn.assert_not_called() @pytest.mark.parametrize( ( 'payload', 'get_profile_response', 'get_identity_response', 'update_profile_response', 'expected_response_status', 'expected_response_message', ), [ # 200 ok, updated ( {'profile_name': 'Updated Profile Name', 'roles': ['pizza', 'hot dogs']}, response.Response( message={ 'profile_id': 555, 'profile_name': 'Test Profile', 'profile_type': 'ArtistProfile', 'roles': ['catalog', 'analytics'], } ), response.Response( message={'id': 'abc-123', 'name': 'test identity', 'email': 'test@theorchard.com'} ), response.Response( message={ 'profile_id': 555, 'profile_name': 'Updated Profile Name', 'profile_type': 'ArtistProfile', 'roles': ['pizza', 'hot dogs'], } ), 200, { 'profile_id': 555, 'profile_name': 'Updated Profile Name', 'profile_type': 'ArtistProfile', 'roles': ['pizza', 'hot dogs'], }, ), # 200 ok, just name ( {'profile_name': 'Updated Profile Name'}, response.Response( message={ 'profile_id': 555, 'profile_name': 'Test Profile', 'profile_type': 'ArtistProfile', 'roles': ['catalog', 'analytics'], } ), response.Response( message={'id': 'abc-123', 'name': 'test identity', 'email': 'test@theorchard.com'} ), response.Response( message={ 'profile_id': 555, 'profile_name': 'Updated Profile Name', 'profile_type': 'ArtistProfile', 'roles': ['catalog', 'analytics'], } ), 200, { 'profile_id': 555, 'profile_name': 'Updated Profile Name', 'profile_type': 'ArtistProfile', 'roles': ['catalog', 'analytics'], }, ), # 404, profile with id and type not found ( {'profile_name': 'Updated Profile Name', 'roles': ['pizza', 'hot dogs']}, response.create_not_found_response(message=constants.ERROR_MESSAGE_PROFILE_NOT_FOUND), None, None, 404, constants.ERROR_MESSAGE_PROFILE_NOT_FOUND, ), ], ) def test_update_profile_logic( mocker, payload, get_profile_response, get_identity_response, update_profile_response, expected_response_status, expected_response_message, ): """Test test_update_profile_logic.""" profile_id = 555 profile_type = 'ArtistProfile' mocker.patch.object( profiles_model, 'get_by_profile_id_and_type', return_value=get_profile_response, autospec=True, ) mocker.patch.object( identities_model, 'get_identity_for_profile', return_value=get_identity_response, autospec=True, ) mocker.patch.object( profiles_model, 'update_profile', return_value=update_profile_response, autospec=True ) result = profiles.update_profile(profile_id, profile_type, payload) assert result.status == expected_response_status if result: assert result.message == expected_response_message else: assert result.errors.get('message') == expected_response_message if get_profile_response is not None: profiles_model.get_by_profile_id_and_type.assert_called_with(profile_id, profile_type) if update_profile_response is not None: profile_data = { 'profile_name': get_profile_response.message.get('profile_name'), 'roles': get_profile_response.message.get('roles'), } profile_data.update(payload) profiles_model.update_profile.assert_called_with( get_identity_response.message.get('id'), profile_id, profile_type, profile_data ) @pytest.mark.parametrize( ( 'payload', 'get_profile_response', 'update_profile_response', 'expected_response_status', 'expected_response_message', ), [ # 200 ok, updated ( {'profile_name': 'Updated Profile Name', 'roles': ['pizza', 'hot dogs']}, response.Response( message={ 'profile_id': 555, 'profile_name': 'Test Profile', 'profile_type': 'ArtistProfile', 'roles': ['catalog', 'analytics'], } ), response.Response( message={ 'profile_id': 555, 'profile_name': 'Updated Profile Name', 'profile_type': 'ArtistProfile', 'roles': ['pizza', 'hot dogs'], } ), 200, { 'profile_id': 555, 'profile_name': 'Updated Profile Name', 'profile_type': 'ArtistProfile', 'roles': ['pizza', 'hot dogs'], }, ), # 200 ok, just name ( {'profile_name': 'Updated Profile Name'}, response.Response( message={ 'profile_id': 555, 'profile_name': 'Test Profile', 'profile_type': 'ArtistProfile', 'roles': ['catalog', 'analytics'], } ), response.Response( message={ 'profile_id': 555, 'profile_name': 'Updated Profile Name', 'profile_type': 'ArtistProfile', 'roles': ['catalog', 'analytics'], } ), 200, { 'profile_id': 555, 'profile_name': 'Updated Profile Name', 'profile_type': 'ArtistProfile', 'roles': ['catalog', 'analytics'], }, ), # 404, profile with id and type not found ( {'profile_name': 'Updated Profile Name', 'roles': ['pizza', 'hot dogs']}, response.create_not_found_response(message=constants.ERROR_MESSAGE_PROFILE_NOT_FOUND), None, 404, constants.ERROR_MESSAGE_PROFILE_NOT_FOUND, ), ], ) def test_update_profile_by_uuid_logic( mocker, payload, get_profile_response, update_profile_response, expected_response_status, expected_response_message, ): """Test update_profile_by_uuid.""" profile_uuid = 'some-uuid' mocker.patch.object(profiles_model, 'get_by_uuid', return_value=get_profile_response) mocker.patch.object( profiles_model, 'update_profile_by_uuid', return_value=update_profile_response ) result = profiles.update_profile_by_uuid(profile_uuid, payload) assert result.status == expected_response_status if result: assert result.message == expected_response_message else: assert result.errors.get('message') == expected_response_message if get_profile_response is not None: profiles_model.get_by_uuid.assert_called_with(profile_uuid) if update_profile_response is not None: profile_data = { 'profile_name': get_profile_response.message.get('profile_name'), 'roles': get_profile_response.message.get('roles'), } profile_data.update(payload) profiles_model.update_profile_by_uuid.assert_called_with(profile_uuid, profile_data) @pytest.mark.parametrize( ( 'get_profile_response', 'get_profiles_response', 'link_profile_response', 'expected_response_status', 'expected_response_message', ), [ # 200 success ( response.Response( message={ 'profile_id': 555, 'profile_name': 'Updated Profile Name', 'profile_type': 'ArtistProfile', 'roles': ['catalog', 'analytics'], } ), response.Response(message=[]), response.Response(), 200, None, ), # 404 profile does not exist ( response.create_not_found_response(message=constants.ERROR_MESSAGE_PROFILE_NOT_FOUND), None, None, 404, constants.ERROR_MESSAGE_PROFILE_NOT_FOUND, ), # 400 relationship already identity_exists ( response.Response( message={ 'profile_id': 555, 'profile_name': 'Updated Profile Name', 'profile_type': 'ArtistProfile', 'roles': ['catalog', 'analytics'], } ), response.Response(message=[{'profile_id': 1234, 'profile_type': 'ArtistProfile'}]), None, 400, constants.ERROR_MESSAGE_RELATIONSHIP_ALREADY_EXISTS, ), ], ) def test_add_profile_to_identity_logic( mocker, get_profile_response, get_profiles_response, link_profile_response, expected_response_status, expected_response_message, ): """Test test_add_profile_to_identity_logic.""" identity_id = 'auth0_user_id' profile_id = 1234 profile_type = 'ArtistProfile' mocker.patch.object( profiles_model, 'get_by_profile_id_and_type', return_value=get_profile_response, autospec=True, ) mocker.patch.object( profiles_model, 'get_profiles', return_value=get_profiles_response, autospec=True ) mocker.patch.object( profiles_model, 'link_identity_to_profile', return_value=link_profile_response, autospec=True, ) result = profiles.add_profile_to_identity(identity_id, profile_id, profile_type) assert result.status == expected_response_status if result.errors: assert result.errors.get('message') == expected_response_message else: assert result.message == expected_response_message if get_profile_response is not None: profiles_model.get_by_profile_id_and_type.assert_called_with(profile_id, profile_type) if get_profiles_response is not None: profiles_model.get_profiles.assert_called_with(identity_id) if link_profile_response is not None: profiles_model.link_identity_to_profile.assert_called_with( identity_id, profile_id, profile_type ) @pytest.mark.parametrize( ('delete_profile_response', 'expected_response_status', 'expected_response_message'), [ # 204 success (response.Response(), 200, None), # 500 delete failed ( response.create_fatal_response(constants.ERROR_MESSAGE_DELETE_RELATIONSHIP_FAILED), 500, constants.ERROR_MESSAGE_DELETE_RELATIONSHIP_FAILED, ), ], ) def test_delete_profile_to_identity( mocker, delete_profile_response, expected_response_status, expected_response_message ): """Test test_delete_profile_to_identity.""" identity_id = 'auth0_user_id' profile_id = 1234 profile_type = 'LabelProfile' mocker.patch.object( profiles_model, 'soft_delete_profile_to_identity_relationship', return_value=delete_profile_response, autospec=True, ) result = profiles.delete_profile_to_identity(identity_id, profile_id, profile_type) assert result.status == expected_response_status if result.errors: assert result.errors.get('message') == expected_response_message else: assert result.message == expected_response_message @pytest.mark.parametrize( ( 'get_profile_response', 'get_profiles_response', 'link_profile_response', 'expected_response_status', 'expected_response_message', ), [ # 200 success ( response.Response( message={ 'profile_id': 555, 'profile_name': 'Updated Profile Name', 'profile_type': 'ArtistProfile', 'roles': ['catalog', 'analytics'], } ), response.Response(message=[]), response.Response(), 200, None, ), # 404 profile does not exist ( response.create_not_found_response(message=constants.ERROR_MESSAGE_PROFILE_NOT_FOUND), None, None, 404, constants.ERROR_MESSAGE_PROFILE_NOT_FOUND, ), # 400 relationship already identity_exists ( response.Response( message={ 'profile_id': 555, 'profile_name': 'Updated Profile Name', 'profile_type': 'ArtistProfile', 'roles': ['catalog', 'analytics'], } ), response.Response( message=[{'profile_id': 1234, 'profile_type': 'ArtistProfile', 'uuid': 'some-uuid'}] ), None, 400, constants.ERROR_MESSAGE_RELATIONSHIP_ALREADY_EXISTS, ), ], ) def test_add_profile_uuid_to_identity( mocker, get_profile_response, get_profiles_response, link_profile_response, expected_response_status, expected_response_message, ): """Test add_profile_uuid_to_identity.""" profile_uuid = 'some-uuid' identity_id = 'identity-uuid' mocker.patch.object( profiles_model, 'get_by_uuid', return_value=get_profile_response, autospec=True ) mocker.patch.object( profiles_model, 'get_profiles', return_value=get_profiles_response, autospec=True ) mocker.patch.object( profiles_model, 'link_identity_to_profile_by_uuid', return_value=link_profile_response, autospec=True, ) result = profiles.add_profile_uuid_to_identity(identity_id, profile_uuid) assert result.status == expected_response_status if result.errors: assert result.errors.get('message') == expected_response_message else: assert result.message == expected_response_message if get_profile_response is not None: profiles_model.get_by_uuid.assert_called_with(profile_uuid) if get_profiles_response is not None: profiles_model.get_profiles.assert_called_with(identity_id) if link_profile_response is not None: profiles_model.link_identity_to_profile_by_uuid.assert_called_with( identity_id, profile_uuid ) @pytest.mark.parametrize( ('delete_profile_response', 'expected_response_status', 'expected_response_message'), [ # 204 success (response.Response(), 200, None), # 500 delete failed ( response.create_fatal_response(constants.ERROR_MESSAGE_DELETE_RELATIONSHIP_FAILED), 500, constants.ERROR_MESSAGE_DELETE_RELATIONSHIP_FAILED, ), ], ) def test_delete_profile_uuid_to_identity( mocker, delete_profile_response, expected_response_status, expected_response_message ): """Test delete_profile_uuid_to_identity.""" profile_uuid = 'some-uuid' identity_id = 'identity-uuid' mocker.patch.object( profiles_model, 'soft_delete_profile_to_identity_relationship_by_uuid', return_value=delete_profile_response, autospec=True, ) result = profiles.delete_profile_uuid_to_identity(identity_id, profile_uuid) assert result.status == expected_response_status if result.errors: assert result.errors.get('message') == expected_response_message else: assert result.message == expected_response_message @pytest.mark.parametrize( ( 'get_profile_response', 'delete_profile_response', 'expected_response_status', 'expected_response_message', ), [ # 204 ok, deleted, no content (response.Response(), response.Response(status=204), 204, None), # 404 profile_not_found ( response.create_not_found_response(message=constants.ERROR_MESSAGE_PROFILE_NOT_FOUND), None, 404, constants.ERROR_MESSAGE_PROFILE_NOT_FOUND, ), ], ) def test_delete_profile_logic( mocker, get_profile_response, delete_profile_response, expected_response_status, expected_response_message, ): """Test test_delete_profile_logic.""" profile_id = 88888 profile_type = 'whatever' mocker.patch.object( profiles_model, 'get_by_profile_id_and_type', return_value=get_profile_response, autospec=True, ) mocker.patch.object( profiles_model, 'delete_profile', return_value=delete_profile_response, autospec=True ) result = profiles.delete_profile(profile_id, profile_type) assert result.status == expected_response_status if result: assert result.message == expected_response_message else: assert result.errors.get('message') == expected_response_message if get_profile_response is not None: profiles_model.get_by_profile_id_and_type.assert_called_with(profile_id, profile_type) if delete_profile_response is not None: profiles_model.delete_profile.assert_called_with(profile_id, profile_type) @pytest.mark.parametrize( ('get_profile_response', 'delete_response', 'expected_status', 'expected_message'), [ # 204 ok, deleted, no content (response.Response(), response.Response(status=204), 204, None), # 404 profile_not_found ( response.create_not_found_response(message=constants.ERROR_MESSAGE_PROFILE_NOT_FOUND), None, 404, constants.ERROR_MESSAGE_PROFILE_NOT_FOUND, ), ], ) def test_delete_profile_by_uuid( mocker, get_profile_response, delete_response, expected_status, expected_message ): """Test delete_profile_by_uuid.""" mocker.patch.object(profiles_model, 'get_by_uuid', return_value=get_profile_response) mocker.patch.object(profiles_model, 'delete_profile_by_uuid', return_value=delete_response) result = profiles.delete_profile_by_uuid('some-uuid') assert result.status == expected_status if result: assert result.message == expected_message else: assert result.errors.get('message') == expected_message if get_profile_response is not None: profiles_model.get_by_uuid.assert_called_with('some-uuid') if delete_response is not None: profiles_model.delete_profile_by_uuid.assert_called_with('some-uuid') @pytest.mark.parametrize( ('get_profile_response', 'expected_response_status', 'expected_response_message'), [ # 200 ok, found ( response.Response( message={ 'profile_id': 555, 'profile_name': 'Updated Profile Name', 'profile_type': 'ArtistProfile', 'roles': ['catalog', 'analytics'], } ), 200, { 'profile_id': 555, 'profile_name': 'Updated Profile Name', 'profile_type': 'ArtistProfile', 'roles': ['catalog', 'analytics'], }, ), # 404 profile_not_found ( response.create_not_found_response(message=constants.ERROR_MESSAGE_PROFILE_NOT_FOUND), 404, constants.ERROR_MESSAGE_PROFILE_NOT_FOUND, ), ], ) def test_get_profile_logic( mocker, get_profile_response, expected_response_status, expected_response_message ): """Test test_get_profile_logic.""" profile_id = 88888 profile_type = 'whatever' mocker.patch.object( profiles_model, 'get_by_profile_id_and_type', return_value=get_profile_response, autospec=True, ) result = profiles.get_profile(profile_id, profile_type) assert result.status == expected_response_status if result: assert result.message == expected_response_message else: assert result.errors.get('message') == expected_response_message profiles_model.get_by_profile_id_and_type.assert_called_with(profile_id, profile_type) @pytest.mark.parametrize( ('get_profile_by_uuid_response', 'expected_response_status', 'expected_response_message'), [ # 200 ok, found ( response.Response( message={ 'profile_id': 555, 'profile_name': 'Updated Profile Name', 'profile_type': 'ArtistProfile', 'roles': ['catalog', 'analytics'], 'uuid': '2389e775-0ae3-41cc-90f1-2f893f780c9r', } ), 200, { 'profile_id': 555, 'profile_name': 'Updated Profile Name', 'profile_type': 'ArtistProfile', 'roles': ['catalog', 'analytics'], 'uuid': '2389e775-0ae3-41cc-90f1-2f893f780c9r', }, ), # 404 profile_not_found ( response.create_not_found_response(message=constants.ERROR_MESSAGE_PROFILE_NOT_FOUND), 404, constants.ERROR_MESSAGE_PROFILE_NOT_FOUND, ), ], ) def test_get_profile_by_uuid_logic( mocker, get_profile_by_uuid_response, expected_response_status, expected_response_message ): """Test test_get_profile_by_uuid_logic.""" uuid = '2389e775-0ae3-41cc-90f1-2f893f780c9r' mocker.patch.object( profiles_model, 'get_by_uuid', return_value=get_profile_by_uuid_response, autospec=True ) result = profiles.get_profile_by_uuid(uuid) assert result.status == expected_response_status if result: assert result.message == expected_response_message else: assert result.errors.get('message') == expected_response_message profiles_model.get_by_uuid.assert_called_with(uuid) @freeze_time('2020-05-05 01:01:01') @pytest.mark.parametrize( ( 'identity_data', 'create_identity_response', 'get_identity_response', 'get_email_response', 'create_settings_profile_response', 'expected_response_status', 'expected_response_message', 'feature_on', ), [ # 200 ok ( { 'email': 'test@theorchard.com', 'name': 'test@theorchard.com', 'identity_id': '123456789123456789123456', }, response.Response( message={ 'email': 'test@theorchard.com', 'name': 'test@theorchard.com', 'identity_id': '123456789123456789123456', } ), response.create_not_found_response(message=constants.ERROR_MESSAGE_IDENTITY_NOT_FOUND), response.create_not_found_response(), response.Response( message={ 'profile_name': 'test@theorchard.com', 'profile_type': 'Settings Profile', 'profile_id': 2, } ), 200, { 'email': 'test@theorchard.com', 'name': 'test@theorchard.com', 'identity_id': '123456789123456789123456', }, response.Response(message='enabled'), ), # 200 ok with audit data ( { 'email': 'test@theorchard.com', 'name': 'test@theorchard.com', 'identity_id': '123456789123456789123456', 'audit_user': 'test_audit_user_identity_id', }, response.Response( message={ 'email': 'test@theorchard.com', 'name': 'test@theorchard.com', 'identity_id': '123456789123456789123456', 'updated_by': 'test_audit_user_identity_id', 'updated_on': '2020-05-05 01:01:01', } ), response.create_not_found_response(message=constants.ERROR_MESSAGE_IDENTITY_NOT_FOUND), response.create_not_found_response(), response.Response( message={ 'profile_name': 'test@theorchard.com', 'profile_type': 'SettingsProfile', 'profile_id': 2, } ), 200, { 'email': 'test@theorchard.com', 'name': 'test@theorchard.com', 'identity_id': '123456789123456789123456', 'updated_by': 'test_audit_user_identity_id', 'updated_on': '2020-05-05 01:01:01', }, response.Response(message='enabled'), ), # 400 identity already exists ( { 'email': 'test@theorchard.com', 'name': 'test@theorchard.com', 'identity_id': '123456789123456789123456', }, None, response.Response( message={ 'email': 'test@theorchard.com', 'name': 'test@theorchard.com', 'identity_id': '123456789123456789123456', } ), response.create_not_found_response(), None, 400, 'Identity already exists.', None, ), ( { 'email': 'test@theorchard.com', 'name': 'test@theorchard.com', 'identity_id': '123456789123456789123456', }, None, response.create_not_found_response(), response.Response('foo'), None, 400, 'Identity already exists with this email.', None, ), ], ) def test_create_identity_in_graph( mocker, context, identity_data, create_identity_response, get_identity_response, get_email_response, create_settings_profile_response, expected_response_status, expected_response_message, feature_on, ): """Test test_create_identity_in_graph.""" mocker.patch.object( identities_model, 'create_identity', return_value=create_identity_response, autospec=True ) mocker.patch.object( identities_model, 'get_identity', return_value=get_identity_response, autospec=True ) mocker.patch.object( identities_model, 'get_identity_by_email', return_value=get_email_response, autospec=True ) mocker.patch.object( profiles_model, 'create_settings_profile', return_value=create_settings_profile_response, autospec=True, ) with context: result = profiles.create_identity_in_graph(identity_data) assert result.status == expected_response_status if result: assert result.message == expected_response_message else: assert result.errors.get('message') == expected_response_message if create_identity_response is not None: identities_model.create_identity.assert_called() assert identities_model.create_identity.call_args[0][0] == identity_data if get_identity_response is not None: identities_model.get_identity.assert_called_with(identity_data.get('identity_id')) if create_settings_profile_response: profiles_model.create_settings_profile.assert_called_with( identity_data.get('identity_id'), identity_data.get('name'), None ) @pytest.mark.parametrize( ( 'identity_data', 'get_identity_response', 'update_identity_response', 'expected_response_status', 'expected_response_message', ), [ # 200 ok ( {'email': 'newtest@theorchard.com'}, response.Response( { 'id': '123456789123456789123456', 'email': 'test@theorchard.com', 'name': 'test@theorchard.com', } ), response.Response( { 'identity_id': '123456789123456789123456', 'email': 'newtest@theorchard.com', 'name': 'test@theorchard.com', } ), 200, { 'identity_id': '123456789123456789123456', 'email': 'newtest@theorchard.com', 'name': 'test@theorchard.com', }, ), # 200 ok with localization ( { 'email': 'newtest@theorchard.com', 'name': 'test@theorchard.com', 'localization': 'en', }, response.Response( { 'id': '123456789123456789123456', 'email': 'test@theorchard.com', 'name': 'test@theorchard.com', 'localization': 'en', } ), response.Response( { 'identity_id': '123456789123456789123456', 'email': 'newtest@theorchard.com', 'name': 'test@theorchard.com', 'localization': 'en', } ), 200, { 'identity_id': '123456789123456789123456', 'email': 'newtest@theorchard.com', 'name': 'test@theorchard.com', 'localization': 'en', }, ), # 200 ok with date_format ( { 'email': 'newtest@theorchard.com', 'name': 'test@theorchard.com', 'date_format': 'DD/MM/YYYY', }, response.Response( { 'id': '123456789123456789123456', 'email': 'test@theorchard.com', 'name': 'test@theorchard.com', 'date_format': 'DD/MM/YYYY', } ), response.Response( { 'identity_id': '123456789123456789123456', 'email': 'newtest@theorchard.com', 'name': 'test@theorchard.com', 'date_format': 'DD/MM/YYYY', } ), 200, { 'identity_id': '123456789123456789123456', 'email': 'newtest@theorchard.com', 'name': 'test@theorchard.com', 'date_format': 'DD/MM/YYYY', }, ), # 400 identity not found ( {'email': 'newtest@theorchard.com'}, response.create_error_response( code=constants.ERROR_CODE_IDENTITY_NOT_FOUND, message=constants.ERROR_MESSAGE_IDENTITY_NOT_FOUND, ), None, 400, 'Identity not found', ), ], ) def test_update_identity_in_graph( mocker, identity_data, get_identity_response, update_identity_response, expected_response_status, expected_response_message, ): """Test update_identity_in_graph from profiles.""" identity_id = '123456789123456789123456' mocker.patch.object( identities_model, 'get_identity', return_value=get_identity_response, autospec=True ) mocker.patch.object( identities_model, 'update_identity', return_value=update_identity_response, autospec=True ) result = profiles.update_identity_in_graph(identity_id, identity_data) assert result.status == expected_response_status if result: assert result.message == expected_response_message else: assert result.errors.get('message') == expected_response_message if get_identity_response is not None: identities_model.get_identity.assert_called_with(identity_id) if update_identity_response: identities_model.update_identity.assert_called_with(identity_id, identity_data) @pytest.mark.parametrize( ( 'get_identity_response', 'delete_identity_response', 'expected_response_status', 'expected_response_message', ), [ # 204 ok ( response.Response( { 'identity_id': '123', 'email': 'test@theorchard.com', 'name': 'test@theorchard.com', } ), response.Response(status=204), 204, None, ), # 400 identity doesn't exist ( response.create_not_found_response(message=constants.ERROR_MESSAGE_IDENTITY_NOT_FOUND), None, 400, 'Identity not found', ), ], ) def test_delete_identity_from_graph( mocker, get_identity_response, delete_identity_response, expected_response_status, expected_response_message, ): """Test test_delete_identity_from_graph.""" identity_id = '123' mocker.patch.object( identities_model, 'get_identity', return_value=get_identity_response, autospec=True ) mocker.patch.object( identities_model, 'delete_identity', return_value=delete_identity_response, autospec=True ) result = profiles.delete_identity_from_graph(identity_id) assert result.status == expected_response_status if result: assert result.message == expected_response_message else: assert result.errors.get('message') == expected_response_message if get_identity_response is not None: identities_model.get_identity.assert_called_with(identity_id) if delete_identity_response is not None: identities_model.delete_identity.assert_called_with(identity_id) @pytest.mark.parametrize( ( 'get_identity_response', 'expected_response_status', 'expected_response_message', ), [ # 200 ok ( response.Response( message={ 'email': 'test@theorchard.com', 'name': 'test@theorchard.com', 'identity_id': '123', } ), 200, {'email': 'test@theorchard.com', 'name': 'test@theorchard.com', 'identity_id': '123'}, ), # 404 identity not found ( response.create_not_found_response( message={constants.ERROR_MESSAGE_IDENTITY_NOT_FOUND} ), 404, {constants.ERROR_MESSAGE_IDENTITY_NOT_FOUND}, ), ], ) def test_get_identity_from_graph_tx( mocker, get_identity_response, expected_response_status, expected_response_message, ): """Test get_identity_from_graph_tx.""" identity_id = 'auth0_user_id' mock_session = mocker.MagicMock() mock_session.execute_read.return_value = get_identity_response mocker.patch('users.logic.profiles.get_session', return_value=mock_session) result = profiles.get_identity_from_graph_tx(identity_id) assert result.status == expected_response_status if result: assert result.message == expected_response_message else: assert result.errors.get('message') == expected_response_message mock_session.execute_read.assert_called_once_with(identities_model.get_identity_tx, identity_id) @pytest.mark.parametrize( ( 'get_identity_response', 'update_identity_response', 'expected_response', ), [ # 200 ok ( response.Response( {'id': '123456789123456789123456', 'email': 'test@theorchard.com', 'name': 'foo'} ), response.Response( { 'identity_id': '123456789123456789123456', 'email': 'newtest@theorchard.com', 'name': 'foo', } ), response.Response( { 'identity_id': '123456789123456789123456', 'email': 'newtest@theorchard.com', 'name': 'foo', } ), ), # 400 identity not found ( response.create_error_response( code=constants.ERROR_CODE_IDENTITY_NOT_FOUND, message=constants.ERROR_MESSAGE_IDENTITY_NOT_FOUND, ), None, response.create_error_response( code=constants.ERROR_CODE_IDENTITY_NOT_FOUND, message=constants.ERROR_MESSAGE_IDENTITY_NOT_FOUND, ), ), ], ) def test_update_identity_by_email_flow( mocker, get_identity_response, update_identity_response, expected_response ): """Test update_identity_by_email.""" email = 'test@theorchard.com' identity_data = { 'identity_id': '123456789123456789123456', 'email': 'test@theorchard.com', 'name': 'foo', } model_param = {'identity_id': '123456789123456789123456', 'email': email, 'name': 'foo'} mocker.patch.object( identities_model, 'get_identity_by_email', return_value=get_identity_response, autospec=True ) mocker.patch.object( identities_model, 'update_identity', return_value=update_identity_response, autospec=True ) result = profiles.update_identity_by_email(email, identity_data) assert result.status == expected_response.status if result: assert result.message == expected_response.message else: assert result.errors.get('message') == expected_response.errors.get('message') if get_identity_response: identities_model.get_identity_by_email.assert_called_once_with(email) if update_identity_response: identities_model.update_identity.assert_called_once_with( get_identity_response.message['id'], model_param ) def test_update_identity_by_email_overwrite(mocker): """Test update_identity_by_email.""" email = 'test@theorchard.com' get_identity_response = response.Response( {'id': '123456789123456789123456', 'email': 'test@theorchard.com', 'name': 'foo'} ) identity_data = { 'name': 'new name', 'email': 'test@theorchard.com', } model_param = {'identity_id': '123456789123456789123456', 'email': email, 'name': 'new name'} mocker.patch.object( identities_model, 'get_identity_by_email', return_value=get_identity_response, autospec=True ) mocker.patch.object(identities_model, 'update_identity', autospec=True) profiles.update_identity_by_email(email, identity_data) identities_model.get_identity_by_email.assert_called_once_with(email) identities_model.update_identity.assert_called_once_with( get_identity_response.message['id'], model_param ) @pytest.mark.parametrize( ('identity_response', 'identity_data', 'data_to_model', 'expected_status'), [ # identity not found ( response.create_not_found_response(), {'name': 'new name', 'auth0_user_id': 'foo'}, None, 400, ), # identity was found ( response.Response(message={'email': 'foo@theorchard.com'}), {'name': 'new name', 'auth0_user_id': 'foo'}, {'name': 'new name', 'auth0_user_id': 'foo', 'email': 'foo@theorchard.com'}, 200, ), ( response.Response(message={'email': 'foo@theorchard.com'}), {'identity_id': 'newId', 'auth0_user_id': 'foo'}, {'id': 'newId', 'auth0_user_id': 'foo', 'email': 'foo@theorchard.com'}, 200, ), # cannot update email. ( response.Response(message={'email': 'foo@theorchard.com'}), {'identity_id': 'newId', 'auth0_user_id': 'foo', 'email': 'new@theorchard.com'}, {'id': 'newId', 'auth0_user_id': 'foo', 'email': 'foo@theorchard.com'}, 200, ), ], ) def test_update_identity_and_id_by_email( identity_response, identity_data, data_to_model, expected_status, mocker ): """Test update_identity_and_id_by_email.""" email = 'foo@theorchard.com' mocker.patch.object( identities_model, 'get_identity_by_email', return_value=identity_response, autospec=True ) mocker.patch.object( identities_model, 'update_identity_by_email', return_value=response.Response(), autospec=True, ) result = profiles.update_identity_and_id_by_email(email, identity_data) assert result.status == expected_status identities_model.get_identity_by_email.assert_called_once_with(email) if expected_status == 200: identities_model.update_identity_by_email.assert_called_once_with(email, data_to_model) @pytest.mark.parametrize( ('get_identity_response', 'expected_response_status', 'expected_response_message'), [ # 200 ok ( response.Response( message={ 'email': 'test@theorchard.com', 'name': 'test@theorchard.com', 'identity_id': '123', } ), 200, {'email': 'test@theorchard.com', 'name': 'test@theorchard.com', 'identity_id': '123'}, ), # 404 identity not found ( response.create_not_found_response( message={constants.ERROR_MESSAGE_IDENTITY_NOT_FOUND} ), 404, {constants.ERROR_MESSAGE_IDENTITY_NOT_FOUND}, ), ], ) def test_get_identity_for_profile_from_graph( mocker, get_identity_response, expected_response_status, expected_response_message ): """Test test_get_identity_for_profile_from_graph.""" profile_id = 12345 profile_type = 'LabelProfile' mocker.patch.object( identities_model, 'get_identity_for_profile', return_value=get_identity_response, autospec=True, ) result = profiles.get_identity_for_profile(profile_id, profile_type) assert result.status == expected_response_status if result: assert result.message == expected_response_message else: assert result.errors.get('message') == expected_response_message if get_identity_response is not None: identities_model.get_identity_for_profile.assert_called_with(profile_id, profile_type) @pytest.mark.parametrize( ('get_identity_response', 'expected_response_status', 'expected_response_message'), [ # 200 ok ( response.Response( message={ 'email': 'test@theorchard.com', 'name': 'test@theorchard.com', 'identity_id': 'uuid', 'auth0_user_id': '123', } ), 200, { 'email': 'test@theorchard.com', 'name': 'test@theorchard.com', 'identity_id': 'uuid', 'auth0_user_id': '123', }, ), # 404 identity not found ( response.create_not_found_response( message={constants.ERROR_MESSAGE_IDENTITY_NOT_FOUND} ), 404, {constants.ERROR_MESSAGE_IDENTITY_NOT_FOUND}, ), ], ) def test_get_identity_by_auth0_id( mocker, get_identity_response, expected_response_status, expected_response_message ): """Test get_identity_by_auth0_id.""" auth0_id = 'foo2334' mocker.patch.object( identities_model, 'get_identity_by_auth0_id', return_value=get_identity_response, autospec=True, ) result = profiles.get_identity_by_auth0_id(auth0_id) assert result.status == expected_response_status if result: assert result.message == expected_response_message else: assert result.errors.get('message') == expected_response_message identities_model.get_identity_by_auth0_id.assert_called_with(auth0_id) @pytest.mark.parametrize( 'db_result', [ ([{'profile_id': 1346777, 'profile_type': 'ArtistProfile'}]), ( [ {'profile_id': 1346777, 'profile_type': 'ArtistProfile'}, {'profile_id': 23234234, 'profile_type': 'LabelProfile'}, ] ), ], ) def test_get_all_profiles(mocker, db_result): """Test get all profiles for identity.""" orchard_identity = '5b83678aad451875aad4aaef' expected = {'items': db_result, 'pagination': {'type': 'none', 'total_records': len(db_result)}} mocker.patch.object( profiles_model, 'get_profiles', return_value=response.Response(db_result), autospec=True ) actual = profiles.get_all_profiles_for_identity(orchard_identity) assert actual.message == expected @pytest.mark.parametrize( ('linked_profiles', 'requested_id', 'expected'), [ # identity has access to the requested LabelProfile (ids as ints) ( [ {'profile_id': 123, 'profile_type': 'LabelProfile'}, {'profile_id': 456, 'profile_type': 'LabelProfile'}, ], '123', True, ), # identity has no access to the requested LabelProfile ( [{'profile_id': 123, 'profile_type': 'LabelProfile'}], '999', False, ), # identity has no linked label profiles at all ([], '123', False), # id comes in as int on the stored profile but string from the url ( [{'profile_id': 123, 'profile_type': 'LabelProfile'}], 123, True, ), ], ) def test_has_label_profile_access(mocker, linked_profiles, requested_id, expected): """Test label profile access check against the linked-label-profiles query.""" orchard_identity = '5b83678aad451875aad4aaef' mocker.patch.object( profiles_model, 'get_linked_label_profiles', return_value=linked_profiles, autospec=True, ) actual = profiles.has_label_profile_access(orchard_identity, requested_id) assert actual.message == {'has_access': expected} profiles_model.get_linked_label_profiles.assert_called_once_with(orchard_identity) @pytest.mark.parametrize( 'role,db_result,' 'show_songwhip_feature,show_account360_app_switcher_feature,' 'show_help_center_feature,' 'brand,expected_result', [ pytest.param( None, [ {'profile_id': 1346777, 'profile_type': 'PodcastProfile'}, {'profile_id': 23234234, 'profile_type': 'LabelProfile'}, {'profile_id': 23234234, 'profile_type': 'InsightsProfile'}, {'profile_id': 23234235, 'profile_type': 'MoneyhubProfile'}, {'profile_id': 23234236, 'profile_type': 'CollaboratorsProfile'}, {'profile_id': 23234237, 'profile_type': 'SongwhipProfile'}, {'profile_id': 23234238, 'profile_type': 'AudienceProfile'}, {'profile_id': 23234239, 'profile_type': 'DistributionProfile'}, ], 'enabled', 'enabled', 'enabled', constants.ORCHARD_BRAND, [ { 'name': 'Workstation', 'id': 'workstation', 'url': 'https://workstation.qaorch.com', 'roles': [], }, { 'name': 'Insights', 'id': 'insights', 'url': 'https://insights.qaorch.com', 'roles': [], }, { 'name': 'Accounting', 'id': 'moneyhub', 'url': 'https://moneyhub.qaorch.com', 'roles': [], }, { 'name': 'Collaborators', 'id': 'collaborators', 'url': 'https://collaborators.qaorch.com', 'roles': [], }, { 'name': 'Songwhip', 'id': 'songwhip', 'url': 'https://staging.songwhip.com/login', 'roles': [], }, { 'name': 'Fansifter', 'id': 'audience-development', 'url': 'https://fansifter.qaorch.com', 'roles': [], }, { 'name': 'Distribution & Collection', 'id': 'distribution', 'url': 'https://distribution.qaorch.com', 'roles': [], }, { 'id': 'help-center', 'name': 'Help & Support', 'roles': [], 'url': 'https://help.theorchard.com/hc', }, ], id='theorchard_1', ), pytest.param( None, [ {'profile_id': 1346777, 'profile_type': 'PodcastProfile'}, {'profile_id': 23234234, 'profile_type': 'LabelProfile'}, {'profile_id': 23234234, 'profile_type': 'InsightsProfile'}, {'profile_id': 23234235, 'profile_type': 'MoneyhubProfile'}, {'profile_id': 23234236, 'profile_type': 'CollaboratorsProfile'}, {'profile_id': 23234237, 'profile_type': 'SongwhipProfile'}, {'profile_id': 23234238, 'profile_type': 'AudienceProfile'}, {'profile_id': 23234239, 'profile_type': 'DistributionProfile'}, ], 'enabled', 'enabled', 'enabled', constants.ORCHARD_BRAND, [ { 'name': 'Workstation', 'id': 'workstation', 'url': 'https://workstation.qaorch.com', 'roles': [], }, { 'name': 'Insights', 'id': 'insights', 'url': 'https://insights.qaorch.com', 'roles': [], }, { 'name': 'Accounting', 'id': 'moneyhub', 'url': 'https://moneyhub.qaorch.com', 'roles': [], }, { 'name': 'Collaborators', 'id': 'collaborators', 'url': 'https://collaborators.qaorch.com', 'roles': [], }, { 'name': 'Songwhip', 'id': 'songwhip', 'url': 'https://staging.songwhip.com/login', 'roles': [], }, { 'name': 'Fansifter', 'id': 'audience-development', 'url': 'https://fansifter.qaorch.com', 'roles': [], }, { 'name': 'Distribution & Collection', 'id': 'distribution', 'url': 'https://distribution.qaorch.com', 'roles': [], }, { 'id': 'help-center', 'name': 'Help & Support', 'roles': [], 'url': 'https://help.theorchard.com/hc', }, ], id='theorchard_2', ), pytest.param( None, [ {'profile_id': 1346777, 'profile_type': 'PodcastProfile'}, {'profile_id': 23234234, 'profile_type': 'LabelProfile'}, {'profile_id': 23234234, 'profile_type': 'InsightsProfile'}, {'profile_id': 23234235, 'profile_type': 'MoneyhubProfile'}, {'profile_id': 23234236, 'profile_type': 'CollaboratorsProfile'}, {'profile_id': 23234237, 'profile_type': 'SongwhipProfile'}, {'profile_id': 23234238, 'profile_type': 'AudienceProfile'}, ], 'control', 'control', 'control', constants.ORCHARD_BRAND, [ { 'name': 'Workstation', 'id': 'workstation', 'url': 'https://workstation.qaorch.com', 'roles': [], }, { 'name': 'Insights', 'id': 'insights', 'url': 'https://insights.qaorch.com', 'roles': [], }, { 'name': 'Accounting', 'id': 'moneyhub', 'url': 'https://moneyhub.qaorch.com', 'roles': [], }, { 'name': 'Collaborators', 'id': 'collaborators', 'url': 'https://collaborators.qaorch.com', 'roles': [], }, { 'name': 'Fansifter', 'id': 'audience-development', 'url': 'https://fansifter.qaorch.com', 'roles': [], }, ], id='theorchard_3', ), pytest.param( None, [ {'profile_id': 1346777, 'profile_type': 'PodcastProfile'}, {'profile_id': 23234234, 'profile_type': 'LabelProfile'}, {'profile_id': 23234234, 'profile_type': 'InsightsProfile'}, {'profile_id': 23234235, 'profile_type': 'MoneyhubProfile'}, {'profile_id': 23234236, 'profile_type': 'CollaboratorsProfile'}, ], 'control', 'control', 'control', constants.ORCHARD_BRAND, [ { 'name': 'Workstation', 'id': 'workstation', 'url': 'https://workstation.qaorch.com', 'roles': [], }, { 'name': 'Insights', 'id': 'insights', 'url': 'https://insights.qaorch.com', 'roles': [], }, { 'name': 'Accounting', 'id': 'moneyhub', 'url': 'https://moneyhub.qaorch.com', 'roles': [], }, { 'name': 'Collaborators', 'id': 'collaborators', 'url': 'https://collaborators.qaorch.com', 'roles': [], }, ], id='theorchard_4', ), pytest.param( None, [ {'profile_id': 1346777, 'profile_type': 'UnknownProfile'}, ], 'control', 'control', 'control', constants.ORCHARD_BRAND, [], id='theorchard_5', ), pytest.param( None, [ {'profile_id': 1346777, 'profile_type': 'PodcastProfile'}, {'profile_id': 23234234, 'profile_type': 'InsightsProfile'}, ], 'control', 'control', 'control', constants.ORCHARD_BRAND, [ { 'name': 'Insights', 'id': 'insights', 'url': 'https://insights.qaorch.com', 'roles': [], } ], id='theorchard_6', ), pytest.param( None, [ { 'profile_id': 1346777, 'profile_type': 'PodcastProfile', 'roles': ['networking', 'catalog', 'admin'], }, { 'profile_id': 23234234, 'profile_type': 'LabelProfile', 'roles': ['analytics'], 'updated_by': 'foo', }, { 'profile_id': 23234235, 'profile_type': 'LabelProfile', 'roles': ['accounting', 'catalog'], }, { 'profile_id': 23234235, 'profile_type': 'InsightsProfile', 'roles': ['administrator'], }, ], 'control', 'control', 'control', constants.ORCHARD_BRAND, [ { 'name': 'Workstation', 'id': 'workstation', 'url': 'https://workstation.qaorch.com', 'roles': ['accounting', 'analytics', 'catalog'], }, { 'name': 'Insights', 'id': 'insights', 'url': 'https://insights.qaorch.com', 'roles': ['administrator'], }, ], id='theorchard_7', ), pytest.param( 'administrator', [ { 'profile_id': 1346777, 'profile_type': 'InsightsProfile', 'roles': ['administrator'], }, { 'profile_id': 23234234, 'profile_type': 'LabelProfile', 'roles': ['analytics'], }, ], 'control', 'control', 'control', constants.ORCHARD_BRAND, [ { 'name': 'Insights', 'id': 'insights', 'url': 'https://insights.qaorch.com', 'roles': ['administrator'], } ], id='theorchard_8', ), pytest.param( 'admin', [ { 'profile_id': 1346777, 'profile_type': 'InsightsProfile', 'roles': ['analytics'], }, { 'profile_id': 23234234, 'profile_type': 'LabelProfile', 'roles': ['analytics'], }, ], 'control', 'control', 'control', constants.ORCHARD_BRAND, [], id='theorchard_9', ), pytest.param( None, [ { 'profile_id': 1346777, 'profile_type': 'InsightsProfile', 'roles': ['analytics'], }, { 'profile_id': 23234234, 'profile_type': 'LabelProfile', 'roles': ['analytics'], }, {'profile_id': 500, 'profile_type': 'SettingsProfile'}, ], 'control', 'control', 'control', constants.ORCHARD_BRAND, [ { 'id': 'insights', 'name': 'Insights', 'roles': ['analytics'], 'url': 'https://insights.qaorch.com', }, { 'id': 'workstation', 'name': 'Workstation', 'roles': ['analytics'], 'url': 'https://workstation.qaorch.com', }, { 'id': 'settings', 'name': 'Settings', 'roles': [], 'url': 'https://settings.qaorch.com', }, ], id='theorchard_10', ), # awal brand pytest.param( None, [ { 'profile_id': 1346777, 'profile_type': 'PodcastProfile', 'roles': ['networking', 'catalog', 'admin'], }, { 'profile_id': 23234234, 'profile_type': 'LabelProfile', 'roles': ['analytics'], 'updated_by': 'foo', }, { 'profile_id': 23234235, 'profile_type': 'LabelProfile', 'roles': ['accounting', 'catalog'], }, { 'profile_id': 23234235, 'profile_type': 'InsightsProfile', 'roles': ['administrator'], }, {'profile_id': 23234237, 'profile_type': 'SongwhipProfile'}, ], 'enabled', 'control', 'control', constants.AWAL_BRAND, [ { 'name': 'Workstation', 'id': 'workstation', 'url': 'https://workstation.qaawal.com', 'roles': ['accounting', 'analytics', 'catalog'], }, { 'name': 'Insights', 'id': 'insights', 'url': 'https://insights.qaawal.com', 'roles': ['administrator'], }, { 'name': 'Songwhip', 'id': 'songwhip', 'url': 'https://staging.songwhip.com/login', 'roles': [], }, ], id='awal_1', ), pytest.param( None, [], 'control', 'control', 'enabled', constants.AWAL_BRAND, [ { 'name': 'Help & Support', 'id': 'help-center', 'url': 'https://awal-support.zendesk.com/', 'roles': [], } ], id='awal_2', ), pytest.param( 'administrator', [ { 'profile_id': 1346777, 'profile_type': 'InsightsProfile', 'roles': ['administrator'], }, { 'profile_id': 23234234, 'profile_type': 'LabelProfile', 'roles': ['analytics'], }, ], 'control', 'control', 'control', constants.AWAL_BRAND, [ { 'name': 'Insights', 'id': 'insights', 'url': 'https://insights.qaawal.com', 'roles': ['administrator'], } ], id='awal_3', ), pytest.param( 'admin', [ { 'profile_id': 1346777, 'profile_type': 'InsightsProfile', 'roles': ['analytics'], }, { 'profile_id': 23234234, 'profile_type': 'LabelProfile', 'roles': ['analytics'], }, ], 'control', 'control', 'control', constants.AWAL_BRAND, [], id='awal_4', ), # knr brand pytest.param( 'administrator', [ { 'profile_id': 1346777, 'profile_type': 'InsightsProfile', 'roles': ['administrator'], }, { 'profile_id': 23234234, 'profile_type': 'LabelProfile', 'roles': ['analytics'], }, {'profile_id': 500, 'profile_type': 'SettingsProfile', 'roles': ['administrator']}, ], 'control', 'control', 'control', constants.KNR_BRAND, [ { 'name': 'Insights', 'id': 'insights', 'url': 'https://insights.qapdesuite.com', 'roles': ['administrator'], }, { 'id': 'settings', 'name': 'Settings', 'roles': ['administrator'], 'url': 'https://settings.qakollectivenr.com', }, ], id='knr_1', ), # sony brand pytest.param( None, [ {'profile_id': 1346777, 'profile_type': 'PodcastProfile'}, {'profile_id': 23234234, 'profile_type': 'LabelProfile'}, {'profile_id': 23234234, 'profile_type': 'InsightsProfile'}, {'profile_id': 23234235, 'profile_type': 'MoneyhubProfile'}, {'profile_id': 23234236, 'profile_type': 'CollaboratorsProfile'}, {'profile_id': 23234237, 'profile_type': 'SongwhipProfile'}, {'profile_id': 23234238, 'profile_type': 'AudienceProfile'}, {'profile_id': 23234239, 'profile_type': 'DistributionProfile'}, ], 'enabled', 'enabled', 'enabled', constants.SONY_BRAND, [ { 'name': 'Workstation', 'id': 'workstation', 'url': 'https://workstation.qaorch.com', 'roles': [], }, { 'name': 'Insights', 'id': 'insights', 'url': 'https://insights.qapdesuite.com', 'roles': [], }, { 'name': 'Accounting', 'id': 'moneyhub', 'url': 'https://moneyhub.qaorch.com', 'roles': [], }, { 'name': 'Collaborators', 'id': 'collaborators', 'url': 'https://collaborators.qaorch.com', 'roles': [], }, { 'name': 'Songwhip', 'id': 'songwhip', 'url': 'https://staging.songwhip.com/login', 'roles': [], }, { 'name': 'Fansifter', 'id': 'audience-development', 'url': 'https://fansifter.qapdesuite.com', 'roles': [], }, { 'name': 'Distribution & Collection', 'id': 'distribution', 'url': 'https://distribution.qaorch.com', 'roles': [], }, { 'id': 'help-center', 'name': 'Help & Support', 'roles': [], 'url': 'https://sonymusic-sme.zendesk.com/hc', }, ], id='sme_1', ), pytest.param( None, [ { 'profile_id': 1346777, 'profile_type': 'PodcastProfile', 'roles': ['networking', 'catalog', 'admin'], }, { 'profile_id': 23234234, 'profile_type': 'LabelProfile', 'roles': ['analytics'], 'updated_by': 'foo', }, { 'profile_id': 23234235, 'profile_type': 'LabelProfile', 'roles': ['accounting', 'catalog'], }, { 'profile_id': 23234235, 'profile_type': 'InsightsProfile', 'roles': ['administrator'], }, ], 'control', 'control', 'control', constants.SONY_BRAND, [ { 'name': 'Workstation', 'id': 'workstation', 'url': 'https://workstation.qaorch.com', 'roles': ['accounting', 'analytics', 'catalog'], }, { 'name': 'Insights', 'id': 'insights', 'url': 'https://insights.qapdesuite.com', 'roles': ['administrator'], }, ], id='sme_2', ), pytest.param( 'administrator', [ { 'profile_id': 1346777, 'profile_type': 'InsightsProfile', 'roles': ['administrator'], }, { 'profile_id': 23234234, 'profile_type': 'LabelProfile', 'roles': ['analytics'], }, ], 'control', 'control', 'control', constants.SONY_BRAND, [ { 'name': 'Insights', 'id': 'insights', 'url': 'https://insights.qapdesuite.com', 'roles': ['administrator'], } ], id='sme_3', ), pytest.param( 'admin', [ { 'profile_id': 1346777, 'profile_type': 'InsightsProfile', 'roles': ['analytics'], }, { 'profile_id': 23234234, 'profile_type': 'LabelProfile', 'roles': ['analytics'], }, ], 'control', 'control', 'control', constants.SONY_BRAND, [], id='sme_4', ), # some other brand will get orchard branded urls. pytest.param( None, [ { 'profile_id': 1346777, 'profile_type': 'PodcastProfile', 'roles': ['networking', 'catalog', 'admin'], }, { 'profile_id': 23234234, 'profile_type': 'LabelProfile', 'roles': ['analytics'], 'updated_by': 'foo', }, { 'profile_id': 23234235, 'profile_type': 'LabelProfile', 'roles': ['accounting', 'catalog'], }, { 'profile_id': 23234235, 'profile_type': 'InsightsProfile', 'roles': ['administrator'], }, ], 'control', 'control', 'control', constants.OVERDRIVE_BRAND, [ { 'name': 'Workstation', 'id': 'workstation', 'url': 'https://workstation.qaorch.com', 'roles': ['accounting', 'analytics', 'catalog'], }, { 'name': 'Insights', 'id': 'insights', 'url': 'https://insights.qaorch.com', 'roles': ['administrator'], }, ], id='overdrive_1', ), # Test case for Abacus application pytest.param( None, [ {'profile_id': 12345678, 'profile_type': 'AbacusProfile'}, {'profile_id': 23234234, 'profile_type': 'LabelProfile'}, {'profile_id': 23234235, 'profile_type': 'InsightsProfile'}, ], 'control', 'control', 'control', constants.ORCHARD_BRAND, [ { 'name': 'Abacus', 'id': 'abacus', 'url': 'https://abacus.qaorch.com', 'roles': [], }, { 'name': 'Workstation', 'id': 'workstation', 'url': 'https://workstation.qaorch.com', 'roles': [], }, { 'name': 'Insights', 'id': 'insights', 'url': 'https://insights.qaorch.com', 'roles': [], }, ], id='abacus_profile_test', ), # Test case for Account360Profile user getting ABACUS access # (even with feature flag disabled) pytest.param( None, [ { 'profile_id': 572918, 'profile_type': 'Account360Profile', 'roles': ['account360'], }, { 'profile_id': 23234234, 'profile_type': 'LabelProfile', 'roles': ['administrator'], }, ], 'control', 'disabled', # Account360 app switcher feature is disabled 'control', constants.ORCHARD_BRAND, [ { 'name': 'Workstation', 'id': 'workstation', 'url': 'https://workstation.qaorch.com', 'roles': ['administrator'], }, { 'name': 'Abacus', 'id': 'abacus', 'url': 'https://abacus.qaorch.com', 'roles': ['account360'], # Should inherit roles from Account360Profile }, ], id='account360_profile_gets_abacus', ), ], ) def test_get_applications_for_identity_tx( mocker, role, db_result, show_songwhip_feature, show_account360_app_switcher_feature, show_help_center_feature, brand, expected_result, ): """Test get list of all applications for a given identity using managed transactions.""" def mocked_get_single_feature_by_attributes(flag): if flag == constants.FEATURE_SHOW_SONGWHIP_APP: return response.Response(message=show_songwhip_feature) if flag == constants.FEATURE_ACCOUNT360_APP_SWITCHER: return response.Response(message=show_account360_app_switcher_feature) elif flag == constants.FEATURE_HELP_CENTER_APP: return response.Response(message=show_help_center_feature) identity_id = 'uuid-1' mock_session = mocker.MagicMock() mock_session.run.return_value = None # ensure hasattr(session, 'run') is True mocker.patch('users.logic.profiles.get_session', return_value=mock_session) mocker.patch.object( profiles_model, 'get_profiles_for_applications_tx', return_value=response.Response(db_result), autospec=True, ) mocker.patch.object( pythonfeatures, 'get_single_feature_by_attributes', side_effect=lambda flag, _: mocked_get_single_feature_by_attributes(flag), ) result = profiles.get_applications_for_identity_tx(identity_id, role, brand) assert result.message['items'] == expected_result