"""Test logic for client.""" from copy import deepcopy from importlib import reload import json import re from unittest.mock import MagicMock, patch from auth0.v3 import Auth0Error import boto3 from freezegun import freeze_time from moto import mock_sqs from owsresponse import response, status import pytest from users import app, config, constants from users.connectors import redis from users.connectors.sentry import sentry_client from users.logic import auth0_client, user_info as user_info_logic from users.models import identities, user_info def test_gen_password(): """Test generate password will always generate valid pass.""" for _ in range(50): password = auth0_client.gen_password() assert len(password) >= 12 assert re.search(r'[a-z]', password) assert re.search(r'[a-z]', password) assert re.search(r'\d', password) assert not re.search(r'((\w)\2)', password) def test_send_password_reset(mocker): """Test send password reset.""" db_mock = MagicMock() db_mock.change_password.return_value = 'Sent!' mocker.patch('users.logic.auth0_client.Database', return_value=db_mock) actual = auth0_client.send_password_reset('larmstead@theorchard.com', 'auth-connection') assert actual == 'Sent!' db_mock.change_password.assert_called_once_with( config.AUTH0_MACHINE_CLIENT_ID, 'larmstead@theorchard.com', 'auth-connection' ) def test_send_password_reset_different_client_id(mocker): """Test send password reset.""" db_mock = MagicMock() mocker.patch('users.logic.auth0_client.Database', return_value=db_mock) auth0_client.send_password_reset('larmstead@theorchard.com', 'auth-connection', 'other') db_mock.change_password.assert_called_once_with( 'other', 'larmstead@theorchard.com', 'auth-connection' ) @pytest.mark.parametrize( ('email', 'user_id'), [ ('test@theorchard.com', 'auth0|5e58f645fceb0b0wke836d75'), ('test1@mail.com', 'auth0|5e5ekil45fceb0b0c26836d75'), ], ) def test_get_auth0_users(mocker, email, user_id): """Test get_auth0_users with some match.""" connection = config.AUTH0_CONNECTION expected = {'user_id': user_id, 'identities': [{'connection': connection}]} token_mock = MagicMock() token_mock.client_credentials.return_value = {'access_token': 'foo'} mocker.patch('users.logic.auth0_client.GetToken', return_value=token_mock) user_mock = MagicMock() user_mock.search_users_by_email.return_value = [expected] mocker.patch('users.logic.auth0_client.UsersByEmail', return_value=user_mock) actual = auth0_client.get_auth0_users(email) assert actual assert actual.message == {'result': [user_id]} @pytest.mark.parametrize( ('status', 'error_code', 'message'), [ (500, 505, 'Boom!'), (401, 'access_denied', 'Unauthorized'), (403, 'access_denied', 'Service not enabled within domain.'), ], ) def test_get_auth0_users_exception(mocker, status, error_code, message): """Test get_auth0_users when Auth0 throws an exception.""" email = 'test@theorchard.com' auth_err = Auth0Error(status, error_code, message) token_mock = MagicMock() token_mock.client_credentials.side_effect = auth_err mocker.patch('users.logic.auth0_client.GetToken', return_value=token_mock) with pytest.raises(Auth0Error): auth0_client.get_auth0_users(email) @pytest.mark.parametrize( ('email', 'connection', 'expected'), [ ( 'test@mock.com', 'google-oauth2', [ { 'user_id': '123mockid', 'identities': [ {'connection': 'google-oauth2', 'access_token': 'someaccesstoken'} ], }, { 'user_id': '123mockid', 'identities': [ {'connection': 'theorchard-gsuite', 'access_token': 'someaccesstoken'} ], }, ], ), ( 'test@mock.com', 'theorchard-gsuite', [ { 'user_id': '123mockid', 'identities': [ {'connection': 'theorchard-gsuite', 'access_token': 'someaccesstoken'} ], } ], ), ('test@mock.com', 'art-relations', []), ], ) def test_get_auth0_users_not_found(context, mocker, email, connection, expected): """Test get_auth0_users when auth0 user not found.""" token_mock = MagicMock() token_mock.client_credentials.return_value = {'access_token': 'foo'} mocker.patch('users.logic.auth0_client.GetToken', return_value=token_mock) user_mock = MagicMock() user_mock.search_users_by_email.return_value = expected mocker.patch('users.logic.auth0_client.UsersByEmail', return_value=user_mock) with context: mocker.spy(context.g.ows.log, 'warning') actual = auth0_client.get_auth0_users(email) assert actual.status == 200 assert actual.message['result'] == [] context.g.ows.log.warning.assert_called_with( constants.WARNING_MESSAGE_EMPTY_RESPONSE.format('get_auth0_users') ) @pytest.mark.parametrize( ('result', 'expected'), [ ( [ { 'email_verified': True, 'identities': [ { 'user_id': '123mock', 'provider': 'auth0', 'connection': 'art-relations', 'isSocial': False, } ], 'user_id': 'authID', } ], ['authID'], ), ( [ { 'email_verified': True, 'identities': [ { 'user_id': '123test', 'provider': 'auth0', 'connection': 'art-relations', 'isSocial': False, } ], 'user_id': 'mockID-123test', }, { 'email_verified': True, 'identities': [ { 'user_id': '123test', 'provider': 'google-apps', 'connection': 'not-relations', 'isSocial': False, } ], 'user_id': 'mockID-123test', }, ], ['mockID-123test'], ), ( [ { 'email_verified': True, 'identities': [ { 'user_id': '123test', 'provider': 'auth0', 'connection': 'art-relations', 'isSocial': False, } ], 'user_id': 'mockID', }, { 'email_verified': True, 'identities': [ { 'user_id': '123test', 'access_token': 'token', 'provider': 'auth0', 'connection': 'art-relations', 'isSocial': False, } ], 'user_id': 'mockID', }, ], ['mockID', 'mockID'], ), ], ) def test_get_auth0_users_mixed_connections(mocker, result, expected): """Test get_auth0_users with varying connections.""" token_mock = MagicMock() token_mock.client_credentials.return_value = {'access_token': 'foo'} mocker.patch('users.logic.auth0_client.GetToken', return_value=token_mock) user_mock = MagicMock() user_mock.search_users_by_email.return_value = result mocker.patch('users.logic.auth0_client.UsersByEmail', return_value=user_mock) actual = auth0_client.get_auth0_users('test@test.com') assert actual.message['result'] == expected def test_get_auth0_users_paginated(mocker): """Test get_auth0_users_paginated.""" token_mock = MagicMock() token_mock.client_credentials.return_value = {'access_token': 'foo'} mocker.patch('users.logic.auth0_client.GetToken', return_value=token_mock) users_mock = MagicMock() users_mock.list.return_value = {'users': [{'pizza': 'party'}]} mocker.patch('users.logic.auth0_client.Users', return_value=users_mock) result = auth0_client.get_auth0_users_paginated( page=0, per_page=100, q=('last_login:[* TO 2019-06-13]' ' AND (blocked:false OR NOT _exists_:blocked)'), ) assert users_mock.list.call_args[1] == { 'per_page': 100, 'search_engine': 'v3', 'q': ('last_login:[* TO 2019-06-13]' ' AND (blocked:false OR NOT _exists_:blocked)'), 'page': 0, } assert result.message == [{'pizza': 'party'}] def test_get_auth0_users_paginated_exception(mocker): """Test get_auth0_users_paginated when exception thrown.""" token_mock = MagicMock() token_mock.client_credentials.return_value = {'access_token': 'foo'} mocker.patch('users.logic.auth0_client.GetToken', return_value=token_mock) users_mock = MagicMock() users_mock.list.side_effect = Auth0Error( status_code=429, error_code='auth0_limit_reached', message='Auth0Error: 429: Global limit has been reached', ) mocker.patch('users.logic.auth0_client.Users', return_value=users_mock) mocker.patch('users.connectors.sentry.sentry_client.capture_exception') result = auth0_client.get_auth0_users_paginated( page=0, per_page=100, q=('last_login:[* TO 2019-06-13]' ' AND (blocked:false OR NOT _exists_:blocked)'), ) assert users_mock.list.call_args[1] == { 'per_page': 100, 'search_engine': 'v3', 'q': ('last_login:[* TO 2019-06-13]' ' AND (blocked:false OR NOT _exists_:blocked)'), 'page': 0, } assert sentry_client.capture_exception.called assert result.status == 429 assert result.errors == { 'code': 'auth0_limit_reached', 'message': 'Auth0Error: 429: Global limit has been reached', } def test_get_bulk_auth0_users(mocker): """Test get_bulk_auth0_users.""" token_mock = MagicMock() token_mock.client_credentials.return_value = {'access_token': 'foo'} mocker.patch('users.logic.auth0_client.GetToken', return_value=token_mock) users_mock = MagicMock() users_mock.list.return_value = {'users': [{'pizza': 'party'}]} mocker.patch('users.logic.auth0_client.Users', return_value=users_mock) result = auth0_client.get_bulk_auth0_users_paginated(['123', '456']) assert users_mock.list.call_args[1] == { 'page': 0, 'per_page': 100, 'q': 'user_id:auth0|123 OR user_id:auth0|456', 'search_engine': 'v3', } assert result.message == [{'pizza': 'party'}] @mock_sqs def test_send_sqs_message(): """Test send_sqs_message.""" # create a test queue to send message to. reload(auth0_client) queue_name = 'test-ows-notification' client = boto3.client('sqs', region_name='us-east-1') client.create_queue(QueueName=queue_name) queue_url = client.get_queue_url(QueueName=queue_name)['QueueUrl'] config.DAEMON_NOTIFICATIONS_SQS_URL = queue_url user_id = 122234 email = 'test@theorchard.com' auth0_id = 'auth0|adsasdasdas' labels = ['test label'] actual = auth0_client.send_sqs_message(user_id, email, auth0_id, labels) assert actual assert 'request' in actual.message assert 'response' in actual.message # Validate Message sqs_obj = boto3.resource('sqs', region_name='us-east-1') sqs_queue = sqs_obj.get_queue_by_name(QueueName=queue_name) sqs_msgs = sqs_queue.receive_messages( AttributeNames=['All'], MessageAttributeNames=['All'], VisibilityTimeout=15, WaitTimeSeconds=20, MaxNumberOfMessages=5, ) assert len(sqs_msgs) == 1 actuals = json.loads(sqs_msgs[0].body) assert actuals == { 'user_ids': [user_id], 'users_info': [ { 'user_id': user_id, 'auth0_id': auth0_id, 'email': email, } ], 'feed_name': constants.SSO_NOTIFICATION_FEED_NAME, 'feed_id': 'sso_{}'.format(auth0_id), 'template': constants.SSO_NOTIFICATION_TEMPLATE, 'payload': {'label_names': labels}, } def test_resend_verify_email(mocker): """Test resend_verify_email success.""" user_id = 'auth0|dummy' mgmt_mock = MagicMock() mgmt_mock.jobs.send_verification_email.return_value = 'Sent!' mocker.patch('users.logic.auth0_client.get_auth0_management_handle', return_value=mgmt_mock) actuals = auth0_client.resend_verify_email(user_id) assert actuals assert actuals.message == 'Sent!' def test_resend_verify_email_error(mocker, context): """Test resend_verify_email when it raises error.""" user_id = 'dummy' expected_error = Auth0Error(500, 'user_error', 'The user does not exist') mgmt_mock = MagicMock() mgmt_mock.jobs.send_verification_email.side_effect = expected_error mocker.patch('users.logic.auth0_client.get_auth0_management_handle', return_value=mgmt_mock) with context, app.app.test_request_context(): mocker.spy(context.g.ows.log, 'error') actuals = auth0_client.resend_verify_email(user_id) assert actuals.status == 500 assert actuals.errors.get('message') == expected_error.message def test_get_guardian_enrollments_for_user(mocker): """Test get_guardian_enrollments_for_user.""" auth0_id = 'auth0|acbd' token_mock = MagicMock() token_mock.client_credentials.return_value = {'access_token': 'foo'} mocker.patch('users.logic.auth0_client.GetToken', return_value=token_mock) users_mock = MagicMock() mock_response = [{'id': 'abc123', 'name': 'Brickphone'}] users_mock.get_guardian_enrollments.return_value = mock_response mocker.patch('users.logic.auth0_client.Users', return_value=users_mock) result = auth0_client.get_guardian_enrollments_for_user(auth0_id) assert users_mock.get_guardian_enrollments.call_args[0] == (auth0_id,) assert result.message == mock_response @patch('users.logic.auth0_client.g') @patch('users.logic.auth0_client.Users') @patch('users.logic.auth0_client.GetToken') def test_get_guardian_enrollments_for_user_error( get_token_mock: MagicMock, users_mock: MagicMock, g_mock: MagicMock, app_context ) -> None: """Test get_guardian_enrollments_for_user behavior when an auth0 error is encountered.""" get_token_mock.return_value.client_credentials.return_value = {'access_token': 'foo'} error_message = 'User does not exist.' users_mock.return_value.get_guardian_enrollments.side_effect = Auth0Error( status_code=404, error_code='error_code', message=error_message ) auth0_user_id = 'auth0|acbd' result = auth0_client.get_guardian_enrollments_for_user(auth0_user_id) assert result.status == 404 assert result.message == error_message g_mock.log.error.assert_called_once_with( 'Auth0 error getting guardian enrollments for user', resources={'error': error_message, 'auth0_user_id': auth0_user_id}, ) def test_reset_mfa_devices_for_user(mocker): """Test reset_mfa_devices_for_user.""" auth0_id = 'auth0|acbd' token_mock = MagicMock() token_mock.client_credentials.return_value = {'access_token': 'foo'} mocker.patch('users.logic.auth0_client.GetToken', return_value=token_mock) guardian_mock = MagicMock() device_id = 'abc123' mock_response = [{'id': device_id, 'name': 'Brickphone'}] guardian_mock.delete_enrollment.return_value = mock_response mocker.patch('users.logic.auth0_client.Guardian', return_value=guardian_mock) mocker.patch( 'users.logic.auth0_client.get_guardian_enrollments_for_user', return_value=response.Response(mock_response), ) result = auth0_client.reset_mfa_devices_for_user(auth0_id) assert guardian_mock.delete_enrollment.call_args[0] == (device_id,) assert result.message == {'deleted_devices': [device_id]} @pytest.mark.parametrize( ('users', 'search_result'), [ (['auth0|foouser'], [{'email': 'foo'}]), (['invalid'], []), (['auth0|foouser', 'dummy'], [{'email': 'foo'}]), ], ) def test_bulk_password_reset(users, search_result, mocker): """Test bulk_password_reset success.""" mgmt_mock = MagicMock() mgmt_mock.users.list.return_value = {'users': search_result} mocker.patch('users.logic.auth0_client.get_auth0_management_handle', return_value=mgmt_mock) db_mock = MagicMock() db_mock.change_password.return_value = 'Sent!' mocker.patch('users.logic.auth0_client.Database', return_value=db_mock) actuals = auth0_client.bulk_password_reset(users) expected = deepcopy(search_result) for user in expected: user['result'] = 'Sent!' assert actuals assert mgmt_mock.users.list.call_count == 1 if actuals.status == 200: assert actuals.message == {'users': expected} assert db_mock.change_password.call_count == len(search_result) else: assert actuals.message == {'message': 'No users found with these ids.'} assert db_mock.change_password.call_count == 0 def test_create_user_sends_email_to_different_client(mocker): """Test bulk_password_reset success.""" mgmt_mock = MagicMock() create_mock = MagicMock() mgmt_mock.users.list.return_value = {'users': create_mock} mocker.patch('users.logic.auth0_client.get_auth0_management_handle', return_value=mgmt_mock) db_mock = MagicMock() mocker.patch('users.logic.auth0_client.Database', return_value=db_mock) auth0_client.create_user({'email_client_id': 'dang', 'email': 'larmstead@theorchard.com'}) db_mock.change_password.assert_called_once_with( 'dang', 'larmstead@theorchard.com', 'art-relations' ) def test_create_user_sends_email_to_default_client(mocker): """Test bulk_password_reset success.""" mgmt_mock = MagicMock() create_mock = MagicMock() mgmt_mock.users.list.return_value = {'users': create_mock} mocker.patch('users.logic.auth0_client.get_auth0_management_handle', return_value=mgmt_mock) db_mock = MagicMock() mocker.patch('users.logic.auth0_client.Database', return_value=db_mock) auth0_client.create_user({'email': 'larmstead@theorchard.com'}) db_mock.change_password.assert_called_once_with( config.AUTH0_MACHINE_CLIENT_ID, 'larmstead@theorchard.com', 'art-relations' ) @pytest.mark.parametrize( ('side_effect', 'expected', 'raise_exception'), [ [ Auth0Error(409, 'error_code', 'message'), response.create_error_response( constants.ERROR_CODE_ALREADY_EXISTS, 'User already exists with this email.' ), False, ], [Auth0Error(405, 'other_error', 'other message'), None, True], [Exception('other exception'), None, True], ], ) def test_create_user_error(mocker, side_effect, expected, raise_exception): """Test bulk_password_reset success.""" mgmt_mock = MagicMock() mgmt_mock.users.create.side_effect = side_effect mocker.patch('users.logic.auth0_client.get_auth0_management_handle', return_value=mgmt_mock) db_mock = MagicMock() mocker.patch('users.logic.auth0_client.Database', return_value=db_mock) error = False try: actual = auth0_client.create_user({'email': 'test@theorchard.com'}) assert db_mock.change_password.call_count == 0 assert not actual assert actual.status == expected.status assert actual.errors == expected.errors except Exception: error = True assert error == raise_exception @pytest.mark.parametrize( 'data', [ ({'email': 'test@theorchard.com'}), ({'email': 'test@theorchard.com', 'user_metadata': {'username': 'test@theorchard.com'}}), ], ) def test_create_user_includes_audit_metadata(mocker, data): """Test create_user pass audit_user with user_metadata.""" mgmt_mock = MagicMock() create_mock = MagicMock() mgmt_mock.users.create = create_mock mocker.patch('users.logic.auth0_client.get_auth0_management_handle', return_value=mgmt_mock) db_mock = MagicMock() mocker.patch('users.logic.auth0_client.Database', return_value=db_mock) audit_user = 'test-audit-user-identity-id' auth0_client.create_user(data, audit_user) call_data = create_mock.mock_calls[0][1][0] assert 'audit_user' in call_data['user_metadata'] assert call_data['user_metadata']['audit_user'] == audit_user @freeze_time('2020-05-05') @pytest.mark.parametrize( ('data', 'update_user_args', 'neo4j_called', 'update_data'), [ ( {'name': 'new name'}, {'connection': 'art-relations', 'name': 'new name'}, True, { 'name': 'new name', 'first_name': 'new', 'last_name': 'name', 'email': 'email@theorchard.test', }, ), ( {'blocked': True}, { 'connection': 'art-relations', 'blocked': True, 'user_metadata': {'blocked_at': '2020-05-05T00:00:00'}, }, False, None, ), ( {'blocked': False}, { 'connection': 'art-relations', 'blocked': False, 'user_metadata': {'blocked_at': None}, }, False, None, ), ( {'email': 'newemail@theorchard.test'}, { 'connection': 'art-relations', 'email': 'newemail@theorchard.test', 'email_verified': True, }, True, { 'email': 'newemail@theorchard.test', 'name': 'new name', 'first_name': 'new', 'last_name': 'name', }, ), ], ) def test_update_user(mocker, data, update_user_args, neo4j_called, update_data): """Test update_user.""" user_id = 'auth0|5b83678aad451875aad4aaef' auth0_response = { 'name': 'new name', 'email': 'email@theorchard.test', 'email_verified': True, 'user_metadata': { 'orchardIdentityId': 'abcd-234', 'vend_contact_id': '82084', 'type': 'alw', 'first_name': 'new', 'last_name': 'name', }, } vend_contact_row = {'user_id': 'alw:1234', 'account': {'vendor_id': 4444}} primary = response.Response(message=vend_contact_row) vend_contact_user = response.Response(message={'foo': 'bar'}) mgmt_mock = MagicMock() mgmt_mock.users.update.return_value = auth0_response mocker.patch('users.logic.auth0_client.get_auth0_management_handle', return_value=mgmt_mock) mocker.patch.object(user_info, 'fetch_primary_for_auth0_user', return_value=primary) mocker.patch.object(user_info_logic, 'update_vend_contact_user', return_value=vend_contact_user) mocker.patch.object( identities, 'update_identity', return_value=response.Response(), autospec=True ) actual = auth0_client.update_user(user_id, data) assert actual assert actual.message == auth0_response mgmt_mock.users.update.assert_called_once_with(user_id, update_user_args) if neo4j_called: identities.update_identity.assert_called_once_with( auth0_response['user_metadata']['orchardIdentityId'], update_data ) @freeze_time('2020-05-05') @pytest.mark.parametrize( ('data', 'update_user_args', 'neo4j_called', 'update_data'), [ ( {'name': 'new name'}, {'connection': 'art-relations', 'name': 'new name'}, True, { 'name': 'new name', 'first_name': 'new', 'last_name': 'name', 'email': 'email@theorchard.test', }, ), ( {'blocked': True}, { 'connection': 'art-relations', 'blocked': True, 'user_metadata': {'blocked_at': '2020-05-05T00:00:00'}, }, False, None, ), ( {'blocked': False}, { 'connection': 'art-relations', 'blocked': False, 'user_metadata': {'blocked_at': None}, }, False, None, ), ( {'email': 'newemail@theorchard.test'}, { 'connection': 'art-relations', 'email': 'newemail@theorchard.test', 'email_verified': True, }, True, { 'email': 'newemail@theorchard.test', 'name': 'new name', 'first_name': 'new', 'last_name': 'name', }, ), ], ) def test_update_user_without_primary(mocker, data, update_user_args, neo4j_called, update_data): """Test update_user.""" user_id = 'auth0|5b83678aad451875aad4aaef' auth0_response = { 'name': 'new name', 'email': 'email@theorchard.test', 'email_verified': True, 'user_metadata': { 'orchardIdentityId': 'abcd-234', 'vend_contact_id': '82084', 'type': 'alw', 'first_name': 'new', 'last_name': 'name', }, } primary = response.Response(status=status.NOT_FOUND) mgmt_mock = MagicMock() mgmt_mock.users.update.return_value = auth0_response mocker.patch('users.logic.auth0_client.get_auth0_management_handle', return_value=mgmt_mock) mocker.patch.object(user_info, 'fetch_primary_for_auth0_user', return_value=primary) mocker.patch.object(user_info_logic, 'update_vend_contact_user', return_value=None) mocker.patch.object( identities, 'update_identity', return_value=response.Response(), autospec=True ) actual = auth0_client.update_user(user_id, data) assert actual assert actual.message == auth0_response mgmt_mock.users.update.assert_called_once_with(user_id, update_user_args) if neo4j_called: identities.update_identity.assert_called_once_with( auth0_response['user_metadata']['orchardIdentityId'], update_data ) user_info_logic.update_vend_contact_user.assert_not_called() @pytest.mark.parametrize( ('params', 'auth0_response', 'expected'), [ ( {'name': 'settings-login', 'app_type': 'spa'}, [ {'name': 'settings-login', 'client_id': '1234'}, {'name': 'insights-login', 'client_id': '5678'}, ], {'name': 'settings-login', 'client_id': '1234'}, ), ( {'name': 'insights-login', 'app_type': 'spa'}, [ {'name': 'settings-login', 'client_id': '1234'}, {'name': 'insights-login', 'client_id': '5678'}, ], {'name': 'insights-login', 'client_id': '5678'}, ), ( {'name': 'podcast-login', 'app_type': 'spa'}, [ {'name': 'settings-login', 'client_id': '1234'}, {'name': 'insights-login', 'client_id': '5678'}, ], None, ), ( {'name': 'workstation-login', 'app_type': 'regular_web'}, [ {'name': 'settings-login', 'client_id': '1234'}, {'name': 'workstation-login', 'client_id': '5678'}, ], {'name': 'workstation-login', 'client_id': '5678'}, ), ( {'name': 'songwhip-login', 'app_type': 'regular_web'}, [ {'name': 'settings-login', 'client_id': '1234'}, {'name': 'songwhip-login', 'client_id': '5678'}, ], {'name': 'songwhip-login', 'client_id': '5678'}, ), ], ) def test_get_application_by_name(mocker, params, auth0_response, expected): """Test _get_application_by_name.""" mocker.patch.object(redis.client, 'get', return_value=[]) mgmt_mock = MagicMock() mgmt_mock.clients.all.return_value = auth0_response mocker.patch('users.logic.auth0_client.get_auth0_management_handle', return_value=mgmt_mock) actual = auth0_client._get_application_by_name(params['name']) assert actual == expected if expected: mgmt_mock.clients.all.assert_called_once_with( fields=['client_id', 'name'], extra_params=params ) @pytest.mark.parametrize( ('connection_name', 'auth0_response', 'expected'), [ ( 'art-relations', [ {'name': 'art-relations', 'id': 'abcd'}, {'name': 'gsuite', 'client_id': 'xyz'}, ], {'name': 'art-relations', 'id': 'abcd'}, ), ( 'gsuite', [ {'name': 'art-relations', 'id': 'abcd'}, {'name': 'gsuite', 'client_id': 'xyz'}, ], {'name': 'gsuite', 'client_id': 'xyz'}, ), ( 'new-conn', [ {'name': 'art-relations', 'id': 'abcd'}, {'name': 'gsuite', 'client_id': 'xyz'}, ], None, ), ], ) def test_get_connection_by_name(mocker, connection_name, auth0_response, expected): """Test _get_connection_by_name.""" mgmt_mock = MagicMock() mgmt_mock.connections.all.return_value = auth0_response mocker.patch('users.logic.auth0_client.get_auth0_management_handle', return_value=mgmt_mock) actual = auth0_client._get_connection_by_name(connection_name) assert actual == expected @pytest.mark.parametrize( ( 'data', 'org_response', 'app_response', 'conn_response', 'expected_invitation_id', 'app_access_response', ), [ ( { 'brand': 'awal', 'email': 'user@theorchard.com', 'admin_name': 'Insights Admin', 'auth0_application_name': 'insights-login', 'user_metadata': {'orchardIdentityId': 'uuid'}, }, {'name': 'awal', 'id': 'org_awal_123'}, {'name': 'insights-login', 'client_id': '5678'}, {'name': 'theorchard-gsuite', 'id': 'gsuite123'}, 'uinv_111', response.Response( message={ 'items': [ {'name': 'Settings', 'id': 'settings'}, {'name': 'Insights', 'id': 'insights'}, ] } ), ), ( { 'brand': 'sme', 'email': 'user@sonymusic.com', 'admin_name': 'Insights Admin', 'auth0_application_name': 'insights-login', 'user_metadata': {'orchardIdentityId': 'uuid'}, }, {'name': 'sme', 'id': 'org_sme_123'}, {'name': 'insights-login', 'client_id': '5678'}, {'name': 'sme-prod-orchadmin', 'id': 'con123'}, 'uinv_222', response.Response( message={ 'items': [ {'name': 'Settings', 'id': 'settings'}, ] } ), ), ( { 'brand': 'nonexistent-brand', 'email': 'user@theorchard.com', 'auth0_application_name': 'insights-login', }, {'name': 'orchard', 'id': 'org_orchard_456'}, {'name': 'insights-login', 'client_id': '5678'}, {'name': 'theorchard-gsuite', 'id': 'gsuite123'}, 'uinv_333', response.Response( message={ 'items': [ {'name': 'Settings', 'id': 'settings'}, ] } ), ), ( { 'brand': 'awal', 'email': 'user@theorchard.com', 'auth0_application_name': 'insights-login', }, {'name': 'awal', 'id': 'org_awal_999'}, {'name': 'insights-login', 'client_id': '5678'}, {'name': 'theorchard-gsuite', 'id': 'gsuite123'}, 'uinv_444', response.Response(message={'items': []}), ), ( { 'brand': 'awal', 'email': 'user@theorchard.com', 'auth0_application_name': 'insights-login', 'user_metadata': {'orchardIdentityId': 'uuid', 'use_new_unified_template': False}, }, {'name': 'awal', 'id': 'org_awal_555'}, {'name': 'insights-login', 'client_id': '5678'}, {'name': 'theorchard-gsuite', 'id': 'gsuite123'}, 'uinv_555', response.Response( message={ 'items': [ {'name': 'Settings', 'id': 'settings'}, {'name': 'Insights', 'id': 'insights'}, ] } ), ), ], ) @patch('users.logic.auth0_client.g') def test_create_organization_invitation_success( mock_g, mocker, data, org_response, app_response, conn_response, expected_invitation_id, app_access_response, app_context, ): mgmt_mock = MagicMock() mgmt_mock.organizations.get_organization_by_name.return_value = org_response mgmt_mock.organizations.create_organization_invitation.return_value = { 'invitation': 'sent', 'id': expected_invitation_id, } mocker.patch('users.logic.auth0_client.get_auth0_management_handle', return_value=mgmt_mock) mocker.patch('users.logic.auth0_client._get_application_by_name', return_value=app_response) mocker.patch('users.logic.auth0_client._get_connection_by_name', return_value=conn_response) mocker.patch( 'users.models.identities.get_identity', return_value=response.Response( message={'first_name': 'Test', 'last_name': 'User', 'id': 'orch123'} ), ) mocker.patch( 'users.models.identities.get_identity_by_email', return_value=response.Response(message={}) ) mocker.patch( 'users.logic.profiles.get_applications_for_identity_tx', return_value=app_access_response, ) update_identity_mock = mocker.patch( 'users.models.identities.update_identity_organization_invitation', return_value=org_response, ) admin_identity_id = 'test_admin_123' actual = auth0_client.create_organization_invitation(data, admin_identity_id) assert actual.status == 200 assert actual.message['invitation'] == 'sent' assert actual.message['id'] == expected_invitation_id assert actual.message['identity']['id'] == org_response['id'] mgmt_mock.organizations.create_organization_invitation.assert_called_once() update_identity_mock.assert_called_once_with( email=data['email'], invitation_id=expected_invitation_id, organization_id=org_response['id'], organization_name=org_response['name'], admin_identity_id=admin_identity_id, ) expected_app_access_items = app_access_response.message['items'] sent_payload = mgmt_mock.organizations.create_organization_invitation.call_args[0][1] actual_app_access_items = sent_payload['user_metadata']['app_access'] actual_new_unified_tpl = sent_payload['user_metadata'].get('use_new_unified_template', None) assert actual_app_access_items == expected_app_access_items # Should honor existing value if provided, otherwise default to True expected_template_value = data.get('user_metadata', {}).get('use_new_unified_template', True) assert actual_new_unified_tpl == expected_template_value @pytest.mark.parametrize( ( 'data', 'org_response', 'app_response', 'conn_response', 'expected_error_message', 'app_access_response', ), [ ( { 'brand': 'awal', 'email': 'user@theorchard.com', 'auth0_application_name': 'invalid-app', }, {'name': 'awal', 'id': 'org_awal_123'}, None, {'name': 'theorchard-gsuite', 'id': 'gsuite123'}, 'Invalid Auth0 application: invalid-app', response.Response(message={}), ), ( { 'brand': 'awal', 'email': 'user@theorchard.com', 'auth0_application_name': 'insights-login', }, {'name': 'awal', 'id': 'org_awal_123'}, {'name': 'insights-login', 'client_id': '5678'}, None, 'Invalid Auth0 connection: theorchard-gsuite', response.Response(message={}), ), ( { 'brand': 'unknown', 'email': 'user@theorchard.com', 'auth0_application_name': 'insights-login', }, None, None, # won’t be reached None, # same 'Invalid brand name: orchard', response.Response(message={'items': []}), ), ], ) @patch('users.logic.auth0_client.g') def test_create_organization_invitation_failing( mock_g, mocker, data, org_response, app_response, conn_response, expected_error_message, app_access_response, app_context, ): mgmt_mock = MagicMock() mgmt_mock.organizations.get_organization_by_name.return_value = org_response mocker.patch('users.logic.auth0_client.get_auth0_management_handle', return_value=mgmt_mock) mocker.patch('users.logic.auth0_client._get_application_by_name', return_value=app_response) mocker.patch('users.logic.auth0_client.get_connection_id_from_name', return_value=conn_response) mocker.patch( 'users.models.identities.get_identity', return_value=response.Response(message={'first_name': 'Failing', 'last_name': 'Admin'}), ) mocker.patch( 'users.models.identities.get_identity_by_email', return_value=response.Response(message={}) ) mocker.patch( 'users.logic.profiles.get_applications_for_identity_tx', return_value=app_access_response, ) update_identity_mock = mocker.patch( 'users.models.identities.update_identity_organization_invitation' ) admin_identity_id = 'admin_error_case' actual = auth0_client.create_organization_invitation(data, admin_identity_id) assert actual.status == 500 assert actual.errors['code'] == 'internal_error' assert actual.errors['message'] == expected_error_message # None of these should happen in failing cases mgmt_mock.organizations.create_organization_invitation.assert_not_called() update_identity_mock.assert_not_called() def test_update_user_neo4j_failed(mocker): """Test update_user.""" user_id = 'auth0|5b83678aad451875aad4aaef' data = {'name': 'new name'} auth0_response = { 'name': 'new name', 'email': 'email@theorchard.test', 'user_metadata': { 'orchardIdentityId': 'abcd-234', 'vend_contact_id': '82084', 'type': 'alw', 'first_name': 'new', 'last_name': 'name', }, } vend_contact_row = {'user_id': 'alw:1234', 'account': {'vendor_id': 4444}} primary = response.Response(message=vend_contact_row) vend_contact_user = response.Response(message={'foo': 'bar'}) mgmt_mock = MagicMock() mgmt_mock.users.update.return_value = auth0_response mocker.patch('users.logic.auth0_client.get_auth0_management_handle', return_value=mgmt_mock) mocker.patch.object( identities, 'update_identity', return_value=response.create_error_response('some', 'error'), autospec=True, ) mocker.patch.object(user_info, 'fetch_primary_for_auth0_user', return_value=primary) mocker.patch.object(user_info_logic, 'update_vend_contact_user', return_value=vend_contact_user) actual = auth0_client.update_user(user_id, data) identities.update_identity.assert_called_once() # log error from neo4j update, no error in response assert actual assert actual.message == auth0_response @pytest.mark.parametrize( ('data', 'org_response', 'expected'), [ # all required fields, valid data ( {'brand': 'orchard', 'members': ['auth0|test1', 'auth0|test2']}, {'name': 'orchard', 'id': 'themass'}, response.Response(status=200, message='', errors=None), ), # test that brand is case sensitive at this point, invalid data ( {'brand': 'KNR', 'members': ['auth0|test1', 'auth0|test2']}, None, response.create_fatal_response('Organization/brand:KNR does not exist.'), ), # Brand is invalid. ( {'brand': 'dummy', 'members': ['auth0|test1']}, None, response.create_fatal_response('Organization/brand:dummy does not exist.'), ), # Atleast 1 member required ( {'brand': 'orchard', 'members': []}, {'name': 'orchard', 'id': 'themass'}, response.create_fatal_response( 'Members list cannot be empty. Atleast 1 auth0 user id required.' ), ), ], ) @patch('users.logic.auth0_client.g') def test_create_organization_members(mock_g, mocker, data, org_response, expected, app_context): """Test create_organization_invitation.""" admin_identity_id = 'testadminidentityid12345' mgmt_mock = MagicMock() mgmt_mock.organizations.create_organization_members.return_value = expected.message mgmt_mock.organizations.get_organization_by_name.return_value = org_response # mgmt_mock.auth mocker.patch('users.logic.auth0_client.get_auth0_management_handle', return_value=mgmt_mock) actual = auth0_client.create_organization_members(data, admin_identity_id) assert actual.status == expected.status assert actual.message == expected.message assert actual.errors == expected.errors def test_list_user_organizations(mocker): """Test list_user_organizations.""" expected = response.Response(message=['awal']) mgmt_mock = MagicMock() mgmt_mock.users.list_organizations.return_value = {'organizations': [{'name': 'awal'}]} mocker.patch('users.logic.auth0_client.get_auth0_management_handle', return_value=mgmt_mock) auth0_user_id = 'auth0|12345abcdef' actual = auth0_client.list_user_organizations(auth0_user_id) assert actual.status == expected.status assert actual.message == expected.message assert actual.errors == expected.errors def test_get_organization_info(mocker): """Test get_organization_info.""" expected = response.Response(message={'name': 'awal'}) mgmt_mock = MagicMock() mgmt_mock.organizations.get_organization_by_name.return_value = {'name': 'awal'} mocker.patch('users.logic.auth0_client.get_auth0_management_handle', return_value=mgmt_mock) org_name = 'awal' actual = auth0_client.get_organization(org_name) assert actual.status == expected.status assert actual.message == expected.message assert actual.errors == expected.errors