"""Test for identity_endpoints_handler.""" import json from unittest.mock import MagicMock, patch import flask from flask import testing from owsresponse import response import pytest from pythonfeatures import pythonfeatures from users import ( constants, # noqa: F401 ) from users.logic import ( auth0_application_access, auth0_client, devices, profiles, user_info, ) from users.models import ( devices as devices_model, identities, ows_notifications, ows_pdp, profiles as profiles_model, ) @pytest.mark.parametrize( ('payload', 'expected_response'), [ # valid requests ( {'push_token': 'foo-2345', 'platform_type': 'ios'}, response.Response(message={'foo': 'bar'}), ), ( {'push_token': 'foo-2345', 'device_id': 'device01', 'platform_type': 'ios'}, response.Response(message={'foo': 'bar'}), ), ( { 'push_token': 'foo-2345', 'device_id': 'device01', 'platform_type': 'ios', 'localization': 'fr', }, response.Response(message={'foo': 'bar'}), ), ( { 'push_token': 'foo-2345', 'device_id': 'device01', 'platform_type': 'ios', 'brand': 'awal', }, response.Response(message={'foo': 'bar'}), ), # invalid requests ( { 'push_token': 'foo-2345', 'device_id': 'device01', 'platform_type': 'ios', 'brand': 'foo', }, response.create_error_response( 'validation_error', { 'brand': [ 'Must be one of: theorchard, sme, overdrive, awal, knr, ' 'orchard, helpcenter, ab, msk, altafonte.' ] }, ), # noqa ), ( {'push_token': 'foo-2345'}, response.create_error_response( 'validation_error', {'platform_type': ['Missing data for required field.']} ), ), ], ) @patch('users.handlers.identity.Neo4jSession.__enter__') @patch('users.handlers.identity.Neo4jSession.__exit__') def test_add_user_device( neo4j_exit, neo4j_enter, payload, expected_response, mocker, fixture_client ): """Test 'POST /users/identity//device' handler.""" orchard_identity_id = 'abcdfghj123456789qweqz' create_profile_mock = mocker.patch('users.logic.devices.add_push_notification_device') create_profile_mock.return_value = response.Response(message={'foo': 'bar'}) handler_response = fixture_client.post( f'/users/identity/{orchard_identity_id}/device', json=payload ) result_data = json.loads(handler_response.data.decode()) assert handler_response.status_code == expected_response.status if expected_response: assert result_data == expected_response.message assert create_profile_mock.called assert create_profile_mock.call_args_list[0][0][0] == orchard_identity_id assert create_profile_mock.call_args_list[0][0][1] == payload # correlation id assert type(create_profile_mock.call_args_list[0][0][2]) is str else: assert result_data == expected_response.errors @patch('users.handlers.identity.Neo4jSession.__enter__') @patch('users.handlers.identity.Neo4jSession.__exit__') @patch('users.handlers.identity.devices') def test_add_user_device_not_found( devices_mock: MagicMock, _exit: MagicMock, _enter: MagicMock, fixture_client: testing.FlaskClient, ) -> None: """Test DeviceNotFoundError handling for POST /users/identity//device.""" devices_mock.add_push_notification_device.side_effect = devices.DeviceNotFoundError('☹️') handler_response = fixture_client.post( '/users/identity/abcdfghj123456789qweqz/device', json={'push_token': 'foo-2345', 'device_id': 'device01', 'platform_type': 'ios'}, ) result_data = json.loads(handler_response.data.decode()) assert handler_response.status_code == 404 assert result_data == {'code': 'not_found', 'message': '☹️'} @patch('users.handlers.identity.Neo4jSession.__enter__') @patch('users.handlers.identity.Neo4jSession.__exit__') def test_get_user_device(neo4j_exit, neo4j_enter, mocker, fixture_client): """Test 'GET /users/identity//device' handler.""" orchard_identity_id = 'abc' expected_response = response.Response(message={'foo': 'bar'}) get_profile_mock = mocker.patch('users.logic.devices.get_push_notification_device') get_profile_mock.return_value = expected_response handler_response = fixture_client.get(f'/users/identity/{orchard_identity_id}/device') result_data = json.loads(handler_response.data.decode()) assert get_profile_mock.called get_profile_mock.assert_called_with(orchard_identity_id) assert handler_response.status_code == expected_response.status assert result_data == expected_response.message @pytest.mark.parametrize( ('delete_device_response', 'revoke_response', 'expected_result', 'revoke_called'), ( ( response.create_not_found_response(), response.Response(message={'foo': 'bar'}), response.Response(message={'foo': 'bar'}), True, ), ( response.Response(status=204), response.Response(message={'foo': 'bar'}), response.Response(message={'foo': 'bar'}), True, ), ( response.create_fatal_response('some error!'), response.Response(message={'foo': 'bar'}), response.create_fatal_response('some error!'), False, ), ), ) @patch.object(auth0_client, 'revoke_refresh_token') @patch('users.handlers.identity.Neo4jSession.__enter__') @patch('users.handlers.identity.Neo4jSession.__exit__') def test_logout_mobile_device( neo4j_exit, neo4j_enter, revoke_mock, delete_device_response, revoke_response, expected_result, revoke_called, mocker, fixture_client, ): """Test 'DELETE /logout/identity/ handler.""" orchard_identity_id = 'abc' revoke_mock.return_value = revoke_response delete_device_mock = mocker.patch('users.logic.devices.delete_push_notification_device') delete_device_mock.return_value = delete_device_response payload = { 'refresh_token': 'foobar1234', 'device_id': 'foo-2345', } handler_response = fixture_client.post(f'/logout/identity/{orchard_identity_id}', json=payload) result_data = json.loads(handler_response.data.decode()) assert handler_response.status_code == expected_result.status if revoke_called: assert revoke_mock.called revoke_mock.assert_called_with(payload['refresh_token']) assert result_data == expected_result.message else: assert not revoke_mock.called assert result_data == expected_result.errors @pytest.mark.parametrize( ('email', 'email_to_logic'), [ ['test@theorchard.com', 'test@theorchard.com'], ['Test@theorchard.com', 'test@theorchard.com'], ['fooBar@theorchard.com', 'foobar@theorchard.com'], ['FOO@theorchard.com', 'foo@theorchard.com'], ], ) @patch('users.handlers.identity.Neo4jSession.__enter__') @patch('users.handlers.identity.Neo4jSession.__exit__') def test_get_identity_by_email( neo4j_exit, neo4j_enter, mocker, fixture_client, email, email_to_logic ): """Test get_identity_by_email.""" expected_response = response.Response( {'name': 'New User', 'email': email, 'id': '5dd7d78c4ec49c0e2c4e4d1f'} ) get_mock = mocker.patch('users.logic.profiles.get_identity_by_email') get_mock.return_value = expected_response handler_response = fixture_client.get(f'/users/identity/email/{email}') assert get_mock.called get_mock.assert_called_with(email_to_logic) assert handler_response.status_code == 200 @patch.object(auth0_client, 'revoke_refresh_token') @patch('users.handlers.identity.Neo4jSession.__enter__') @patch('users.handlers.identity.Neo4jSession.__exit__') def test_logout_mobile_device_error(neo4j_exit, neo4j_enter, revoke_mock, mocker, fixture_client): """Test 'DELETE /logout/identity/ handler with exception.""" orchard_identity_id = 'abc' revoke_mock.side_effect = Exception('Some Auth0 error') payload = { 'refresh_token': 'foobar1234', 'device_id': 'foo-2345', } expected_response = response.Response(status=204, message='') delete_device_mock = mocker.patch('users.logic.devices.delete_push_notification_device') delete_device_mock.return_value = expected_response handler_response = fixture_client.post(f'/logout/identity/{orchard_identity_id}', json=payload) assert revoke_mock.called revoke_mock.assert_called_with(payload['refresh_token']) assert handler_response.status_code == 500 # flask returns generic msg. @patch('users.handlers.identity.Neo4jSession.__enter__') @patch('users.handlers.identity.Neo4jSession.__exit__') def test_remove_user_device(neo4j_exit, neo4j_enter, mocker, fixture_client): """Test 'DELETE /users/identity//device/ handler.""" orchard_identity_id = 'abc' device_id = 'xyz' expected_response = response.Response(status=204, message='') delete_device_mock = mocker.patch('users.logic.devices.delete_push_notification_device') delete_device_mock.return_value = expected_response handler_response = fixture_client.delete( f'/users/identity/{orchard_identity_id}/device/{device_id}' ) assert delete_device_mock.called assert delete_device_mock.called_with(orchard_identity_id, device_id) assert handler_response.status_code == expected_response.status result_data = handler_response.data.decode() assert result_data == expected_response.message @pytest.mark.parametrize( ('email', 'email_to_logic'), [ ['test@theorchard.com', 'test@theorchard.com'], ['Test@theorchard.com', 'test@theorchard.com'], ['fooBar@theorchard.com', 'foobar@theorchard.com'], ['FOO@theorchard.com', 'foo@theorchard.com'], ], ) @patch('users.handlers.identity.Neo4jSession.__enter__') @patch('users.handlers.identity.Neo4jSession.__exit__') def test_update_identity_by_email( neo4j_exit, neo4j_enter, mocker, fixture_client, email, email_to_logic ): """Test update_identity_by_email handler.""" data = {'google_user_id': 'id1234567891011121314'} expected_response = response.Response(status=204, message='') update_mock = mocker.patch('users.logic.profiles.update_identity_by_email') update_mock.return_value = expected_response handler_response = fixture_client.patch(f'/users/identity/email/{email}', json=data) assert handler_response.status_code == expected_response.status result_data = handler_response.data.decode() assert result_data == expected_response.message assert update_mock.called assert update_mock.called_once_with(email_to_logic, data) @pytest.mark.parametrize( ('email', 'email_to_logic'), [ ['test@theorchard.com', 'test@theorchard.com'], ['Test@theorchard.com', 'test@theorchard.com'], ['fooBar@theorchard.com', 'foobar@theorchard.com'], ['FOO@theorchard.com', 'foo@theorchard.com'], ], ) @patch('users.handlers.identity.Neo4jSession.__enter__') @patch('users.handlers.identity.Neo4jSession.__exit__') def test_update_identity_by_email_backend( neo4j_exit, neo4j_enter, mocker, fixture_client, email, email_to_logic ): """Test update_identity_by_email_backend handler.""" data = { 'auth0_user_id': 'id1234567891011121314345', 'identity_id': 'eb480170-c5de-49ff-bb31-3a0306b12a07', } expected_response = response.Response(status=204, message='') update_mock = mocker.patch('users.logic.profiles.update_identity_and_id_by_email') update_mock.return_value = expected_response handler_response = fixture_client.patch(f'/ows/users/identity/email/{email}', json=data) result_data = handler_response.data.decode() assert handler_response.status_code == expected_response.status assert result_data == expected_response.message assert update_mock.called assert update_mock.called_once_with(email_to_logic, data) @patch('users.handlers.identity.Neo4jSession.__enter__') @patch('users.handlers.identity.Neo4jSession.__exit__') def test_add_user_device_adds_profile_subs(neo4j_exit, neo4j_enter, fixture_client, mocker): """Test adding a device adds profile subs. It should check for feature flags from request context in device logic. """ identity_id = 'whatever' mocker.patch.object( identities, 'get_identity', return_value=response.Response(message={'id': 'whatever'}) ) mocker.patch.object( profiles_model, 'get_profiles', return_value=response.Response( message=[ {'profile_id': 'whatever1', 'profile_type': 'banana'}, {'profile_id': 'whatever2', 'profile_type': 'pizza'}, ] ), ) mocker.patch.object( pythonfeatures, 'get_single_feature', return_value=response.Response(message='enabled') ) mocker.patch.object( ows_notifications, 'create_notification_subscription', return_value=response.Response() ) mocker.patch.object(devices.SNS_CLIENT, 'create_topic', return_value=MagicMock()) mocker.patch.object(devices.SNS_CLIENT, 'subscribe', return_value=MagicMock()) mocker.patch.object(devices, 'register_device', return_value=MagicMock()) mocker.patch.object(devices, 'delete_stale_registrations', return_value=MagicMock()) mocker.patch.object( devices_model, 'create_push_notification_device', return_value=response.Response() ) fixture_client.post( '/users/identity/{}/device'.format(identity_id), json={ 'device_id': '34D9725F-1B6D-460A-8799-43586EE66A9C', 'platform_type': 'ios', 'push_token': '0a50f757b40e89108bb249effcadca4c86630d10e82fbf7d2ca1850c275f7d59', }, headers={'Orchard-Identity-Id': identity_id}, ) @patch('users.handlers.identity.Neo4jSession.__enter__') @patch('users.handlers.identity.Neo4jSession.__exit__') def test_get_identity_by_auth0_id_success(neo4j_exit, neo4j_enter, mocker, fixture_client): """Test get_identity_by_auth0_id success case.""" auth0_user_id = '5e8b890c5581720c54fb002b' expected_response = response.Response( { 'name': 'Test User', 'email': 'test@example.com', 'id': '5dd7d78c4ec49c0e2c4e4d1f', 'auth0UserId': auth0_user_id, } ) get_mock = mocker.patch('users.logic.profiles.get_identity_by_auth0_id') get_mock.return_value = expected_response handler_response = fixture_client.get(f'/users/identity/auth0/{auth0_user_id}') assert get_mock.called get_mock.assert_called_with(auth0_user_id) assert handler_response.status_code == 200 @patch('users.handlers.identity.Neo4jSession.__enter__') @patch('users.handlers.identity.Neo4jSession.__exit__') def test_get_identity_by_auth0_id_not_found(neo4j_exit, neo4j_enter, mocker, fixture_client): """Test get_identity_by_auth0_id not found case.""" auth0_user_id = 'nonexistent123456789' expected_response = response.create_not_found_response(message='Identity not found') get_mock = mocker.patch('users.logic.profiles.get_identity_by_auth0_id') get_mock.return_value = expected_response handler_response = fixture_client.get(f'/users/identity/auth0/{auth0_user_id}') assert get_mock.called get_mock.assert_called_with(auth0_user_id) assert handler_response.status_code == 404 @patch('users.handlers.identity.Neo4jSession.__enter__') @patch('users.handlers.identity.Neo4jSession.__exit__') def test_get_identity_by_auth0_id_forbidden_in_prod( neo4j_exit, neo4j_enter, mocker, fixture_client ): """Test get_identity_by_auth0_id is forbidden in production.""" auth0_user_id = '5e8b890c5581720c54fb002b' mocker.patch('users.handlers.identity.config.ENVIRONMENT', 'prod') get_mock = mocker.patch('users.logic.profiles.get_identity_by_auth0_id') handler_response = fixture_client.get(f'/users/identity/auth0/{auth0_user_id}') # Should not call the logic function in production assert not get_mock.called assert handler_response.status_code == 403 @pytest.mark.parametrize('url_prefix', ['/identities', '/auth0/users/identities']) @patch('users.utils.api_utils.g') def test_get_application_access_no_jwt( mock_g: MagicMock, fixture_client: testing.FlaskClient, app_context: flask.ctx.AppContext, url_prefix: str, ) -> None: """Test get_application_access endpoint without a JWT.""" mock_g.request_context.jwt_identity_id = None resp = fixture_client.get(f'{url_prefix}/test-identity-id/application-access/seat') assert resp.status_code == 401 assert resp.json == { 'code': 'authorization_error', 'message': 'Request context has no identity uuid.', } @pytest.mark.parametrize('url_prefix', ['/identities', '/auth0/users/identities']) @patch('users.handlers.identity.auth0_application_access') @patch('users.utils.api_utils.g') def test_get_application_access( mock_g: MagicMock, mock_application_access: MagicMock, fixture_client: testing.FlaskClient, app_context: flask.ctx.AppContext, url_prefix: str, ) -> None: """Test get_application_access endpoint.""" mock_g.request_context.jwt_identity_id = 'test-identity-id' mock_application_access.check_application_access.return_value = False resp = fixture_client.get(f'{url_prefix}/test-identity-id/application-access/seat') assert resp.status_code == 200 assert resp.json == {'has_access': False} mock_application_access.check_application_access.assert_called_once_with( 'test-identity-id', 'seat' ) @pytest.mark.parametrize( 'exception', [ pytest.param( ows_pdp.OwsPdpError(status_code=403, message='😡'), id='OwsPdpError', ), pytest.param( auth0_application_access.AppAccessCheckError(status_code=422, message='☹️'), id='AppAccessCheckError', ), ], ) @pytest.mark.parametrize('url_prefix', ['/identities', '/auth0/users/identities']) @patch('users.handlers.identity.auth0_application_access') @patch('users.utils.api_utils.g') def test_get_application_access_error( mock_g: MagicMock, mock_application_access: MagicMock, exception: Exception, fixture_client: testing.FlaskClient, app_context: flask.ctx.AppContext, url_prefix: str, ) -> None: """Test get_application_access endpoint with an error.""" mock_g.request_context.jwt_identity_id = 'test-identity-id' mock_application_access.check_application_access.side_effect = exception resp = fixture_client.get(f'{url_prefix}/test-identity-id/application-access/seat') assert resp.status_code == exception.status_code assert resp.json == {'code': 'access_check_error', 'message': str(exception)} # Tests moved from test_common_handlers.py for identity endpoints @pytest.mark.parametrize( ('post_data', 'expected_post_data'), [ [ # test email formatter. { 'email': 'test@theorchard.com', 'name': 'user name', 'identity_id': '46b961bb-ecaa-4fa7-a7b5-93ec651ddd08', }, { 'email': 'test@theorchard.com', 'name': 'user name', 'identity_id': '46b961bb-ecaa-4fa7-a7b5-93ec651ddd08', 'active': 'Y', 'user_types': [], 'localization': 'en', 'number_format': 'us', }, ], [ { 'email': 'Test@theorchard.com', 'name': 'user name', 'identity_id': '46b961bb-ecaa-4fa7-a7b5-93ec651ddd08', }, { 'email': 'test@theorchard.com', 'name': 'user name', 'identity_id': '46b961bb-ecaa-4fa7-a7b5-93ec651ddd08', 'active': 'Y', 'user_types': [], 'localization': 'en', 'number_format': 'us', }, ], [ { 'email': 'fooBar@theorchard.com', 'name': 'user name', 'identity_id': '46b961bb-ecaa-4fa7-a7b5-93ec651ddd08', }, { 'email': 'foobar@theorchard.com', 'name': 'user name', 'identity_id': '46b961bb-ecaa-4fa7-a7b5-93ec651ddd08', 'active': 'Y', 'user_types': [], 'localization': 'en', 'number_format': 'us', }, ], [ { 'email': 'FOO@theorchard.com', 'name': 'user name', 'identity_id': '46b961bb-ecaa-4fa7-a7b5-93ec651ddd08', }, { 'email': 'foo@theorchard.com', 'name': 'user name', 'identity_id': '46b961bb-ecaa-4fa7-a7b5-93ec651ddd08', 'active': 'Y', 'user_types': [], 'localization': 'en', 'number_format': 'us', }, ], [ # with optional fields. { 'email': 'FOO@theorchard.com', 'name': 'Foo Bar User', 'first_name': 'Foo', 'last_name': 'Bar', 'identity_id': '46b961bb-ecaa-4fa7-a7b5-93ec651ddd08', 'localization': 'es', 'number_format': 'eu', }, { 'email': 'foo@theorchard.com', 'name': 'Foo Bar User', 'first_name': 'Foo', 'last_name': 'Bar', 'identity_id': '46b961bb-ecaa-4fa7-a7b5-93ec651ddd08', 'active': 'Y', 'user_types': [], 'localization': 'es', 'number_format': 'eu', }, ], [ # with optional active fields. { 'email': 'FOO@theorchard.com', 'name': 'Foo Bar User', 'first_name': 'Foo', 'last_name': 'Bar', 'identity_id': '46b961bb-ecaa-4fa7-a7b5-93ec651ddd08', 'active': 'N', }, { 'email': 'foo@theorchard.com', 'name': 'Foo Bar User', 'first_name': 'Foo', 'last_name': 'Bar', 'identity_id': '46b961bb-ecaa-4fa7-a7b5-93ec651ddd08', 'active': 'N', 'user_types': [], 'localization': 'en', 'number_format': 'us', }, ], [ # with user_type. { 'email': 'FOO@theorchard.com', 'name': 'Foo Bar User', 'first_name': 'Foo', 'last_name': 'Bar', 'identity_id': '46b961bb-ecaa-4fa7-a7b5-93ec651ddd08', 'active': 'N', 'user_types': ['artist'], }, { 'email': 'foo@theorchard.com', 'name': 'Foo Bar User', 'first_name': 'Foo', 'last_name': 'Bar', 'identity_id': '46b961bb-ecaa-4fa7-a7b5-93ec651ddd08', 'active': 'N', 'user_types': ['artist'], 'localization': 'en', 'number_format': 'us', }, ], ], ) @patch('users.handlers.identity.Neo4jSession.__enter__') @patch('users.handlers.identity.Neo4jSession.__exit__') def test_create_identity( neo4j_exit, neo4j_enter, mocker, fixture_client, post_data, expected_post_data ): """Test create_identity handler.""" expected_response = {'user': 'details'} mocker.patch.object( profiles, 'create_identity_in_graph', return_value=response.Response(expected_response) ) handler_response = fixture_client.post('/users/identity', json=post_data) assert handler_response.status_code == 200 result = json.loads(handler_response.data.decode()) assert result == {'user': 'details'} profiles.create_identity_in_graph.assert_called_with(expected_post_data) @patch('users.handlers.identity.Neo4jSession.__enter__') @patch('users.handlers.identity.Neo4jSession.__exit__') def test_create_identity_rejects_invalid_emails(neo4j_exit, neo4j_enter, mocker, fixture_client): mocker.patch.object( profiles, 'create_identity_in_graph', side_effect=AssertionError('create_identity_in_graph method should not have been called'), ) handler_response = fixture_client.post( '/users/identity', json={ 'email': 'not_an_email', 'name': 'user name', 'identity_id': '46b961bb-ecaa-4fa7-a7b5-93ec651ddd08', }, ) assert handler_response.status_code == 400, handler_response.text profiles.create_identity_in_graph.assert_not_called() @pytest.mark.parametrize( ('post_data', 'expected_post_data'), [ [ # only required fields. {'email': 'FOO@theorchard.com', 'name': 'user name'}, {'email': 'foo@theorchard.com', 'name': 'user name'}, ], [ # with optional fields. { 'email': 'FOO@theorchard.com', 'name': 'Foo Bar User', 'first_name': 'Foo', 'last_name': 'Bar', }, { 'email': 'foo@theorchard.com', 'name': 'Foo Bar User', 'first_name': 'Foo', 'last_name': 'Bar', }, ], ], ) @patch('users.handlers.identity.Neo4jSession.__enter__') @patch('users.handlers.identity.Neo4jSession.__exit__') def test_update_identity( neo4j_exit, neo4j_enter, mocker, fixture_client, post_data, expected_post_data ): """Test update_identity handler.""" expected_response = {'user': 'details'} identity_id = '46b961bb-ecaa-4fa7-a7b5-93ec651ddd08' mocker.patch.object( profiles, 'update_identity_in_graph', return_value=response.Response(expected_response) ) handler_response = fixture_client.patch(f'/users/identity/{identity_id}', json=post_data) assert handler_response.status_code == 200 result = json.loads(handler_response.data.decode()) assert result == {'user': 'details'} profiles.update_identity_in_graph.assert_called_with(identity_id, expected_post_data) @patch('users.handlers.identity.Neo4jSession.__enter__') @patch('users.handlers.identity.Neo4jSession.__exit__') def test_update_identity_rejects_invalid_emails(neo4j_exit, neo4j_enter, mocker, fixture_client): identity_id = '46b961bb-ecaa-4fa7-a7b5-93ec651ddd08' mocker.patch.object( profiles, 'update_identity_in_graph', side_effect=AssertionError( 'update_identity_in_graph method should not have been called', ), ) handler_response = fixture_client.patch( f'/users/identity/{identity_id}', json={ 'email': 'not_an_email', }, ) assert handler_response.status_code == 400, handler_response.text profiles.update_identity_in_graph.assert_not_called() @pytest.mark.parametrize( ('identity_id', 'participant_id', 'platform', 'logic_response', 'expected_status_code'), [ ('a', 'b', 'c', {'can_unlink': True}, 200), ('a', 'b', 'c', {'can_unlink': False}, 200), ], ) def test_get_can_unlink_participant_social_account( mocker, fixture_client, identity_id, participant_id, platform, logic_response, expected_status_code, ): """Test get_can_unlink_participant_social_account on valid request.""" mocker.patch.object( user_info, 'can_unlink_participant_social_account', return_value=response.Response(status=200, message=logic_response), ) handler_response = fixture_client.get( '/users/identity/{}/can_unlink_social_account?participant_id={}&platform={}'.format( identity_id, participant_id, platform ) ) result = json.loads(handler_response.data.decode()) assert handler_response.status_code == expected_status_code assert result['can_unlink'] == logic_response['can_unlink'] @pytest.mark.parametrize( 'query', [ '', 'participant_id=&platform=b', 'participant_id=a&platform=', 'participant_id=&platform=', 'a=&b=', ], ) def test_get_can_unlink_participant_social_account_bad_request(fixture_client, query): """Test get_can_unlink_participant_social_account on bad request.""" handler_response = fixture_client.get( '/users/identity/a/can_unlink_social_account?{}'.format(query) ) assert handler_response.status_code == 400 @pytest.mark.parametrize( 'query', [ '', 'participant_id=&platform=b', 'participant_id=a&platform=', 'participant_id=&platform=', 'a=&b=', ], ) def test_patch_unlink_participant_social_account_bad_request(fixture_client, query): """Test get_can_unlink_participant_social_account on bad request.""" handler_response = fixture_client.patch( '/users/identity/a/unlink_social_account?{}'.format(query) ) assert handler_response.status_code == 400 @pytest.mark.parametrize( ('identity_id', 'participant_id', 'platform', 'logic_response'), [ ( 'a', 'b', 'c', response.Response( status=200, message={ 'identity_id': 'a', 'participant_id': 'b', 'platform': 'c', 'linked': True, }, ), ), ( 'a', 'b', 'c', response.create_error_response( code='user_error', message='Participant b social account for c cannot be unlinked by a', ), ), ], ) def test_patch_unlink_participant_social_account( mocker, fixture_client, identity_id, participant_id, platform, logic_response ): """Test get_can_unlink_participant_social_account on valid request.""" mocker.patch.object(user_info, 'unlink_participant_social_account', return_value=logic_response) handler_response = fixture_client.patch( '/users/identity/{}/unlink_social_account?participant_id={}&platform={}'.format( identity_id, participant_id, platform ) ) result = json.loads(handler_response.data.decode()) assert handler_response.status_code == logic_response.status if handler_response.status_code == 200: assert result == logic_response.message else: assert result == logic_response.errors @patch('users.handlers.identity.Neo4jSession.__enter__') @patch('users.handlers.identity.Neo4jSession.__exit__') def test_get_all_profiles_for_identity(neo4j_exit, neo4j_enter, mocker, fixture_client): """Test get_all_profiles for identity.""" identity_id = '5dd7d78c4ec49c0e2c4e4d1f' expected_response = response.Response( { 'items': [ {'profile_id': 1346777, 'profile_type': 'ArtistProfile'}, {'profile_id': 23234234, 'profile_type': 'LabelProfile'}, ], 'pagination': {'type': 'none', 'total_records': 2}, } ) get_mock = mocker.patch('users.logic.profiles.get_all_profiles_for_identity') get_mock.return_value = expected_response handler_response = fixture_client.get(f'/users/identity/{identity_id}/profiles') # noqa assert get_mock.called get_mock.assert_called_with(identity_id) assert handler_response.status_code == 200 @patch('users.handlers.identity.Neo4jSession.__enter__') @patch('users.handlers.identity.Neo4jSession.__exit__') def test_check_label_profile_access(neo4j_exit, neo4j_enter, mocker, fixture_client): """Test label-profile-access returns {has_access: bool}.""" identity_id = '5dd7d78c4ec49c0e2c4e4d1f' label_profile_id = '1346777' logic_mock = mocker.patch('users.logic.profiles.has_label_profile_access') logic_mock.return_value = response.Response({'has_access': True}) handler_response = fixture_client.get( f'/users/identity/{identity_id}/label-profile-access/{label_profile_id}' ) logic_mock.assert_called_once_with(identity_id, label_profile_id) assert handler_response.status_code == 200 assert json.loads(handler_response.data.decode()) == {'has_access': True} @pytest.mark.parametrize( 'query_string, expected_brand', ( (None, None), ({'brand_experience': constants.ORCHARD_BRAND}, constants.ORCHARD_BRAND), ), ) @patch('users.handlers.identity.Neo4jSession.__enter__') @patch('users.handlers.identity.Neo4jSession.__exit__') def test_get_zendesk_token(_exit, _enter, query_string, expected_brand, fixture_client, mocker): """Test 'GET /users/identity//zendesk_token/' handler.""" identity_id = '' token_version = '' generate_zendesk_jwt_mock = mocker.patch( 'users.logic.zendesk.generate_zendesk_token', return_value=response.Response({'value': ''}), ) handler_response = fixture_client.get( f'/users/identity/{identity_id}/zendesk_token/{token_version}', query_string=query_string, ) result = json.loads(handler_response.data.decode()) generate_zendesk_jwt_mock.assert_called_once_with(identity_id, expected_brand, token_version) assert handler_response.status_code == 200 assert result == {'value': ''} @patch('users.handlers.identity.Neo4jSession.__enter__') @patch('users.handlers.identity.Neo4jSession.__exit__') def test_get_primary_vend_contact_success(neo4j_exit, neo4j_enter, mocker, fixture_client): """Test GET /users/identity//primary-vend-contact returns vend_contact_id.""" identity_id = 'abc12345-def6-7890-abcd-ef1234567890' mock_logic = mocker.patch( 'users.logic.user_info.get_primary_vend_contact_for_identity', return_value=response.Response(message={'vend_contact_id': 12345}), ) handler_response = fixture_client.get(f'/users/identity/{identity_id}/primary-vend-contact') result = json.loads(handler_response.data.decode()) assert handler_response.status_code == 200 assert result == {'vend_contact_id': 12345} mock_logic.assert_called_once_with(identity_id) @patch('users.handlers.identity.Neo4jSession.__enter__') @patch('users.handlers.identity.Neo4jSession.__exit__') def test_get_primary_vend_contact_not_found(neo4j_exit, neo4j_enter, mocker, fixture_client): """Test GET /users/identity//primary-vend-contact returns 404.""" identity_id = 'abc12345-def6-7890-abcd-ef1234567890' mocker.patch( 'users.logic.user_info.get_primary_vend_contact_for_identity', return_value=response.create_not_found_response( message=constants.ERROR_MESSAGE_IDENTITY_NOT_FOUND ), ) handler_response = fixture_client.get(f'/users/identity/{identity_id}/primary-vend-contact') assert handler_response.status_code == 404 @patch('users.handlers.identity.profiles') @patch('users.handlers.identity.g') @patch('users.handlers.identity.Neo4jSession.__enter__') @patch('users.handlers.identity.Neo4jSession.__exit__') def test_get_identity_by_id_logs_cross_identity_warning( neo4j_exit, neo4j_enter, mock_g, mock_profiles, fixture_client, app_context ): """Test warning is logged when caller identity differs from requested identity.""" mock_g.request_context.context_type = constants.PROFILE_CONTEXT_TYPE mock_g.request_context.profile_type = 'LabelProfile' mock_g.request_context.jwt_identity_id = 'caller-uuid' mock_profiles.get_identity_from_graph_tx.return_value = response.Response({'id': 'other-uuid'}) resp = fixture_client.get('/users/identity/other-uuid') assert resp.status_code == 200 mock_g.log.warning.assert_called_once_with( 'Cross-identity get_identity_by_id', resources={ 'caller_identity_id': 'caller-uuid', 'requested_identity_id': 'other-uuid', 'profile_type': 'LabelProfile', }, ) @patch('users.handlers.identity.profiles') @patch('users.handlers.identity.g') @patch('users.handlers.identity.Neo4jSession.__enter__') @patch('users.handlers.identity.Neo4jSession.__exit__') def test_get_identity_by_id_logs_ows_to_ows_warning( neo4j_exit, neo4j_enter, mock_g, mock_profiles, fixture_client, app_context ): """Test ows-to-ows warning is logged for non-profile (headerless) context.""" mock_g.request_context.context_type = 'headerless' mock_g.request_context.requestor_service_name = 'ows-permissions' mock_profiles.get_identity_from_graph_tx.return_value = response.Response({'id': 'other-uuid'}) resp = fixture_client.get('/users/identity/other-uuid') assert resp.status_code == 200 mock_g.log.warning.assert_called_once_with( 'ows-to-ows get_identity_by_id', resources={ 'requested_identity_id': 'other-uuid', 'context_type': 'headerless', 'requestor_service_name': 'ows-permissions', }, ) @patch('users.handlers.identity.profiles') @patch('users.handlers.identity.g') @patch('users.handlers.identity.Neo4jSession.__enter__') @patch('users.handlers.identity.Neo4jSession.__exit__') def test_get_identity_by_id_no_warning_same_identity( neo4j_exit, neo4j_enter, mock_g, mock_profiles, fixture_client, app_context ): """Test no warning is logged when caller and requested identity match.""" mock_g.request_context.context_type = constants.PROFILE_CONTEXT_TYPE mock_g.request_context.profile_type = 'LabelProfile' mock_g.request_context.jwt_identity_id = 'same-uuid' mock_profiles.get_identity_from_graph_tx.return_value = response.Response({'id': 'same-uuid'}) resp = fixture_client.get('/users/identity/same-uuid') assert resp.status_code == 200 mock_g.log.warning.assert_not_called()