"""Test user_info logic. Tests for logic layer functions for user information retrieval. """ from datetime import datetime, timedelta, timezone from unittest.mock import call, MagicMock, patch import flask from freezegun import freeze_time from neo4j.time import DateTime as NeoDateTime from owsresponse import response import pytest from pythonfeatures import pythonfeatures from pythonfeatures.constants import context as context_constants from users import config, constants from users.app import app from users.logic import auth0_client, user_info from users.models import ( identities, ows_account as ows_account_model, social_auth_item, user_info as user_info_model, user_info_raw, ) @pytest.mark.parametrize( ('user_type', 'user_ids', 'model_fn', 'model_fn_params', 'expected_result'), [ (constants.USER_INFO_USER_TYPE_OA, None, 'fetch_oa_users', None, 'oa users'), (constants.USER_INFO_USER_TYPE_OA, [123], 'fetch_oa_users', [123], 'oa users'), (constants.USER_INFO_USER_TYPE_ALW, None, 'fetch_alw_users', None, 'alw users'), (constants.USER_INFO_USER_TYPE_ALW, [123], 'fetch_alw_users', [123], 'alw users'), ], ) def test_get_users(mocker, user_type, user_ids, model_fn, model_fn_params, expected_result): """Test successfully getting user information.""" mocker.patch.object( user_info_model, 'fetch_oa_users', return_value=response.Response('oa users'), autospec=True ) mocker.patch.object( user_info_model, 'fetch_alw_users', return_value=response.Response('alw users'), autospec=True, ) result = user_info.get_users(user_type, user_ids) getattr(user_info.user_info, model_fn).assert_called_with(model_fn_params) assert result assert result.message == expected_result def test_get_users_not_implemented(mocker): """Test getting user information for unknown user type.""" with pytest.raises(NotImplementedError): user_info.get_users('applesauce_bananas') @pytest.mark.parametrize( 'user_ids, mock_model_data, expected_formatted', [ pytest.param( ['1', '2'], [ {'user_id': '1', 'first_name': 'Test', 'last_name': 'User 1', 'active': True}, {'user_id': '2', 'first_name': 'Test', 'last_name': 'User 2', 'active': False}, ], { 'users': [ {'user_id': '1', 'first_name': 'Test', 'last_name': 'User 1', 'active': True}, {'user_id': '2', 'first_name': 'Test', 'last_name': 'User 2', 'active': False}, ] }, id='all users found', ), pytest.param( ['1', '2', '3'], [ {'user_id': '1', 'first_name': 'Test', 'last_name': 'User 1', 'active': True}, {'user_id': '3', 'first_name': 'Test', 'last_name': 'User 2', 'active': True}, ], { 'users': [ {'user_id': '1', 'first_name': 'Test', 'last_name': 'User 1', 'active': True}, None, {'user_id': '3', 'first_name': 'Test', 'last_name': 'User 2', 'active': True}, ] }, id='some users missing', ), pytest.param([], [], {'users': []}, id='empty input'), pytest.param( ['5', '3', '1'], [ {'user_id': '1', 'first_name': 'Test', 'last_name': 'user 1', 'active': True}, {'user_id': '3', 'first_name': 'Test', 'last_name': 'user 2', 'active': False}, {'user_id': '5', 'first_name': 'Test', 'last_name': 'user 3', 'active': True}, ], { 'users': [ {'user_id': '5', 'first_name': 'Test', 'last_name': 'user 3', 'active': True}, {'user_id': '3', 'first_name': 'Test', 'last_name': 'user 2', 'active': False}, {'user_id': '1', 'first_name': 'Test', 'last_name': 'user 1', 'active': True}, ] }, id='preserve order', ), ], ) @patch('users.models.user_info.get_orchadmin_users') def test_get_orchadmin_users_variants( mock_model_get_orchadmin_users, user_ids, mock_model_data, expected_formatted ): """Test get_orchadmin_users returns expected user data for different input scenarios.""" mock_model_get_orchadmin_users.return_value = mock_model_data result = user_info.get_orchadmin_users(user_ids) assert result == expected_formatted mock_model_get_orchadmin_users.assert_called_once_with(user_ids) @pytest.mark.parametrize( ('user_id', 'expected_result'), [ ('alw:123456', response.Response({'alw': 'user_details'})), ('oa:123456', response.Response({'oa': 'user_details'})), # this request has been in sentry's mysql error. So handle it gracefully. ('alw:0', response.create_not_found_response(message='User not found.')), ( 'bad:123', response.create_fatal_response( message='user_id must be in the form oa:123 or alw:123.' ), ), ( '123456', response.create_fatal_response( message='user_id must be in the form oa:123 or alw:123.' ), ), ], ) def test_get_user_raw(mocker, user_id, expected_result): """Test get_user_raw with different user_ids.""" mocker.patch.object( user_info_raw, 'fetch_oa_user_raw', return_value=response.Response({'oa': 'user_details'}) ) mocker.patch.object( user_info_raw, 'fetch_vend_contact_details', return_value=response.Response({'alw': 'user_details'}), ) user_result = user_info.get_user_raw(user_id) assert user_result.status == expected_result.status assert user_result.message == expected_result.message @pytest.mark.parametrize( ('user_type', 'user_id', 'expected_result'), [ ('alw', '123456', response.Response('alw_response')), ('oa', '123456', response.Response('oa_response')), ], ) def test_get_users_minimum_details(mocker, user_type, user_id, expected_result): """Test get_users_minimum_details with minimum_details=True.""" mocker.patch.object( user_info_raw, 'fetch_oa_user_raw', return_value=response.Response(message='oa_response') ) mocker.patch.object( user_info_raw, 'fetch_vend_contact_raw', return_value=response.Response(message='alw_response'), ) user_result = user_info.get_users_minimum_details(user_type, user_id, False) assert user_result.status == expected_result.status assert user_result.message == expected_result.message @pytest.mark.parametrize( ('user_type', 'user_id', 'expected_result'), [ ('alw', '123456', response.Response('alw_response')), ('oa', '123456', response.Response('oa_response')), ], ) def test_get_users_minimum_details_include_roles(mocker, user_type, user_id, expected_result): """Test get_users_minimum_details with minimum_details=True.""" mocker.patch.object( user_info_raw, 'fetch_oa_user_raw', return_value=response.Response(message='oa_response') ) mocker.patch.object( user_info_raw, 'fetch_vend_contact_raw_with_roles', return_value=response.Response(message='alw_response'), ) user_result = user_info.get_users_minimum_details(user_type, user_id, True) assert user_result.status == expected_result.status assert user_result.message == expected_result.message @pytest.mark.parametrize( ('user_id', 'app', 'model_response', 'expected_result'), [ # 200 ok, ALW ('4444', 'alw', response.Response('whatever'), response.Response('whatever')), # 404 user not found, ALW ( '99999999999', 'alw', response.create_not_found_response(), response.create_not_found_response(), ), # 501 OA not implemented ( '333', 'oa', None, response.create_error_response( status=501, code='NOT_IMPLEMENTED', message='Not implemented.' ), ), # 400 invalid app ( '333', 'pizza', None, response.create_error_response(status=400, code='INVALID_APP', message='Invalid app.'), ), ], ) def test_get_user_session_metadata_for_app(mocker, user_id, app, model_response, expected_result): """Test get_user_session_metadata_for_app.""" if model_response is not None: mocker.patch.object( user_info_model, 'fetch_alw_session_user_metatada', return_value=model_response, autospec=True, ) result = user_info.get_user_session_metadata_for_app(user_id, app) assert result.status == expected_result.status assert result.message == expected_result.message @pytest.mark.parametrize( ('user_id', 'app', 'model_response', 'expected_result'), [ # 200 ok, ALW ( '4444', 'alw', response.Response({'what': 'ever'}), response.Response({'what': 'ever'}), ), # 404 user not found, ALW ( '99999999999', 'alw', response.create_not_found_response(), response.create_not_found_response(), ), # 501 OA not implemented ( '333', 'oa', None, response.create_error_response( status=501, code='NOT_IMPLEMENTED', message='Not implemented.' ), ), # 400 invalid app ( '333', 'pizza', None, response.create_error_response(status=400, code='INVALID_APP', message='Invalid app.'), ), ], ) def test_get_user_session_metadata_for_app_raw( mocker, user_id, app, model_response, expected_result ): """Test get_user_session_metadata_for_app_raw.""" if model_response is not None: mocker.patch.object( user_info_model, 'fetch_alw_session_user_metatada_raw', return_value=model_response, autospec=True, ) result = user_info.get_user_session_metadata_for_app_raw(user_id, app) assert result.status == expected_result.status assert result.message == expected_result.message @pytest.mark.parametrize( ( 'model_response', 'feature_response', 'identity_response', 'identity_called', 'expected_result', ), [ # when FFlag is OFF ( response.Response({'what': 'ever', 'language': 'en'}), response.Response(message='disabled'), None, 0, response.Response({'what': 'ever', 'language': 'en'}), ), # when FFlag is ON but no auth0_user_id ( response.Response({'what': 'ever', 'language': 'en'}), response.Response(message='enabled'), None, 0, response.Response({'what': 'ever', 'language': 'en'}), ), # when FFlag is ON and has auth0_user_id, but there is no identity record. ( response.Response({'what': 'ever', 'language': 'en', 'auth0_user_id': 'abcdef'}), response.Response(message='enabled'), response.create_not_found_response(), 1, response.Response({'what': 'ever', 'language': 'en', 'auth0_user_id': 'abcdef'}), ), # when FFlag is ON, has auth0_user_id and identity record with same localization. ( response.Response({'what': 'ever', 'language': 'en', 'auth0_user_id': 'abcdef'}), response.Response(message='enabled'), response.Response({'neo4j': 'data', 'localization': 'en'}), 1, response.Response( {'what': 'ever', 'language': 'en', 'auth0_user_id': 'abcdef', 'number_format': None} ), ), # when FFlag is ON, has auth0_user_id and identity record with diff localization. ( response.Response({'what': 'ever', 'language': 'en', 'auth0_user_id': 'abcdef'}), response.Response(message='enabled'), response.Response({'neo4j': 'data', 'localization': 'fr'}), 1, response.Response( {'what': 'ever', 'language': 'fr', 'auth0_user_id': 'abcdef', 'number_format': None} ), ), ( response.Response({'language': 'en', 'auth0_user_id': 'abcdef', 'number_format': 'us'}), response.Response(message='enabled'), response.Response({'neo4j': 'data', 'localization': 'fr', 'number_format': 'eu'}), 1, response.Response( {'language': 'fr', 'auth0_user_id': 'abcdef', 'number_format': 'europe'} ), ), # when FFlag is ON, has auth0_user_id and identity record with no localization. # in case backfill did not happen, use AR level language. ( response.Response({'what': 'ever', 'language': 'en', 'auth0_user_id': 'abcdef'}), response.Response(message='enabled'), response.Response({'neo4j': 'data'}), 1, response.Response( {'what': 'ever', 'language': 'en', 'auth0_user_id': 'abcdef', 'number_format': None} ), ), ], ) def test_get_user_session_alw( model_response, feature_response, identity_response, identity_called, expected_result, mocker ): """Test get_user_session_metadata_for_app_raw for alw users.""" mocker.patch.object( user_info_model, 'fetch_alw_session_user_metatada_raw', return_value=model_response ) mocker.patch.object( pythonfeatures, 'get_single_feature_by_attributes', return_value=feature_response ) mocker.patch.object(identities, 'get_identity_by_auth0_id', return_value=identity_response) result = user_info.get_user_session_metadata_for_app_raw('1234', 'alw') assert result.status == expected_result.status assert result.message == expected_result.message if identity_called: assert identities.get_identity_by_auth0_id.call_count == 1 else: assert identities.get_identity_by_auth0_id.call_count == 0 @pytest.mark.parametrize( ('user_id', 'app', 'model_response', 'neo4j_response', 'expected_result'), [ # 200 ok, ALW ( '4444', 'alw', response.Response({'foo': 'bar'}), None, response.Response({'foo': 'bar', 'linked_accounts': []}), ), ( '5555', 'alw', response.Response( { 'language': 'en', 'auth0_user_id': 'abcd', 'number_format': 'us', 'vendor': {'name': 'vendor'}, } ), response.Response( { 'identity': {'auth0_user_id': 'abcd', 'various': 'props'}, 'service_tier': {'tier': 'tier1'}, 'company_brand': {'name': 'brand'}, } ), response.Response( { 'language': 'en', 'linked_accounts': [], 'auth0_user_id': 'abcd', 'number_format': 'us', 'identity': {'auth0_user_id': 'abcd', 'various': 'props'}, 'vendor': { 'name': 'vendor', 'service_tier': {'tier': 'tier1'}, 'company_brand': {'name': 'brand'}, }, } ), ), ( '6666', 'alw', response.Response( { 'language': 'en', 'vendor': {'name': 'vendor'}, 'auth0_user_id': 'abcd', 'number_format': 'us', } ), response.Response( { 'identity': {'auth0_user_id': 'abcd', 'various': 'props'}, 'service_tier': {'tier': 'tier1'}, 'company_brand': {'name': 'brand'}, } ), response.Response( { 'language': 'en', 'linked_accounts': [], 'auth0_user_id': 'abcd', 'number_format': 'us', 'identity': {'auth0_user_id': 'abcd', 'various': 'props'}, 'vendor': { 'name': 'vendor', 'service_tier': {'tier': 'tier1'}, 'company_brand': {'name': 'brand'}, }, } ), ), # Not all vendors have service_tier as of 3/22 ( '7777', 'alw', response.Response( { 'language': 'en', 'auth0_user_id': 'abcd', 'number_format': 'us', 'vendor': {'name': 'vendor'}, } ), response.Response( { 'identity': { 'auth0_user_id': 'abcd', 'localization': 'es', 'number_format': 'us', }, 'service_tier': None, 'company_brand': {'name': 'brand'}, } ), response.Response( { 'language': 'en', 'linked_accounts': [], 'auth0_user_id': 'abcd', 'number_format': 'us', 'identity': { 'auth0_user_id': 'abcd', 'localization': 'es', 'number_format': 'us', }, 'vendor': { 'name': 'vendor', 'service_tier': None, 'company_brand': {'name': 'brand'}, }, } ), ), # 404 user not found, ALW ( '99999999999', 'alw', response.create_not_found_response(), None, response.create_not_found_response(), ), # 404 identity not found neo4j will only return ar data ( '99999999999', 'alw', response.Response( { 'language': 'en', 'auth0_user_id': 'abcd', 'number_format': 'us', 'vendor': {'name': 'vendor'}, } ), response.create_not_found_response(), response.Response( { 'language': 'en', 'linked_accounts': [], 'auth0_user_id': 'abcd', 'number_format': 'us', 'vendor': {'name': 'vendor'}, } ), ), # 501 OA not implemented ( '333', 'oa', None, None, response.create_error_response( status=501, code='NOT_IMPLEMENTED', message='Not implemented.' ), ), # 400 invalid app ( '333', 'pizza', None, None, response.create_error_response(status=400, code='INVALID_APP', message='Invalid app.'), ), ], ) def test_get_account_session_data( mocker, user_id, app, model_response, neo4j_response, expected_result ): """Test get_account_session_data.""" if model_response is not None: mocker.patch.object( user_info_model, 'fetch_alw_session_user_metatada_raw', return_value=model_response ) mocker.patch.object(user_info, 'get_linked_accounts', return_value=response.Response([])) mocker.patch.object( identities, 'get_account_session_by_identity', return_value=neo4j_response ) result = user_info.get_account_session_data(user_id, app) assert result.status == expected_result.status assert result.message == expected_result.message def test_verify_alw_login(mocker): """Test verify_alw_login.""" mocker.patch.object( user_info_raw, 'fetch_alw_user_by_login', return_value=response.Response(message={'new': 'fn'}), ) result = user_info.verify_alw_login('login', 'pass') assert result assert result.message == {'new': 'fn'} @pytest.mark.parametrize(('user_type', 'status'), [('alw', 200), ('foo', 501)]) def test_update_auth0_details(mocker, user_type, status): """Test update_auth0_details.""" expected = response.Response(message={'foo': 'bar'}) primary = response.create_not_found_response() vc = response.Response(message={'auth0_user_id': None}) auth0_id = 'auth0132234' user_id = 123 now = datetime.now() mocker.patch.object(user_info_model, 'update_vend_contact_details', return_value=expected) mocker.patch.object(user_info_model, 'fetch_primary_for_auth0_user', return_value=primary) mocker.patch.object(user_info_model, 'fetch_vend_contact_and_contact', return_value=vc) expected_data = {'auth0_user_id': auth0_id, 'auth0_migration_date': now, 'auth0_primary': 'Y'} with patch('users.logic.user_info.datetime') as mock_date: mock_date.now.return_value = now result = user_info.update_auth0_details(user_id, user_type, auth0_id) assert result.status == status if status == 200: user_info_model.update_vend_contact_details.assert_called_with(user_id, expected_data) assert result == expected @pytest.mark.parametrize( ('include_deleted', 'include_support_email'), [ (None, None), (True, None), (False, None), (None, True), (None, False), (True, True), (False, False), (True, False), ], ) def test_get_accounts_with_auth0_id(include_deleted, include_support_email, mocker): """Test get_accounts_with_auth0_id.""" expected = {'foo': 'bar'} auth0_id = 'auth0132234' mocker.patch.object( user_info_model, 'get_all_accounts_with_auth0_id', return_value=response.Response(message=expected), ) mocker.patch.object( user_info_model, 'get_all_account_details_with_auth0_id', return_value=response.Response(message=expected), ) result = user_info.get_accounts_with_auth0_id(auth0_id, include_deleted, include_support_email) assert result if include_support_email: user_info_model.get_all_account_details_with_auth0_id.assert_called_with( auth0_id, include_deleted ) assert user_info_model.get_all_accounts_with_auth0_id.call_count == 0 else: user_info_model.get_all_accounts_with_auth0_id.assert_called_with(auth0_id, include_deleted) assert user_info_model.get_all_account_details_with_auth0_id.call_count == 0 assert result.message == expected @pytest.mark.parametrize( ('login', 'requested_email'), [(None, None), (None, 'aaa@test.com'), ('foo', None), ('foo', 'aaa@test.com')], ) def test_update_auth0_details_no_login(mocker, login, requested_email): """Test update_auth0_details for user with no login.""" expected = response.Response(message={'foo': 'bar'}) primary = response.create_not_found_response() vc = response.Response( message={ 'auth0_user_id': None, 'login': login, 'account': {'vendor_id': 7123}, 'requested_login_email': requested_email, } ) auth0_id = 'auth0132234' user_id = 123 now = datetime.now() mocker.patch.object(user_info_model, 'update_vend_contact_details', return_value=expected) mocker.patch.object(user_info_model, 'fetch_primary_for_auth0_user', return_value=primary) mocker.patch.object(user_info_model, 'fetch_vend_contact_and_contact', return_value=vc) expected_data = {'auth0_user_id': auth0_id, 'auth0_migration_date': now, 'auth0_primary': 'Y'} if not login and requested_email: expected_data.update({'login': '7123__aaa@test.com'}) with patch('users.logic.user_info.datetime') as mock_date: mock_date.now.return_value = now result = user_info.update_auth0_details(user_id, 'alw', auth0_id) assert result user_info_model.update_vend_contact_details.assert_called_with(user_id, expected_data) assert result == expected @pytest.mark.parametrize(('user_type', 'status'), [('alw', 200), ('foo', 501)]) def test_update_auth0_details_with_primary(mocker, user_type, status): """Test update_auth0_details when there is a primary account already.""" expected = response.Response(message={'foo': 'bar'}) primary = response.Response(message=[{'foo'}]) vc = response.Response(message={'auth0_user_id': None}) auth0_id = 'auth0132234' user_id = 123 now = datetime.now() mocker.patch.object(user_info_model, 'update_vend_contact_details', return_value=expected) mocker.patch.object(user_info_model, 'fetch_primary_for_auth0_user', return_value=primary) mocker.patch.object(user_info_model, 'fetch_vend_contact_and_contact', return_value=vc) expected_data = {'auth0_user_id': auth0_id, 'auth0_migration_date': now} with patch('users.logic.user_info.datetime') as mock_date: mock_date.now.return_value = now result = user_info.update_auth0_details(user_id, user_type, auth0_id) assert result.status == status if status == 200: user_info_model.update_vend_contact_details.assert_called_with(user_id, expected_data) assert result == expected @pytest.mark.parametrize( ('user_type', 'auth0_id', 'target_auth0_id', 'status'), [('alw', 'auth1', 'auth2', 200), ('foo', 'auth1', 'auth2', 501)], ) def test_update_auth0_details_migrate(mocker, user_type, auth0_id, target_auth0_id, status): """Test update_auth0_details for an existing Auth0 user.""" expected = response.Response(message={'foo': 'bar'}) primary = response.Response(message={'foo': 'bar'}) vc = response.Response(message={'auth0_user_id': target_auth0_id, 'primary': 'Y'}) user_id = 123 now = datetime.now() mocker.patch.object(user_info_model, 'update_vend_contact_details', return_value=expected) mocker.patch.object(user_info_model, 'fetch_primary_for_auth0_user', return_value=primary) mocker.patch.object(user_info_model, 'fetch_vend_contact_and_contact', return_value=vc) mocker.patch.object(user_info, 'cleanup_auth0_user', return_value=vc) expected_data = {'auth0_user_id': auth0_id, 'auth0_migration_date': now, 'auth0_primary': None} with patch('users.logic.user_info.datetime') as mock_date: mock_date.now.return_value = now result = user_info.update_auth0_details(user_id, user_type, auth0_id) assert result.status == status if status == 200: user_info_model.update_vend_contact_details.assert_called_with(user_id, expected_data) user_info.cleanup_auth0_user.assert_called_with(target_auth0_id) assert result == expected @pytest.mark.parametrize(('user_type', 'status'), [('alw', 200), ('foo', 501)]) def test_get_label_names(mocker, user_type, status): """Test get_label_names.""" expected = response.Response(message={'foo': 'bar'}) mocker.patch.object(user_info_model, 'fetch_vendor_names', return_value=expected) result = user_info.get_label_names(user_type, 12323) assert result.status == status if status == 200: user_info_model.fetch_vendor_names.assert_called_with(12323) assert result == expected else: user_info_model.fetch_vendor_names.asset_not_called() def test_get_label_names_for_auth0(mocker): """Test get_label_names_for_auth0.""" expected = response.Response(message={'foo': 'bar'}) mocker.patch.object(user_info_model, 'fetch_vendor_names', return_value=expected) result = user_info.get_label_names_for_auth0(12323) assert result.status == 200 user_info_model.fetch_vendor_names.assert_called_with(None, 12323) @pytest.mark.parametrize(('user_type', 'status'), [('alw', 200), ('oa', 501), ('foo', 400)]) def test_get_linked_accounts(mocker, user_type, status): """Test get_linked_accounts.""" vc_id = '12323' identity_id = 'identity-uuid-123' expected = response.Response( message=[ {'vc_id': 102, 'vendor_id': 8869, 'vendor_name': 'Label A'}, ] ) mocker.patch( 'users.logic.user_info.identities.get_identity_for_profile', return_value=response.Response(message={'id': identity_id}), ) mocker.patch( 'users.logic.user_info.profiles.get_linked_label_profiles', return_value=[ {'profile_id': 102, 'profile_type': 'LabelProfile', 'uuid': 'aaaa-bbbb-cccc-1111'}, ], ) mocker.patch.object( user_info_model, 'get_linked_account_details_by_profile_ids', return_value=expected ) result = user_info.get_linked_accounts(vc_id, user_type) assert result.status == status if status == 200: user_info_model.get_linked_account_details_by_profile_ids.assert_called_with(vc_id, [102]) assert result.message[0]['profile_uuid'] == 'aaaa-bbbb-cccc-1111' def test_get_linked_accounts_mysql_error_skips_enrichment(mocker): """Test that profile_uuid enrichment is skipped when MySQL returns an error.""" vc_id = '12323' identity_id = 'identity-uuid-123' mocker.patch( 'users.logic.user_info.identities.get_identity_for_profile', return_value=response.Response(message={'id': identity_id}), ) mocker.patch( 'users.logic.user_info.profiles.get_linked_label_profiles', return_value=[ {'profile_id': 102, 'profile_type': 'LabelProfile', 'uuid': 'aaaa-bbbb-cccc-1111'}, ], ) mocker.patch.object( user_info_model, 'get_linked_account_details_by_profile_ids', return_value=response.Response(status=500, message='internal error'), ) result = user_info.get_linked_accounts(vc_id, 'alw') assert result.status == 500 assert result.message == 'internal error' @pytest.mark.parametrize( ('minimum_details_response', 'expected_response'), [ ( # vendor id, and subaccount id response.Response( status=200, message={'account': {'vendor_id': 1234, 'subaccount_id': 1}} ), response.Response( message={ context_constants.ORCHARD_USER_ID: 'alw:whatever', context_constants.VENDOR_ID: 1234, context_constants.SUBACCOUNT_ID: 1, } ), ), ( # vendor id, and subaccount NULL response.Response( status=200, message={'account': {'vendor_id': 1234, 'subaccount_id': None}} ), response.Response( message={ context_constants.ORCHARD_USER_ID: 'alw:whatever', context_constants.VENDOR_ID: 1234, } ), ), ( # user not found response.create_not_found_response(message={'pizza': 'party'}), response.create_not_found_response(message={'pizza': 'party'}), ), ], ) def test_get_attributes_for_vend_contact(mocker, minimum_details_response, expected_response): """Test get_attributes_for_vend_contact().""" mocker.patch.object( user_info, 'get_users_minimum_details', return_value=minimum_details_response ) result = user_info.get_attributes_for_vend_contact('whatever') if result: assert result.message == expected_response.message else: assert result.errors == expected_response.errors def test_reset_users_auth0_details(mocker): """Test reset_users_auth0_details.""" auth0_id = 'testAuth0Id' expected = response.Response(message={'foo': 'bar'}) mocker.patch.object(user_info_model, 'reset_users_auth0_details', return_value=expected) result = user_info.reset_users_auth0_details(auth0_id) assert result.status == 200 user_info_model.reset_users_auth0_details.assert_called_with(auth0_id) @pytest.mark.parametrize( ('auth0_id', 'user_id', 'target_auth0_id', 'identity_id', 'primary_user_id', 'status'), [ ('auth0|0132234', 123, 'auth0|0132234', 'a-uuid', None, 200), ('auth0|0132234', 123, '0132234', 'a-uuid', None, 200), ('auth0|0132234', 123, None, None, None, 404), ('auth0|123', 123, 'auth0|0132234', 'a-uuid', None, 404), ('auth0|0132234', 123, 'auth0|0132234', 'a-uuid', 'alw:345', 200), ('auth0|0132234', 123, 'auth0|0132234', 'a-uuid', 'alw:fail', 404), # Auth0 id on vend_contact doesn't match auth0 id in request, but DOES match identity id pytest.param('auth0|0132234', 123, 'a-uuid', 'a-uuid', 'alw:123', 200), ], ) def test_set_auth0_primary_user_initial( mocker, auth0_id: str, user_id: int, identity_id: str, target_auth0_id: str, primary_user_id: str, status: int, context: flask.app.AppContext, ): """Test setting a primary auth0 user.""" login = 'testuser' formatted_user_id = 'alw:123' expected = response.create_not_found_response() raw_auth0_id = auth0_id.split('auth0|')[1] if target_auth0_id: expected = response.Response( message={'auth0_user_id': target_auth0_id, 'login': login, 'user_id': formatted_user_id} ) primary = response.create_not_found_response() if primary_user_id: primary = response.Response(message={'user_id': primary_user_id}) update = expected if primary_user_id == 'alw:fail': update = response.create_not_found_response() mocker.patch.object(user_info_model, 'fetch_vend_contact_for_user', return_value=expected) mocker.patch.object(user_info_model, 'fetch_primary_for_auth0_user', return_value=primary) mocker.patch.object(user_info_model, 'update_vend_contact_details', return_value=update) mocker.patch.object(auth0_client, 'update_user', return_value=expected) mocker.patch.object(identities, 'update_auth0_primary_to_identity') mocker.patch.object(identities, 'activate_deactivate_identity') mocker.patch.object( identities, 'get_identity_for_profile', return_value=MagicMock(message={'id': identity_id}) ) data = {'user_metadata': {'username': login, 'vend_contact_id': user_id, 'type': 'alw'}} with context: mocker.patch.object(user_info, 'g') result = user_info.set_auth0_primary_user(auth0_id=auth0_id, user_id=user_id) assert result.status == status if status == 200: user_info_model.fetch_vend_contact_for_user.assert_called_with(user_id) user_info_model.fetch_primary_for_auth0_user.assert_called_with(raw_auth0_id) auth0_client.update_user.assert_called_with(auth0_id, data) if primary: primary_user_parts = primary_user_id.split(':') call1 = call(primary_user_parts[1], {'auth0_primary': None}) call2 = call(user_id, {'auth0_primary': 'Y'}) user_info_model.update_vend_contact_details.assert_has_calls([call1, call2]) else: user_info_model.update_vend_contact_details.assert_called_with( user_id, {'auth0_primary': 'Y'} ) def test_set_auth0_primary_user_pending_skips_auth0_update(mocker, context: flask.app.AppContext): """Test that a pending user gets primary flags set in the db but no auth0 update. A pending user (invite not yet accepted) has their identity id in place of an auth0 id, so there is no auth0 user to update. """ identity_id = 'a-uuid' user_id = 123 vend_contact = response.Response( message={'auth0_user_id': identity_id, 'login': 'testuser', 'user_id': 'alw:123'} ) mocker.patch.object(user_info_model, 'fetch_vend_contact_for_user', return_value=vend_contact) mocker.patch.object( user_info_model, 'fetch_primary_for_auth0_user', return_value=response.create_not_found_response(), ) mocker.patch.object(user_info_model, 'update_vend_contact_details', return_value=vend_contact) mocker.patch.object(auth0_client, 'update_user') mocker.patch.object(identities, 'update_auth0_primary_to_identity') mocker.patch.object(identities, 'activate_deactivate_identity') with context: mocker.patch.object(user_info, 'g') result = user_info.set_auth0_primary_user( auth0_id=f'auth0|{identity_id}', user_id=user_id, identity_id=identity_id ) assert result.status == 200 assert result.message == { 'user_metadata': {'username': 'testuser', 'vend_contact_id': user_id, 'type': 'alw'} } user_info_model.update_vend_contact_details.assert_called_with(user_id, {'auth0_primary': 'Y'}) identities.update_auth0_primary_to_identity.assert_called_with(identity_id, user_id) auth0_client.update_user.assert_not_called() identities.activate_deactivate_identity.assert_called_with(identity_id, False) @pytest.mark.parametrize('pending', [True, False]) def test_set_auth0_primary_user_success_shape_is_consistent( mocker, pending: bool, context: flask.app.AppContext ): """The pending and auth0 success paths return the same response contract. Auth0's raw user object should not appear into the response; both paths return the primary vend_contact's metadata so body-reading callers get a consistent shape. """ identity_id = 'a-uuid' user_id = 123 # Pending: the auth0 id addressed is the identity id. Non-pending: a real auth0 id. auth0_id = f'auth0|{identity_id}' if pending else 'auth0|real-auth0-id' vend_contact = response.Response( message={'auth0_user_id': identity_id, 'login': 'testuser', 'user_id': 'alw:123'} ) mocker.patch.object(user_info_model, 'fetch_vend_contact_for_user', return_value=vend_contact) mocker.patch.object( user_info_model, 'fetch_primary_for_auth0_user', return_value=response.create_not_found_response(), ) mocker.patch.object(user_info_model, 'update_vend_contact_details', return_value=vend_contact) # Distinct, richer object so auth0's response would be visible in the assert. mocker.patch.object( auth0_client, 'update_user', return_value=response.Response( message={'user_id': 'auth0|real-auth0-id', 'user_metadata': {'vend_contact_id': '999'}}, status=200, ), ) mocker.patch.object(identities, 'update_auth0_primary_to_identity') mocker.patch.object(identities, 'activate_deactivate_identity') with context: mocker.patch.object(user_info, 'g') result = user_info.set_auth0_primary_user( auth0_id=auth0_id, user_id=user_id, identity_id=identity_id ) expected = { 'user_metadata': {'username': 'testuser', 'vend_contact_id': user_id, 'type': 'alw'} } assert result.status == 200 assert result.message == expected if pending: auth0_client.update_user.assert_not_called() else: auth0_client.update_user.assert_called_once_with(auth0_id, expected) def test_set_auth0_primary_user_propagates_auth0_failure(mocker, context: flask.app.AppContext): """A genuine auth0 failure on the non-pending path is surfaced, not masked as success.""" identity_id = 'a-uuid' user_id = 123 vend_contact = response.Response( message={'auth0_user_id': identity_id, 'login': 'testuser', 'user_id': 'alw:123'} ) mocker.patch.object(user_info_model, 'fetch_vend_contact_for_user', return_value=vend_contact) mocker.patch.object( user_info_model, 'fetch_primary_for_auth0_user', return_value=response.create_not_found_response(), ) mocker.patch.object(user_info_model, 'update_vend_contact_details', return_value=vend_contact) auth0_error = response.create_error_response(code='AUTH0_ERROR', message='boom', status=502) mocker.patch.object(auth0_client, 'update_user', return_value=auth0_error) mocker.patch.object(identities, 'update_auth0_primary_to_identity') mocker.patch.object(identities, 'activate_deactivate_identity') with context: mocker.patch.object(user_info, 'g') result = user_info.set_auth0_primary_user( auth0_id='auth0|real-auth0-id', user_id=user_id, identity_id=identity_id ) assert result is auth0_error assert result.status == 502 @pytest.mark.parametrize( ('user_id', 'app', 'linked_accounts', 'primary', 'auth0_id', 'status'), [ (123, 'alw', [], None, 'auth0|abc', 200), (123, 'oa', [], None, 'auth0|abc', 200), (123, 'foo', [], None, 'auth0|abc', 400), (123, 'alw', [{'vc_id': 456}], None, 'auth0|abc', 304), (123, 'alw', [{'vc_id': 456}], 'Y', 'auth0|abc', 200), (123, 'alw', [], 'Y', 'auth0|abc', 200), (123, 'alw', [], 'Y', 'abc', 200), # legacy user that has not been migrated to Auth0. (123, 'alw', [], 'Y', None, 200), ], ) def test_deactivate_user(mocker, user_id, app, linked_accounts, primary, auth0_id, status): """Test deactivating a user.""" expected_auth0_id = auth0_id if auth0_id and 'auth0' not in auth0_id: expected_auth0_id = 'auth0|{}'.format(auth0_id) expected = response.Response(message={'auth0_user_id': auth0_id, 'primary': primary}) expected_output = response.Response(message=[{'auth0_user_id': auth0_id}]) update_primary = response.Response(message={'auth0_user_id': auth0_id, 'primary': primary}) linked = response.Response(message=linked_accounts) mocker.patch.object(user_info_model, 'fetch_vend_contact_for_user', return_value=expected) mocker.patch.object(user_info, 'get_linked_accounts', return_value=linked) mocker.patch.object(user_info_model, 'fetch_oa_users', return_value=expected_output) mocker.patch.object(user_info, 'set_auth0_primary_user', return_value=update_primary) mocker.patch.object(auth0_client, 'update_user', return_value=update_primary) mocker.patch.object(identities, 'activate_deactivate_identity') result = user_info.deactivate_user(user_id, app) assert result.status == status if primary and len(linked_accounts): user_info.set_auth0_primary_user.assert_called_with( expected_auth0_id, linked_accounts[0]['vc_id'] ) if status == 200 and result != update_primary and auth0_id: auth0_client.update_user.assert_called_with(expected_auth0_id, {'blocked': True}) @pytest.mark.parametrize(('user_type', 'status'), [('alw', 200), ('oa', 501), ('foo', 400)]) def test_reset_user_auth0_details(mocker, user_type, status): """Test get_linked_accounts.""" user_id = '12323' expected = response.Response(message={'foo': 'bar'}) mocker.patch.object(user_info_model, 'reset_user_auth0_details', return_value=expected) result = user_info.reset_user_auth0_details(user_id, user_type) assert result.status == status if status == 200: user_info_model.reset_user_auth0_details.assert_called_with(user_id) assert result == expected else: user_info_model.reset_user_auth0_details.asset_not_called() @pytest.mark.parametrize( ('expected', 'data', 'second_call'), [ (response.Response(message={'foo': 'bar'}), {'login': 'new_name'}, 0), (response.create_fatal_response({'db': 'err'}), {'login': 'new_name'}, 0), ( response.Response(message={'login': 'new_name', 'contact_id': '7789'}), {'login': 'new_name', 'contact': {'first_name': 'foo'}}, 1, ), ( response.create_fatal_response({'db': 'err'}), {'login': 'new_name', 'contact': {'first_name': 'foo'}}, 0, ), ], ) def test_update_vend_contact_user(mocker, expected, data, second_call): """Test update_vend_contact_user.""" user_id = '12323' mocker.patch.object(user_info_model, 'update_vend_contact_details', return_value=expected) mocker.patch.object(user_info_model, 'update_contact_details') result = user_info.update_vend_contact_user(user_id, data) assert result.status == expected.status assert result == expected user_info_model.update_vend_contact_details.assert_called_with(user_id, data) assert user_info_model.update_contact_details.call_count == second_call def test_update_vend_contacts(mocker): """Test update_vend_contacts.""" old_auth0_user_id = 'identity-uuid' new_auth0_user_id = 'new-auth0-id' expected = response.Response(message='1 rows updated.') mocker.patch.object(user_info_model, 'update_vend_contacts', return_value=expected) result = user_info.update_vend_contacts(old_auth0_user_id, new_auth0_user_id) assert result.status == expected.status assert result.message == expected.message user_info_model.update_vend_contacts.assert_called_with(old_auth0_user_id, new_auth0_user_id) @freeze_time('2017-11-15 11:31:0') def test_get_user_info_feature_fm(monkeypatch): """Test get user info for feature.fm.""" account_type = 'vendor' account_id = 1234 user_id = 'alw:123' correlation_id = 'correlation' user = { 'type': 'alw', 'language': 'en', 'account': {'subaccount_id': None, 'vendor_id': 1234}, 'email': 'john.doe@theorchard.com', 'last_name': 'Doe', 'first_name': 'John', 'user_id': 'alw:123', 'company': 'Company', 'role_ids': [1], 'timestamp': 1510745460, } features_response = response.Response({'items': []}) monkeypatch.setattr( user_info, 'get_users_minimum_details', MagicMock(return_value=response.Response(user)) ) monkeypatch.setattr( ows_account_model, 'get_enabled_features_for_vendor', MagicMock(return_value=features_response), ) monkeypatch.setattr( ows_account_model, 'get_vendor_currency_code', MagicMock(return_value='USD') ) with app.test_request_context(): result = user_info.get_user_info_feature_fm( user_id, account_type, account_id, correlation_id ) user_info.get_users_minimum_details.assert_called_once_with('alw', user_id, True) ows_account_model.get_enabled_features_for_vendor.assert_called_once_with(1234, correlation_id) assert result.message == { 'user_id': 'alw:123', 'first_name': 'John', 'last_name': 'Doe', 'email': 'john.doe@theorchard.com', 'language': 'en', 'currency': 'USD', 'company': 'Company', 'permissions': {'action_pages': 'W', 'smart_links': 'W', 'campaigns': 'N'}, 'facebook_boost': False, 'timestamp': 1510745460, 'account_id': 1234, 'account_type': 'vendor', 'ga_flags': { 'advertising_role': False, 'facebook_campaign_boost': False, 'orchard_advertising': False, 'orchard_advertising_tier_1': False, }, } @pytest.mark.parametrize( ('activated_features'), [(constants.CAMPAIGNS_FEATURE_NAME, constants.FACEBOOK_BOOST_NAME)] ) @freeze_time('2017-11-15 11:31:0') def test_get_user_info_feature_fm_campaign_advertising_role_facebook_boost( monkeypatch, activated_features ): """Test get user info for feature.fm with advertising role.""" account_type = 'vendor' account_id = 1234 user_id = 'alw:123' correlation_id = 'correlation' user = { 'type': 'alw', 'language': 'en', 'account': {'subaccount_id': None, 'vendor_id': 1234}, 'email': 'john.doe@theorchard.com', 'last_name': 'Doe', 'first_name': 'John', 'user_id': 'alw:123', 'company': 'Company', 'role_ids': [constants.ADVERTISING_ROLE_ID], 'timestamp': 1510745460, } features_response = response.Response( { 'items': [ {'feature_name': activated_features[0]}, {'feature_name': activated_features[1]}, ] } ) monkeypatch.setattr( user_info, 'get_users_minimum_details', MagicMock(return_value=response.Response(user)) ) monkeypatch.setattr( ows_account_model, 'get_enabled_features_for_vendor', MagicMock(return_value=features_response), ) monkeypatch.setattr( ows_account_model, 'get_vendor_currency_code', MagicMock(return_value='USD') ) with app.test_request_context(): result = user_info.get_user_info_feature_fm( user_id, account_type, account_id, correlation_id ) user_info.get_users_minimum_details.assert_called_once_with('alw', user_id, True) ows_account_model.get_enabled_features_for_vendor.assert_called_once_with(1234, correlation_id) assert result.message == { 'user_id': 'alw:123', 'first_name': 'John', 'last_name': 'Doe', 'email': 'john.doe@theorchard.com', 'language': 'en', 'currency': 'USD', 'company': 'Company', 'permissions': {'action_pages': 'W', 'smart_links': 'W', 'campaigns': 'W'}, 'facebook_boost': True, 'timestamp': 1510745460, 'account_id': 1234, 'account_type': 'vendor', 'ga_flags': { 'advertising_role': True, 'facebook_campaign_boost': True, 'orchard_advertising': True, 'orchard_advertising_tier_1': False, }, } @freeze_time('2017-11-15 11:31:0') def test_get_user_info_feature_fm_campaign_admin_role(monkeypatch): """Test get user info for feature.fm with admin role.""" account_type = 'vendor' account_id = 1234 user_id = 'alw:123' correlation_id = 'correlation' user = { 'type': 'alw', 'language': 'en', 'account': {'subaccount_id': None, 'vendor_id': 1234}, 'email': 'john.doe@theorchard.com', 'last_name': 'Doe', 'first_name': 'John', 'user_id': 'alw:123', 'company': 'Company', 'role_ids': [constants.ADMIN_ROLE_ID], 'timestamp': 1510745460, } features_response = response.Response( {'items': [{'feature_name': constants.CAMPAIGNS_FEATURE_NAME}]} ) monkeypatch.setattr( user_info, 'get_users_minimum_details', MagicMock(return_value=response.Response(user)) ) monkeypatch.setattr( ows_account_model, 'get_enabled_features_for_vendor', MagicMock(return_value=features_response), ) monkeypatch.setattr( ows_account_model, 'get_vendor_currency_code', MagicMock(return_value='USD') ) with app.test_request_context(): result = user_info.get_user_info_feature_fm( user_id, account_type, account_id, correlation_id ) user_info.get_users_minimum_details.assert_called_once_with('alw', user_id, True) ows_account_model.get_enabled_features_for_vendor.assert_called_once_with(1234, correlation_id) assert result.message == { 'user_id': 'alw:123', 'first_name': 'John', 'last_name': 'Doe', 'email': 'john.doe@theorchard.com', 'language': 'en', 'currency': 'USD', 'company': 'Company', 'permissions': {'action_pages': 'W', 'smart_links': 'W', 'campaigns': 'W'}, 'facebook_boost': False, 'timestamp': 1510745460, 'account_id': 1234, 'account_type': 'vendor', 'ga_flags': { 'advertising_role': True, 'facebook_campaign_boost': False, 'orchard_advertising': True, 'orchard_advertising_tier_1': False, }, } def test_get_user_info_feature_fm_error(monkeypatch): """Test get user info when the request to user_info fails.""" account_type = 'vendor' account_id = 1234 user_id = 'alw:123' correlation_id = 'correlation' features_response = response.Response( {'items': [{'feature_name': constants.CAMPAIGNS_FEATURE_NAME}]} ) monkeypatch.setattr( ows_account_model, 'get_enabled_features_for_vendor', MagicMock(return_value=features_response), ) monkeypatch.setattr( user_info, 'get_users_minimum_details', MagicMock(return_value=response.create_fatal_response()), ) with app.test_request_context(): result = user_info.get_user_info_feature_fm( user_id, account_type, account_id, correlation_id ) assert result.status == 500 def test_get_user_info_feature_fm_features_error(monkeypatch): """Test get user info when the request to user_info fails.""" account_type = 'vendor' account_id = 1234 user_id = 'alw:123' correlation_id = 'correlation' user = { 'type': 'alw', 'language': 'en', 'account': {'subaccount_id': None, 'vendor_id': 1234}, 'email': 'john.doe@theorchard.com', 'last_name': 'Doe', 'first_name': 'John', 'user_id': 'alw:123', 'company': 'Company', 'role_ids': [constants.ADMIN_ROLE_ID], } monkeypatch.setattr( ows_account_model, 'get_enabled_features_for_vendor', MagicMock(return_value=response.create_fatal_response()), ) monkeypatch.setattr( user_info, 'get_users_minimum_details', MagicMock(return_value=response.Response(user)) ) monkeypatch.setattr( ows_account_model, 'get_vendor_currency_code', MagicMock(return_value=response.create_fatal_response()), ) with app.test_request_context(): result = user_info.get_user_info_feature_fm( user_id, account_type, account_id, correlation_id ) assert result.status == 500 def test_get_user_info_feature_fm_wrong_account(monkeypatch): """Test get user info for feature.fm with admin role.""" account_type = 'vendor' account_id = 1255 user_id = 'alw:123' correlation_id = 'correlation' user = { 'type': 'alw', 'language': 'en', 'account': {'subaccount_id': None, 'vendor_id': 1234}, 'email': 'john.doe@theorchard.com', 'last_name': 'Doe', 'first_name': 'John', 'user_id': 'alw:123', 'company': 'Company', 'role_ids': [constants.ADMIN_ROLE_ID], } features_response = response.Response( {'items': [{'feature_name': constants.CAMPAIGNS_FEATURE_NAME}]} ) monkeypatch.setattr( user_info, 'get_users_minimum_details', MagicMock(return_value=response.Response(user)) ) monkeypatch.setattr( ows_account_model, 'get_enabled_features_for_vendor', MagicMock(return_value=features_response), ) with app.test_request_context(): result = user_info.get_user_info_feature_fm( user_id, account_type, account_id, correlation_id ) assert result.status == 404 @pytest.mark.parametrize( ('user_type', 'auth0_id', 'primary_user', 'linked', 'status'), [ ('alw', 'auth0|1', 'auth0|abc', [], 304), ('alw', 'auth0|1', None, ['123'], 200), ('alw', 'auth0|1', None, [], 200), ('alw', '1', None, [], 200), ], ) def test_cleanup_auth0_user(mocker, user_type, auth0_id, primary_user, linked, status): """Test cleanup_auth0_user.""" expected = response.Response(message={'foo': 'bar'}) primary = response.create_not_found_response() if primary_user: primary = expected if 'auth0' not in auth0_id: raw_auth0_id = auth0_id auth0_id = 'auth0|{}'.format(auth0_id) else: raw_auth0_id = auth0_id.replace('auth0|', '') link_response = response.Response(message=linked) data = {'blocked': True} mocker.patch.object(user_info_model, 'fetch_primary_for_auth0_user', return_value=primary) mocker.patch.object(user_info_model, 'fetch_users_by_auth0_id', return_value=link_response) mocker.patch.object(auth0_client, 'update_user', return_value=expected) mocker.patch.object(user_info, 'set_auth0_primary_user', return_value=expected) mocker.patch.object(identities, 'activate_deactivate_identity') result = user_info.cleanup_auth0_user(auth0_id) assert result.status == status user_info_model.fetch_primary_for_auth0_user.assert_called_with(raw_auth0_id) if len(linked): user_info.set_auth0_primary_user.assert_called_with(auth0_id, linked[0]) if not primary_user and not linked: auth0_client.update_user.assert_called_with(auth0_id, data) if status == 200: user_info_model.fetch_users_by_auth0_id.assert_called_with(raw_auth0_id) assert result == expected @pytest.mark.parametrize( ('user_id', 'contact_id', 'subaccount_id', 'expected_result'), [ ( 123, None, None, [ { 'user_id': 123, 'subaccount': 'El Subaccount', 'is_active': 0, 'master': 'N', 'vendor_id': 1234, 'contact_first_name': 'test', 'contact_last_name': 'test', 'contact_email': 'test@test.com', 'permissions': 'Administrator,Accounting', 'subaccount_id': 3, } ], ), ( None, 123, None, [ { 'user_id': 123, 'subaccount': 'El Subaccount', 'is_active': 0, 'master': 'N', 'vendor_id': 1234, 'contact_first_name': 'test', 'contact_last_name': 'test', 'contact_email': 'test@test.com', 'permissions': 'Administrator,Accounting', 'subaccount_id': 3, } ], ), ( None, None, 3, [ { 'user_id': 123, 'subaccount': 'El Subaccount', 'is_active': 0, 'master': 'N', 'vendor_id': 1234, 'contact_first_name': 'test', 'contact_last_name': 'test', 'contact_email': 'test@test.com', 'permissions': 'Administrator,Accounting', 'subaccount_id': 3, } ], ), ], ) def test_get_user_basic_info(mocker, user_id, contact_id, subaccount_id, expected_result): """Test successfully getting user basic information.""" mocker.patch.object(user_info_model, 'fetch_user_basic_info', return_value=expected_result) result = user_info.get_user_basic_info( user_id=user_id, contact_id=contact_id, subaccount_id=subaccount_id ) assert result == expected_result @pytest.mark.parametrize( ('expected', 'data'), [ (response.Response(message={'foo': 'bar'}), {'active': 'Y'}), (response.create_fatal_response({'db': 'err'}), {'active': 'Y'}), ], ) def test_update_user_status(mocker, expected, data): """Test update_user_status for success.""" user_id = '12323' mocker.patch.object(user_info, 'update_vend_contact_user', return_value=expected) result = user_info.update_user_status(user_id, data) assert result == expected def test_update_user_status_bad_request(): """Test update_user_status for invalid input.""" user_id = 1234 data = {'test': 'test'} result = user_info.update_user_status(user_id, data) assert result.status == 400 assert result.message == constants.MISSING_STATUS_MESSAGE def test_get_contact_details(monkeypatch, user_data): """Test test_get_contact_details.""" monkeypatch.setattr( user_info_model, 'get_contact_details', MagicMock(return_value=response.Response(user_data)) ) result = user_info.get_contact_details('subaccount', 'disable', 28518, 1, 0, 50) assert result.status == 200 assert result.message == user_data def test_get_alw_users_by_email(monkeypatch): """Test get_alw_users_by_email.""" monkeypatch.setattr( user_info_raw, 'fetch_alw_users_by_email', MagicMock( return_value=response.Response( {'items': [{'email': 'email', 'alw_id': 123, 'vendor_id': 456}]} ) ), ) result = user_info.get_alw_users_by_email('email') assert result.status == 200 assert result.message == {'items': [{'email': 'email', 'alw_id': 123, 'vendor_id': 456}]} @pytest.mark.parametrize( ('email', 'expected_connection'), [ # Valid Google tenant emails ('test@theorchard.com', 'theorchard-gsuite'), ('john.doe@kollectivenr.com', 'theorchard-gsuite'), ('employee@human-re-sources.com', 'theorchard-gsuite'), ('user@awal.com', 'theorchard-gsuite'), ('user_2@awal.com', 'theorchard-gsuite'), ('user-20@awal.com', 'theorchard-gsuite'), ('user.200@awal.com', 'theorchard-gsuite'), ('person@sonymusic-pde.com', 'theorchard-gsuite'), ('staff@magicstarkids.com', 'theorchard-gsuite'), ('member@in2unemusic.com', 'theorchard-gsuite'), ('artist@indiemerch.com', 'theorchard-gsuite'), ('manager@aboveboarddist.co.uk', 'theorchard-gsuite'), ('admin@drm.co.nz', 'theorchard-gsuite'), ('friend@altafonte.com', 'theorchard-gsuite'), # Valid Sony emails ('user@sonymusic.com', 'sme-prod-orchadmin'), ('user_1@sonymusic.com', 'sme-prod-orchadmin'), ('user-10@sonymusic.com', 'sme-prod-orchadmin'), ('user.100@sonymusic.com', 'sme-prod-orchadmin'), ('artist@rcarecords.com', 'sme-prod-orchadmin'), ('manager@epicrecords.com', 'sme-prod-orchadmin'), ('contact@alamo-records.com', 'sme-prod-orchadmin'), ('admin@ministryofsoundrecords.com', 'sme-prod-orchadmin'), ('staff@aristarecordings.com', 'sme-prod-orchadmin'), ('user@ultrarecords.com', 'sme-prod-orchadmin'), ('support@somlivre.com.br', 'sme-prod-orchadmin'), # Invalid emails ('test+123@theorchard.com', None), # '+' is not allowed ('invalid@unknown.com', None), # Unlisted domain ('@theorchard.com', None), # Missing username ('user@', None), # Missing domain ('user@.com', None), # Invalid domain ('user@@theorchard.com', None), # Double '@' ('pipe|user@theorchard.com', None), # Invalid character '|' ('user#email@theorchard.com', None), # Invalid character '#' ('user@theorchard..com', None), # Double dots in domain ('test@randomcompany.com', None), # Unlisted domain ('@sonymusic.com', None), # Missing username ('artist@', None), # Missing domain ('user@.br', None), # Invalid domain ('staff@@sonymusic.com', None), # Double '@' ('pipe|user@sonymusic.com', None), # Invalid character '|' ('contact#sony@sonymusic.com', None), # Invalid character '#' ('admin@sonymusic..com', None), # Double dots in domain ], ) def test_get_auth0_connection_name_from_email(email, expected_connection): """Test get_auth0_connection_name_from_email with various email patterns.""" result = user_info.get_auth0_connection_name_from_email(email) assert result == expected_connection def test_auth0_domains_no_overlaps(): """Test that AUTH0_DOMAINS_CONNECTIONS has no overlapping domains. Overlaps cause critical bugs because the Python code creates DB records with specific connections. Each domain must belong to exactly ONE connection to prevent wrong connection assignments and permission issues. """ # Collect all domains from all connections all_domains = [] domain_to_connections = {} for connection_name, domains in constants.AUTH0_DOMAINS_CONNECTIONS.items(): for domain in domains: domain_lower = domain.lower() all_domains.append(domain_lower) if domain_lower not in domain_to_connections: domain_to_connections[domain_lower] = [] domain_to_connections[domain_lower].append(connection_name) # Find duplicates duplicates = { domain: connections for domain, connections in domain_to_connections.items() if len(connections) > 1 } # Assert no duplicates assert duplicates == {}, ( 'OVERLAP: The following domains appear in MULTIPLE connections:\\n' + '\\n'.join( [f" - {domain}: {', '.join(conns)}" for domain, conns in sorted(duplicates.items())] ) + '\\n\\nEach domain must belong to exactly ONE connection to prevent bugs.' ) def test_get_alw_roles_for_user(monkeypatch): """Test get_roles_for_user for workstation.""" monkeypatch.setattr( user_info_model, 'get_alw_roles_for_user', MagicMock( return_value=response.Response({'role_ids': [4], 'role_names': ['Administrator']}) ), ) result = user_info.get_roles_for_user(123, 'alw') assert result.status == 200 assert result.message == {'role_ids': [4], 'role_names': ['Administrator']} def test_get_oa_roles_for_user(monkeypatch): """Test get_roles_for_user for orchAdmin.""" monkeypatch.setattr( user_info_model, 'get_oa_roles_for_user', MagicMock(return_value=response.Response({'role_ids': [1], 'role_names': ['Engineering']})), ) result = user_info.get_roles_for_user(123, 'oa') assert result.status == 200 assert result.message == {'role_ids': [1], 'role_names': ['Engineering']} @pytest.mark.parametrize( ('get_users_result', 'update_user_result', 'expected_result', 'auth0_called'), [ ( response.Response({'auth0_user_id': '123ABC'}), response.Response('update_user_result'), response.Response('update_user_result'), True, ), ( response.create_not_found_response(message='not found'), response.Response('update_user_result'), response.create_not_found_response(message='not found'), False, ), ( response.Response({'auth0_user_id': None}), None, response.Response(message='User has not been migrated to Auth0'), False, ), ], ) def test_reactivate_user( mocker, get_users_result, update_user_result, expected_result, auth0_called ): """Test reactivate_user.""" user_id = '1234' mocker.patch.object( user_info_model, 'fetch_vend_contact_for_user', return_value=get_users_result ) mocker.patch.object(auth0_client, 'update_user', return_value=update_user_result) mocker.patch.object(identities, 'activate_deactivate_identity') user_result = user_info.reactivate_user(user_id) assert user_result.status == expected_result.status assert user_result.message == expected_result.message user_info_model.fetch_vend_contact_for_user.assert_called_once_with(user_id) if auth0_called: auth0_client.update_user.assert_called_once_with('auth0|123ABC', {'blocked': False}) def test_get_primary_contact(monkeypatch, user_data): """Test test_get_primary_contact.""" monkeypatch.setattr( user_info_model, 'get_primary_contact', MagicMock(return_value=response.Response(user_data)) ) result = user_info.get_primary_contact('vendor', 25428, None) assert result.status == 200 assert result.message == user_data @pytest.mark.parametrize( ('identity_id', 'participant_id', 'platform', 'fetch_result', 'expected_response'), [ ( 'a', 'b', 'c', {'identity_id': 'a'}, response.Response(status=200, message={'can_unlink': True}), ), ('a', 'b', 'c', None, response.Response(status=200, message={'can_unlink': False})), ], ) def test_can_unlink_participant_social_account( mocker, identity_id, participant_id, platform, fetch_result, expected_response ): """Test can_unlink_participant_social_account.""" mocker.patch.object(social_auth_item, 'fetch', return_value=fetch_result) user_result = user_info.can_unlink_participant_social_account( identity_id, participant_id, platform ) assert user_result.status == expected_response.status assert user_result.message['can_unlink'] == expected_response.message['can_unlink'] @pytest.mark.parametrize( ( 'identity_id', 'participant_id', 'platform', 'fetch_result', 'update_result', 'expected_response', ), [ ( 'a', 'b', 'c', {'identity_id': 'a'}, {'linked': False}, response.Response( status=200, message={ 'identity_id': 'a', 'participant_id': 'b', 'platform': 'c', 'linked': False, }, ), ), ( 'a', 'b', 'c', None, {'linked': False}, response.create_error_response( code='user_error', message='Participant b social account for c cannot be unlinked by a', ), ), ( 'a', 'b', 'c', {'identity_id': 'a'}, {}, response.create_error_response( code='user_error', message='Participant b social account for c cannot be unlinked by a', ), ), ( 'a', 'b', 'c', {'identity_id': 'a'}, {'linked': True}, response.create_error_response( code='user_error', message='Participant b social account for c cannot be unlinked by a', ), ), ], ) def test_unlink_participant_social_account( mocker, identity_id, participant_id, platform, fetch_result, update_result, expected_response ): """Test unlink_participant_social_account.""" mocker.patch.object(social_auth_item, 'fetch', return_value=fetch_result) mocker.patch.object(social_auth_item, 'set_linked_to_false', return_value=update_result) user_result = user_info.unlink_participant_social_account(identity_id, participant_id, platform) assert user_result.status == expected_response.status if user_result.status == response.status.BAD_REQUEST: assert user_result.errors == expected_response.errors else: assert user_result.message == expected_response.message def test_get_account_managers(monkeypatch): """Test get account managers.""" expected = [{'id': 2, 'f_name': 'test', 'l_name': 'user'}] monkeypatch.setattr( user_info, 'get_account_managers', MagicMock(return_value=response.Response(expected)) ) result = user_info.get_account_managers() assert result assert result.status == 200 assert result.message == expected def test_get_active_closers(monkeypatch): """Test get active closers.""" expected = {'items': [{'user_id': 1, 'first_name': 'test', 'last_name': 'user'}]} monkeypatch.setattr( user_info, 'get_active_closers', MagicMock(return_value=response.Response(expected)) ) result = user_info.get_active_closers() assert result assert result.status == 200 assert result.message == expected @pytest.mark.parametrize( ('identity_id', 'label_profile_id', 'model_response', 'expected_result'), [ ( '12345', 4321, response.Response({'what': 'ever'}), response.Response({'what': 'ever'}), ), ], ) def test_get_identity_vendor( mocker, identity_id, label_profile_id, model_response, expected_result ): """Test get_identity_vendor.""" mocker.patch.object( identities, 'get_identity_vendor', return_value=model_response, autospec=True ) result = user_info.get_identity_vendor(identity_id, label_profile_id) assert result.status == expected_result.status assert result.message == expected_result.message NEO_NOW_MINUS_THREE_MINUTES = NeoDateTime.from_native( datetime.now(timezone.utc) - timedelta(minutes=3) ) @pytest.mark.parametrize( 'email,identity_response,expected_result', [ pytest.param( 'test@example.com', response.Response( status=200, message={ 'id': '123e4567-e89b-12d3-a456-426614174000', 'auth0_user_id': '123e4567-e89b-12d3-a456-426614174000', 'updated_on': NEO_NOW_MINUS_THREE_MINUTES, 'auth0_user_created_by': 'invitation', 'active': 'Y', 'invitation_id': 'uinv_112233445', 'organization_id': 'org_1234567', }, ), response.Response( status=200, message={ 'valid': True, 'invitation_id': 'uinv_112233445', 'organization_id': 'org_1234567', }, ), id='Validates and includes invitation_id and organization_id', ), pytest.param( 'nonexistent@example.com', response.Response(status=404, message=None), response.Response(status=200, message={'valid': False}), id="Reject if user doesn't exist", ), pytest.param( 'mismatch@example.com', response.Response( status=200, message={ 'id': '123e4567-e89b-12d3-a456-426614174000', 'auth0_user_id': '987fcdeb-e89b-12d3-a456-426614174000', 'updated_on': NEO_NOW_MINUS_THREE_MINUTES, 'auth0_user_created_by': 'invitation', 'active': 'Y', }, ), response.Response(status=200, message={'valid': False}), id='Reject when Auth0 UUID differs from Identity UUID', ), pytest.param( 'expired@example.com', response.Response( status=200, message={ 'id': '123e4567-e89b-12d3-a456-426614174000', 'auth0_user_id': '123e4567-e89b-12d3-a456-426614174000', 'updated_on': NeoDateTime(2024, 1, 1, 12, 0), 'auth0_user_created_by': 'invitation', 'active': 'Y', }, ), response.Response(status=200, message={'valid': False}), id='Reject when invitation has expired (> AUTH0_ORG_INVITE_TTL)', ), pytest.param( 'manual@example.com', response.Response( status=200, message={ 'id': '123e4567-e89b-12d3-a456-426614174000', 'auth0_user_id': '123e4567-e89b-12d3-a456-426614174000', 'updated_on': NEO_NOW_MINUS_THREE_MINUTES, 'auth0_user_created_by': 'manual', 'active': 'Y', }, ), response.Response(status=200, message={'valid': False}), id='Reject on recent creation but not created by invitation', ), pytest.param( 'inactive@example.com', response.Response( status=200, message={ 'id': '123e4567-e89b-12d3-a456-426614174000', 'auth0_user_id': '123e4567-e89b-12d3-a456-426614174000', 'updated_on': NEO_NOW_MINUS_THREE_MINUTES, 'auth0_user_created_by': 'invitation', 'active': 'N', }, ), response.Response(status=200, message={'valid': False}), id='Reject on recent creation but user is not active', ), pytest.param( 'notime@example.com', response.Response( status=200, message={ 'id': '123e4567-e89b-12d3-a456-426614174000', 'auth0_user_id': '123e4567-e89b-12d3-a456-426614174000', 'auth0_user_created_by': 'invitation', 'active': 'Y', }, ), response.Response(status=200, message={'valid': False}), id='Reject on missing any update/creation time to compare', ), pytest.param( 'baduuid@example.com', response.Response( status=200, message={ 'id': 'not-a-uuid', 'auth0_user_id': '1111222233b4a555566b7777', 'updated_on': NEO_NOW_MINUS_THREE_MINUTES, 'auth0_user_created_by': 'invitation', 'active': 'Y', }, ), response.Response(status=200, message={'valid': False}), id='Reject on invalid UUID formats', ), pytest.param( 'test_without_invitation_id@example.com', response.Response( status=200, message={ 'id': '123e4567-e89b-12d3-a456-426614174000', 'auth0_user_id': '123e4567-e89b-12d3-a456-426614174000', 'updated_on': NEO_NOW_MINUS_THREE_MINUTES, 'auth0_user_created_by': 'invitation', 'active': 'Y', }, ), response.Response(status=200, message={'valid': False}), id='Rejects when invitation_id is missing', ), pytest.param( 'auth0_invitations_only@example.com', response.Response( status=200, message={ 'id': '123e4567-e89b-12d3-a456-426614174000', 'auth0_user_id': '123e4567-e89b-12d3-a456-426614174000', 'updated_on': NEO_NOW_MINUS_THREE_MINUTES, 'auth0_user_created_by': 'invitation', 'active': 'Y', 'auth0_invitations': ( '{"awal": {"invitation_id": "uinv_112233445",' ' "organization_id": "org_awal_123",' ' "timestamp": "2026-07-06T12:00:00+00:00"}}' ), }, ), response.Response( status=200, message={ 'valid': True, 'invitation_id': 'uinv_112233445', 'organization_id': 'org_awal_123', }, ), id='Valid when auth0_invitations is set but flat invitation_id/organization_id are not', ), pytest.param( 'auth0_invitations_multi@example.com', response.Response( status=200, message={ 'id': '123e4567-e89b-12d3-a456-426614174000', 'auth0_user_id': '123e4567-e89b-12d3-a456-426614174000', 'updated_on': NEO_NOW_MINUS_THREE_MINUTES, 'auth0_user_created_by': 'invitation', 'active': 'Y', 'auth0_invitations': ( '{"awal": {"invitation_id": "uinv_old",' ' "organization_id": "org_awal_old",' ' "timestamp": "2026-01-01T00:00:00+00:00"},' ' "orchard": {"invitation_id": "uinv_new",' ' "organization_id": "org_orchard_new",' ' "timestamp": "2026-07-06T12:00:00+00:00"}}' ), }, ), response.Response( status=200, message={ 'valid': True, 'invitation_id': 'uinv_new', 'organization_id': 'org_orchard_new', }, ), id='Returns the most recent invitation when auth0_invitations has multiple entries', ), ], ) def test_verify_auth0_invitation(mocker, context, email, identity_response, expected_result): """Test verify_auth0_invitation with various scenarios.""" with context: mocker.patch.object(user_info, 'g') mocker.patch.object(config, 'AUTH0_ORG_INVITE_TTL', 3600) # 1 hour mocker.patch.object(config, 'SERVICE_NAME', 'test-service') if identity_response is not None: mocker.patch.object(identities, 'get_identity_by_email', return_value=identity_response) result = user_info.verify_auth0_invitation(email) assert result.status == expected_result.status assert result.message == expected_result.message MOCK_IDENTITY_ID = 'abc12345-def6-7890-abcd-ef1234567890' class TestGetPrimaryVendContactForIdentity: """Tests for get_primary_vend_contact_for_identity logic function.""" def test_identity_not_found(self, mocker): """Returns 404 when identity doesn't exist in Neo4j.""" mocker.patch.object(identities, 'get_identity_label_profile_ids', return_value=None) result = user_info.get_primary_vend_contact_for_identity(MOCK_IDENTITY_ID) assert result.status == 404 def test_no_label_profiles(self, mocker): """Returns 404 when identity has no LabelProfiles.""" mocker.patch.object( identities, 'get_identity_label_profile_ids', return_value={'profile_ids': []}, ) result = user_info.get_primary_vend_contact_for_identity(MOCK_IDENTITY_ID) assert result.status == 404 def test_single_profile(self, mocker): """Returns the profileId directly when there's only one LabelProfile.""" mocker.patch.object( identities, 'get_identity_label_profile_ids', return_value={'profile_ids': [12345]}, ) result = user_info.get_primary_vend_contact_for_identity(MOCK_IDENTITY_ID) assert result.status == 200 assert result.message == {'vend_contact_id': 12345} def test_multiple_profiles_mysql_primary(self, mocker): """Uses MySQL auth0_primary='Y' to disambiguate multiple profiles.""" mocker.patch.object( identities, 'get_identity_label_profile_ids', return_value={'profile_ids': [111, 222]}, ) mocker.patch.object( user_info_model, 'get_active_vend_contacts_by_ids', return_value=[ {'vend_contact_id': 111, 'vendor_id': 1, 'auth0_primary': 'Y'}, {'vend_contact_id': 222, 'vendor_id': 2, 'auth0_primary': None}, ], ) result = user_info.get_primary_vend_contact_for_identity(MOCK_IDENTITY_ID) assert result.status == 200 assert result.message == {'vend_contact_id': 111} def test_multiple_profiles_single_active_in_mysql(self, mocker): """Falls back to single active vend_contact when no primary flag is set.""" mocker.patch.object( identities, 'get_identity_label_profile_ids', return_value={'profile_ids': [111, 222]}, ) mocker.patch.object( user_info_model, 'get_active_vend_contacts_by_ids', return_value=[ {'vend_contact_id': 111, 'vendor_id': 1, 'auth0_primary': None}, ], ) result = user_info.get_primary_vend_contact_for_identity(MOCK_IDENTITY_ID) assert result.status == 200 assert result.message == {'vend_contact_id': 111} def test_multiple_profiles_ambiguous(self, mocker): """Returns 404 when multiple profiles and no way to disambiguate.""" mocker.patch.object( identities, 'get_identity_label_profile_ids', return_value={'profile_ids': [111, 222]}, ) mocker.patch.object( user_info_model, 'get_active_vend_contacts_by_ids', return_value=[ {'vend_contact_id': 111, 'vendor_id': 1, 'auth0_primary': None}, {'vend_contact_id': 222, 'vendor_id': 2, 'auth0_primary': None}, ], ) result = user_info.get_primary_vend_contact_for_identity(MOCK_IDENTITY_ID) assert result.status == 404