"""Tests for user invite...module...""" from unittest import mock import pytest from permissions.constants import application from permissions.exceptions.incomplete_result_error import IncompleteResultError from permissions.logic import user_invite from permissions.types import Identity, IdentityInput, Tenant, TenantRolesInput, TenantType VALID_IDENTITY = Identity( id='c-uuid', first_name='ab', last_name='cd', name='ab cd', email='e@ma.il', auth0_user_id='auth0|1234', user_types=['label'], active='Y', default_brand='theorchard', ) @mock.patch('permissions.logic.user_invite.vendor') def test_get_vendor_and_subaccount_id_for_vend_contact_vendor(vendor_mock): """Test get_vendor_and_subaccount_id_for_vend_contact for a vendor.""" vendor_mock.get_vendor_id_by_uuid.return_value = 111 session = mock.Mock() result = user_invite.get_vendor_and_subaccount_id_for_vend_contact( session=session, tenant=Tenant(tenant_type=TenantType.ACCOUNT, tenant_uuid='a-uuid') ) assert result == (111, None) vendor_mock.get_vendor_id_by_uuid.assert_called_with(session=session, uuid='a-uuid') @mock.patch('permissions.logic.user_invite.subaccount') def test_get_vendor_and_subaccount_id_for_vend_contact_subaccount(subaccount_mock): """Test get_vendor_and_subaccount_id_for_vend_contact for a subaccount.""" session = mock.Mock() subaccount_instance = mock.Mock(vendor_id=111, subaccount_id=55) subaccount_mock.get_subaccount_by_uuid.return_value = subaccount_instance result = user_invite.get_vendor_and_subaccount_id_for_vend_contact( session=session, tenant=Tenant(tenant_type=TenantType.SUBACCOUNT, tenant_uuid='b-uuid') ) assert result == (111, 55) subaccount_mock.get_subaccount_by_uuid.assert_called_with(session=session, uuid='b-uuid') @pytest.mark.parametrize( 'existing_profile', [ pytest.param(None, id='no existing profile'), pytest.param( { 'profile_id': 11111, 'profile_type': 'LabelProfile', 'uuid': 'profile-uuid', 'roles': ['analytics'], 'tenant_relationship': 'HAS_ACCESS_TO', }, id='existing profile with HAS_ACCESS_TO relationship', ), pytest.param( { 'profile_id': 11111, 'profile_type': 'LabelProfile', 'uuid': 'profile-uuid', 'roles': ['analytics'], 'tenant_relationship': 'DELETED_HAS_ACCESS_TO', }, id='existing profile with DELETED_HAS_ACCESS_TO relationship', ), ], ) @mock.patch('permissions.logic.user_invite.vend_contact_logic') @mock.patch('permissions.logic.user_invite.get_vendor_and_subaccount_id_for_vend_contact') @mock.patch('permissions.logic.user_invite.vendor_role') @mock.patch('permissions.logic.user_invite.mysql') @mock.patch('permissions.logic.user_invite.g') @mock.patch('permissions.logic.user_invite.vend_contact_role') @mock.patch('permissions.logic.user_invite.profile_model') def test_add_label_profile_with_tenant_relationship_existing_profile_or_non( profile_model_mock, vcr_model_mock, g_mock, mysql_mock, vendor_role_mock, get_ids, vc_logic_mock, existing_profile: None | dict[str, list[str]], app_context, ): """Test add_label_profile_with_tenant_relationship behavior when a profile exists or not.""" profile_model_mock.get_label_profile_by_identity_and_tenant.return_value = existing_profile vendor_role_mock.vendor_role_ids_from_label_profile_roles.return_value = [1, 2] get_ids.return_value = (111, None) vc_logic_mock.create_vend_contact_and_roles.return_value = mock.Mock(id=9876) mysql_session_mock = mock.Mock() mysql_mock.db_session.return_value.__enter__.return_value = mysql_session_mock neo4j_tx_mock = mock.Mock() params = { 'tx': neo4j_tx_mock, 'identity': VALID_IDENTITY, 'tenant': Tenant( tenant_type=TenantType.ACCOUNT, tenant_uuid='🔟-🔟-🔟', ), 'roles': ['catalog'], 'audit_user_id': '🆔-🆔-🆔', 'master_contact': False, } if existing_profile and existing_profile['tenant_relationship'] == 'HAS_ACCESS_TO': roles = list(set(params['roles'] + existing_profile['roles'])) else: roles = params['roles'] user_invite.add_label_profile_with_tenant_relationship(**params) profile_model_mock.get_label_profile_by_identity_and_tenant.assert_called_with( tx=neo4j_tx_mock, identity_id=params['identity'].id, tenant=params['tenant'], ) assert mysql_session_mock.expire_on_commit is False # Have to do some convoluted stuff due to non-deterministic array ordering of roles get_role_ids_mock = vendor_role_mock.vendor_role_ids_from_label_profile_roles get_role_ids_mock.assert_called_once() assert get_role_ids_mock.call_args.kwargs['session'] == mysql_session_mock assert sorted(get_role_ids_mock.call_args.kwargs['roles']) == sorted(roles) get_ids.assert_called_with( session=mysql_session_mock, tenant=params['tenant'], ) vc_logic_mock.create_vend_contact_and_roles.assert_called_with( session=mysql_session_mock, role_ids=[1, 2], identity=params['identity'], vendor_id=111, subaccount_id=None, master_contact=False, existing_profile=existing_profile, ) if existing_profile: g_mock.log.info.assert_called_with( 'Existing label profile found for identity and tenant', resources={ 'identity_id': 'c-uuid', 'tenant_type': 'account', 'tenant_uuid': '🔟-🔟-🔟', 'admin_identity_id': '🆔-🆔-🆔', }, ) if existing_profile['tenant_relationship'] == 'HAS_ACCESS_TO': vcr_model_mock.VendContactRole.delete_roles_by_vend_contact_id.assert_not_called() else: vcr_model_mock.VendContactRole.delete_roles_by_vend_contact_id.assert_called_with( session=mysql_session_mock, vend_contact_id=existing_profile['profile_id'], ) # More stuff to deal with non-deterministic array ordering upsert_lp_mock = profile_model_mock.create_or_update_label_profile_with_tenant_relationship upsert_lp_mock.assert_called_once() assert upsert_lp_mock.call_args.kwargs['tx'] == neo4j_tx_mock assert upsert_lp_mock.call_args.kwargs['identity_id'] == params['identity'].id assert upsert_lp_mock.call_args.kwargs['profile_id'] == 9876 assert sorted(upsert_lp_mock.call_args.kwargs['roles']) == sorted(roles) assert upsert_lp_mock.call_args.kwargs['tenant'] == params['tenant'] assert upsert_lp_mock.call_args.kwargs['audit_user_id'] == params['audit_user_id'] @mock.patch('permissions.logic.user_invite._create_vend_contact_records') @mock.patch('permissions.connectors.mysql._db_transaction_session') @mock.patch('permissions.logic.user_invite.g') @mock.patch('permissions.logic.user_invite.profile_model') def test_add_label_profile_with_tenant_relationship_mysql_rollback( profile_model_mock, g_mock, mysql_session_mock, create_vend_contact_records_mock, app_context, ): """Test mysql rollback for create vend contact records.""" profile_model_mock.get_label_profile_by_identity_and_tenant.return_value = { 'profileId': 11111, 'profileType': 'LabelProfile', 'roles': ['analytics'], 'uuid': 'profile-uuid', 'tenant_relationship': 'HAS_ACCESS_TO', } create_vend_contact_records_mock.return_value = mock.Mock(id='1') profile_model_mock.create_or_update_label_profile_with_tenant_relationship.side_effect = ( Exception('test') ) neo4j_tx_mock = mock.Mock() params = { 'tx': neo4j_tx_mock, 'identity': mock.Mock(id='i-1'), 'tenant': Tenant(tenant_type=TenantType.ACCOUNT, tenant_uuid='a-1'), 'roles': ['catalog'], 'audit_user_id': 'a-2', 'master_contact': False, } with pytest.raises(IncompleteResultError): user_invite.add_label_profile_with_tenant_relationship(**params) profile_model_mock.get_label_profile_by_identity_and_tenant.assert_called_with( tx=neo4j_tx_mock, identity_id='i-1', tenant=params['tenant'], ) # ensure that created vend contact with roles in mysql will rollback mysql_session_mock.return_value.rollback.assert_called() @mock.patch('permissions.logic.user_invite.g') @mock.patch('permissions.logic.user_invite.vendor_role') @mock.patch('permissions.logic.user_invite.mysql') @mock.patch('permissions.logic.user_invite.profile_model') def test_add_label_profile_with_tenant_relationship_error( profile_model_mock, _, vendor_role_mock, g_mock, app_context ): """Test add_label_profile_with_tenant_relationship when an error is raised.""" profile_model_mock.get_label_profile_by_identity_and_tenant.return_value = None err = Exception('❗️') vendor_role_mock.vendor_role_ids_from_label_profile_roles.side_effect = err neo4j_tx_mock = mock.Mock() params = { 'tx': neo4j_tx_mock, 'identity': VALID_IDENTITY, 'tenant': Tenant( tenant_type=TenantType.ACCOUNT, tenant_uuid='🔟-🔟-🔟', ), 'roles': ['catalog'], 'audit_user_id': '🆔-🆔-🆔', 'master_contact': False, } with pytest.raises(Exception): user_invite.add_label_profile_with_tenant_relationship(**params) g_mock.log.error.assert_called_with( 'Error creating vend_contact records or label profile for user', resources={ 'error': err, 'tenant_uuid': '🔟-🔟-🔟', 'tenant_type': 'account', 'identity_id': 'c-uuid', }, ) neo4j_tx_mock.rollback.assert_not_called() @pytest.mark.parametrize( ('profiles_and_roles', 'label_profile_roles', 'profile_role_dicts'), [ # If settings role is passed in, a settings profile is created [{'SettingsProfile': []}, [], [{'profile_type': 'SettingsProfile', 'roles': []}]], # If settings role is not passed in, a settings profile is still created [ {'InsightsProfile': ['analytics'], 'SongwhipProfile': ['songwhip']}, [], [ {'profile_type': 'InsightsProfile', 'roles': ['analytics']}, {'profile_type': 'SongwhipProfile', 'roles': ['songwhip']}, {'profile_type': 'SettingsProfile', 'roles': []}, ], ], # If label profile role is passed in, it's not created...until later [ {'LabelProfile': ['catalog']}, ['catalog'], [ {'profile_type': 'SettingsProfile', 'roles': []}, ], ], ], ) @mock.patch('permissions.logic.user_invite.add_label_profile_with_tenant_relationship') @mock.patch('permissions.logic.user_invite.profile_model') @mock.patch('permissions.logic.user_invite.profile_logic') def test_add_profiles_to_identity_from_v2_roles( profile_logic_mock, profile_model_mock, add_label_profile_mock, profiles_and_roles: dict[str, list[str]], label_profile_roles: list[str], profile_role_dicts: list[dict[str, list[str]]], ): """Test add_profiles_to_identity_from_v2_roles.""" profile_logic_mock.v2_roles_to_profiles_and_roles_dict.return_value = profiles_and_roles neo4j_tx_mock = mock.Mock() params = { 'tx': neo4j_tx_mock, 'identity': VALID_IDENTITY, 'tenant_roles_input': TenantRolesInput( roles_to_attach=['this gets patched away immediately'], roles_to_detach=[], tenant=Tenant( tenant_type=TenantType.ACCOUNT, tenant_uuid='🔟-🔟-🔟', ), ), 'audit_user_id': '🆔-🆔-🆔', 'master_contact': True, } user_invite.add_profiles_to_identity_from_v2_roles(**params) profile_logic_mock.v2_roles_to_profiles_and_roles_dict.assert_called_with( params['tenant_roles_input'].roles_to_attach ) profile_model_mock.update_existing_profiles_with_roles.assert_called_with( tx=neo4j_tx_mock, identity_id=params['identity'].id, profile_types_and_roles=profile_role_dicts, audit_user=params['audit_user_id'], ) profile_model_mock.create_profiles_if_not_exist.assert_called_with( tx=neo4j_tx_mock, identity_id=params['identity'].id, profile_name=params['identity'].name, profile_types_and_roles=profile_role_dicts, audit_user=params['audit_user_id'], ) calls = [ mock.call( tx=neo4j_tx_mock, identity_id=params['identity'].id, profile_type=profile_type, tenant=params['tenant_roles_input'].tenant, audit_user_id=params['audit_user_id'], ) for profile_type in profiles_and_roles ] profile_model_mock.add_tenant_connection_to_profile_and_clear_cache.assert_has_calls(calls) if label_profile_roles: add_label_profile_mock.assert_called_with( tx=neo4j_tx_mock, identity=params['identity'], tenant=params['tenant_roles_input'].tenant, roles=label_profile_roles, audit_user_id=params['audit_user_id'], master_contact=True, ) else: add_label_profile_mock.assert_not_called() @pytest.mark.parametrize('is_transaction_closed', [True, False]) @mock.patch('permissions.logic.user_invite.sentry_client') @mock.patch('permissions.logic.user_invite.update_identity_with_v2_roles') @mock.patch('permissions.connectors.neo4j.db_session') def test_create_or_update_user_error_after_commit( db_session_mock, update_identity_mock, _, is_transaction_closed: bool, ): """Test create_or_update_user behavior if error is raised after transaction is committed.""" neo_session_mock = mock.Mock() db_session_mock.return_value.__enter__.return_value = neo_session_mock transaction_mock = mock.Mock() neo_session_mock.begin_transaction.return_value = transaction_mock err = Exception('💔') update_identity_mock.side_effect = err transaction_mock.closed.return_value = is_transaction_closed with pytest.raises(Exception) as caught_err: user_invite.create_or_update_user( admin_identity=mock.Mock(), assignee_identity=mock.Mock(), tenant_roles_input=mock.Mock(), brand='whatever', master_contact=False, localization=None, ) assert caught_err == err if is_transaction_closed: transaction_mock.rollback.assert_not_called() else: transaction_mock.rollback.assert_called() @pytest.mark.parametrize( ['brand', 'default_brand'], [ ('knr', 'knr'), ('awal', 'awal'), ('altafonte', 'theorchard'), ('msk', 'theorchard'), ('theorchard', 'theorchard'), ], ) @mock.patch('permissions.connectors.neo4j.db_session') @mock.patch('permissions.logic.user_invite.identity_model') @mock.patch('permissions.logic.user_invite.add_profiles_to_identity_from_v2_roles') def test_create_or_update_user_create_with_defaultbrand( _add_profiles_mock: mock.MagicMock, identity_model_mock: mock.MagicMock, db_session_mock: mock.MagicMock, brand: str, default_brand: str, ) -> None: """Test that brand is mapped to an auth0 org when saved on the identity node.""" admin_identity = mock.Mock(id='admin-identity-id') assignee_identity = IdentityInput( first_name='Balthazar', last_name='Turtle', email='b@tu.rtle', ) tenant_roles_input = mock.Mock(tenant=mock.Mock(tenant_type='account')) session_mock = mock.Mock() db_session_mock.return_value.__enter__.return_value = session_mock user_invite.create_or_update_user( admin_identity=admin_identity, assignee_identity=assignee_identity, tenant_roles_input=tenant_roles_input, brand=brand, master_contact=False, localization='es', ) identity_model_mock.create_identity.assert_called_with( session=session_mock.begin_transaction.return_value, audit_user_id=admin_identity.id, first_name=assignee_identity.first_name, last_name=assignee_identity.last_name, email=assignee_identity.email, default_brand=default_brand, tenant_type='account', is_employee=False, localization='es', ) @mock.patch('permissions.connectors.neo4j.db_session') @mock.patch('permissions.logic.user_invite.identity_model') @mock.patch('permissions.logic.user_invite.add_profiles_to_identity_from_v2_roles') def test_create_or_update_user_create_employee( _add_profiles_mock: mock.MagicMock, identity_model_mock: mock.MagicMock, db_session_mock: mock.MagicMock, ) -> None: """Test employee status is set to true when creating an identity with an employee domain.""" admin_identity = mock.Mock(id='admin-identity-id') assignee_identity = IdentityInput( first_name='Balthazar', last_name='Turtle', email='b.turtle@sony.com', ) tenant_roles_input = mock.Mock(tenant=mock.Mock(tenant_type='account')) session_mock = mock.Mock() db_session_mock.return_value.__enter__.return_value = session_mock user_invite.create_or_update_user( admin_identity=admin_identity, assignee_identity=assignee_identity, tenant_roles_input=tenant_roles_input, brand='sme', master_contact=False, localization=None, ) identity_model_mock.create_identity.assert_called_with( session=session_mock.begin_transaction.return_value, audit_user_id=admin_identity.id, first_name=assignee_identity.first_name, last_name=assignee_identity.last_name, email=assignee_identity.email, default_brand='sme', tenant_type='account', is_employee=True, localization='en', ) @mock.patch('permissions.logic.user_invite.add_profiles_to_identity_from_v2_roles') def test_update_identity_with_v2_roles(add_profiles_mock): """Test update_identity_with_v2_roles adds profiles and returns (identity_id, vend_contact).""" tx_mock = mock.Mock() vend_contact = mock.Mock() add_profiles_mock.return_value = vend_contact admin_identity = mock.Mock(id='admin-id') assignee_identity = mock.Mock(id='identity-id') tenant_roles_input = mock.Mock() result = user_invite.update_identity_with_v2_roles( tx=tx_mock, admin_identity=admin_identity, assignee_identity=assignee_identity, tenant_roles_input=tenant_roles_input, master_contact=False, ) add_profiles_mock.assert_called_once_with( tx=tx_mock, identity=assignee_identity, tenant_roles_input=tenant_roles_input, audit_user_id=admin_identity.id, master_contact=False, ) tx_mock.commit.assert_called_once() assert result == vend_contact @mock.patch('permissions.logic.user_invite.add_profiles_to_identity_from_v2_roles') @mock.patch('permissions.logic.user_invite.identity_model') def test_create_identity_with_v2_roles( identity_model_mock, add_profiles_mock, ): """Test _create_identity_with_v2_roles creates identity and returns (identity_id, vend_contact).""" admin_identity = mock.Mock(id='an-admin-uuid') tenant_roles_input = TenantRolesInput( roles_to_attach=[], roles_to_detach=[], tenant=Tenant( tenant_type=TenantType.ACCOUNT, tenant_uuid='fe215625-1e69-4b89-b30a-d9e097970a59' ), ) identity = mock.Mock(id='identity-id', email='b@b.com') identity_model_mock.create_identity.return_value = identity vend_contact = mock.Mock() add_profiles_mock.return_value = vend_contact tx_mock = mock.Mock() identity_result, returned_vc = user_invite.create_identity_with_v2_roles( tx=tx_mock, admin_identity=admin_identity, assignee_identity=mock.Mock(first_name='B', last_name='B', email='b@b.com'), tenant_roles_input=tenant_roles_input, brand='theorchard', master_contact=False, localization=None, ) assert identity_result == identity assert returned_vc == vend_contact tx_mock.commit.assert_called_once() @pytest.mark.parametrize( ['tenant_type', 'tenant_uuid', 'expected_target_tenants'], [ pytest.param( TenantType.PARENT_COMPANY, 'parent-company-uuid', [ Tenant(tenant_type=TenantType.PARENT_COMPANY, tenant_uuid='parent-company-uuid'), Tenant( tenant_type=TenantType.ACCOUNT, tenant_uuid='053a1a75-acc5-4cd8-9206-a194335d2afa', ), # VENDOR_STAR_UUID ], id='Parent company tenant type results in both parent company and vendor * access', ), pytest.param( TenantType.ACCOUNT, 'account-uuid', [Tenant(tenant_type=TenantType.ACCOUNT, tenant_uuid='account-uuid')], id='Account tenant type (not supported yet but whatever) passes through as-is', ), ], ) @mock.patch('permissions.logic.user_invite.profile_model') @mock.patch('permissions.logic.user_invite.profile_logic') @mock.patch('permissions.logic.user_invite.g', spec=['log']) def test_add_profiles_to_employee_from_v2_roles( g_mock: mock.MagicMock, profile_logic_mock: mock.MagicMock, profile_model_mock: mock.MagicMock, tenant_type: TenantType, tenant_uuid: str, expected_target_tenants: list[Tenant], ) -> None: """Test add_profiles_to_employee_from_v2_roles with different tenant types.""" profile_logic_mock.v2_roles_to_profiles_and_roles_dict.return_value = { 'SettingsProfile': [], 'InsightsProfile': ['insights'], } neo4j_tx_mock = mock.Mock() identity = VALID_IDENTITY tenant_roles_input = TenantRolesInput( tenant=Tenant(tenant_type=tenant_type, tenant_uuid=tenant_uuid), roles_to_attach=[application.SETTINGS_BASE_ROLE, application.INSIGHTS_BASE_ROLE], roles_to_detach=[], ) audit_user_id = 'audit-user-id' user_invite.add_profiles_to_employee_from_v2_roles( tx=neo4j_tx_mock, identity=VALID_IDENTITY, tenant_roles_input=tenant_roles_input, audit_user_id=audit_user_id, ) g_mock.log.info.assert_called_with( 'Adding profiles to employee identity', resources={ 'identity_id': identity.id, 'tenant_uuid': tenant_roles_input.tenant.tenant_uuid, 'roles': tenant_roles_input.roles_to_attach, 'admin_identity_id': audit_user_id, }, ) profile_logic_mock.v2_roles_to_profiles_and_roles_dict.assert_called_once_with( [application.SETTINGS_BASE_ROLE, application.INSIGHTS_BASE_ROLE] ) profile_model_mock.update_existing_profiles_with_roles.assert_called_once_with( tx=neo4j_tx_mock, identity_id=identity.id, profile_types_and_roles=[ {'profile_type': 'SettingsProfile', 'roles': []}, {'profile_type': 'InsightsProfile', 'roles': ['insights']}, ], audit_user=audit_user_id, ) profile_model_mock.create_profiles_if_not_exist.assert_called_once_with( tx=neo4j_tx_mock, identity_id=identity.id, profile_name=identity.name, profile_types_and_roles=[ {'profile_type': 'SettingsProfile', 'roles': []}, {'profile_type': 'InsightsProfile', 'roles': ['insights']}, ], audit_user=audit_user_id, ) # Build expected calls for all profiles and all tenants expected_calls = [] for profile_type in ['SettingsProfile', 'InsightsProfile']: for expected_target_tenant in expected_target_tenants: expected_calls.append( mock.call( tx=neo4j_tx_mock, identity_id=identity.id, profile_type=profile_type, tenant=expected_target_tenant, audit_user_id=audit_user_id, ) ) profile_model_mock.add_tenant_connection_to_profile_and_clear_cache.assert_has_calls( expected_calls, any_order=True ) # Verify we have the correct number of calls (2 profiles * number of tenants) assert ( profile_model_mock.add_tenant_connection_to_profile_and_clear_cache.call_count == 2 * len(expected_target_tenants) ) @mock.patch('permissions.logic.user_invite.add_profiles_to_employee_from_v2_roles') @mock.patch('permissions.logic.user_invite.identity_model') @mock.patch('permissions.logic.user_invite.neo4j_connector') @mock.patch('permissions.logic.user_invite.g', spec=['log']) def test_create_employee_success( g_mock: mock.MagicMock, neo4j_connector_mock: mock.MagicMock, identity_model_mock: mock.MagicMock, add_profiles_mock: mock.MagicMock, ) -> None: """Test successful employee creation.""" session_mock = mock.Mock() tx_mock = mock.Mock() tx_mock.closed.return_value = False session_mock.begin_transaction.return_value = tx_mock neo4j_connector_mock.db_session.return_value.__enter__.return_value = session_mock created_identity = Identity( id='new-employee-id', first_name='Hunter', last_name='Dog', name='Hunter Dog', email='hunter@pear.tree', auth0_user_id=None, user_types=['employee'], active='Y', default_brand='theorchard', ) identity_model_mock.create_identity.return_value = created_identity assignee_identity = IdentityInput( first_name='Hunter', last_name='Dog', email='hunter@pear.tree', ) admin_identity = mock.Mock(id='admin-id') tenant_roles_input = TenantRolesInput( tenant=Tenant(tenant_type=TenantType.PARENT_COMPANY, tenant_uuid='parent-uuid'), roles_to_attach=[application.SETTINGS_BASE_ROLE, application.INSIGHTS_BASE_ROLE], roles_to_detach=[], ) brand = 'theorchard' result = user_invite.create_employee( assignee_identity=assignee_identity, admin_identity=admin_identity, tenant_roles_input=tenant_roles_input, brand=brand, ) assert result == created_identity g_mock.log.info.assert_called_with( 'Creating employee identity', resources={'admin_identity_id': admin_identity.id}, ) neo4j_connector_mock.db_session.assert_called_once() session_mock.begin_transaction.assert_called_once() identity_model_mock.create_identity.assert_called_once_with( session=tx_mock, audit_user_id='admin-id', first_name='Hunter', last_name='Dog', email='hunter@pear.tree', default_brand='theorchard', tenant_type=TenantType.PARENT_COMPANY, localization='en', is_employee=True, ) add_profiles_mock.assert_called_once_with( tx=tx_mock, identity=created_identity, tenant_roles_input=tenant_roles_input, audit_user_id='admin-id', ) tx_mock.commit.assert_called_once() @mock.patch('permissions.logic.user_invite.tenant_logic') @mock.patch('permissions.logic.user_invite.add_profiles_to_employee_from_v2_roles') @mock.patch('permissions.logic.user_invite.identity_model') @mock.patch('permissions.logic.user_invite.neo4j_connector') @mock.patch('permissions.logic.user_invite.g', spec=['log']) def test_create_employee_with_account_tenant_no_brand( g_mock: mock.MagicMock, neo4j_connector_mock: mock.MagicMock, identity_model_mock: mock.MagicMock, add_profiles_mock: mock.MagicMock, tenant_logic_mock: mock.MagicMock, ) -> None: """Test employee creation with account tenant and no brand provided.""" session_mock = mock.Mock() tx_mock = mock.Mock() tx_mock.closed.return_value = False session_mock.begin_transaction.return_value = tx_mock neo4j_connector_mock.db_session.return_value.__enter__.return_value = session_mock tenant_logic_mock.get_parent_company_brand_for_tenant.return_value = 'theorchard' created_identity = Identity( id='new-employee-id', first_name='Hunter', last_name='Dog', name='Hunter Dog', email='hunter@pear.tree', auth0_user_id=None, user_types=['employee'], active='Y', default_brand='theorchard', ) identity_model_mock.create_identity.return_value = created_identity assignee_identity = IdentityInput( first_name='Hunter', last_name='Dog', email='hunter@pear.tree', ) admin_identity = mock.Mock(id='admin-id') tenant_roles_input = TenantRolesInput( tenant=Tenant(tenant_type=TenantType.ACCOUNT, tenant_uuid='account-uuid'), roles_to_attach=[application.FANSIFTER_BASE_ROLE, application.SONGWHIP_BASE_ROLE], roles_to_detach=[], ) # Call with brand=None to trigger the brand lookup logic result = user_invite.create_employee( assignee_identity=assignee_identity, admin_identity=admin_identity, tenant_roles_input=tenant_roles_input, brand=None, ) assert result == created_identity tenant_logic_mock.get_parent_company_brand_for_tenant.assert_called_once_with( tenant_roles_input.tenant ) identity_model_mock.create_identity.assert_called_once_with( session=tx_mock, audit_user_id='admin-id', first_name='Hunter', last_name='Dog', email='hunter@pear.tree', default_brand='theorchard', tenant_type=TenantType.ACCOUNT, localization='en', is_employee=True, ) add_profiles_mock.assert_called_once_with( tx=tx_mock, identity=created_identity, tenant_roles_input=tenant_roles_input, audit_user_id='admin-id', ) tx_mock.commit.assert_called_once() @mock.patch('permissions.logic.user_invite.sentry_client') @mock.patch('permissions.logic.user_invite.add_profiles_to_employee_from_v2_roles') @mock.patch('permissions.logic.user_invite.identity_model') @mock.patch('permissions.logic.user_invite.neo4j_connector') @mock.patch('permissions.logic.user_invite.g', spec=['log']) def test_create_employee_neo4j_error( g_mock: mock.MagicMock, neo4j_connector_mock: mock.MagicMock, identity_model_mock: mock.MagicMock, add_profiles_mock: mock.MagicMock, sentry_client_mock: mock.MagicMock, ) -> None: """Test employee creation with Neo4j error.""" session_mock = mock.Mock() tx_mock = mock.Mock() tx_mock.closed.return_value = False session_mock.begin_transaction.return_value = tx_mock neo4j_connector_mock.db_session.return_value.__enter__.return_value = session_mock neo4j_error = Exception('Neo4j connection error') identity_model_mock.create_identity.side_effect = neo4j_error assignee_identity = IdentityInput( first_name='Hunter', last_name='Dog', email='hunter@pear.tree', ) admin_identity = mock.Mock(id='admin-id') tenant_roles_input = TenantRolesInput( tenant=Tenant(tenant_type=TenantType.PARENT_COMPANY, tenant_uuid='parent-uuid'), roles_to_attach=[application.SETTINGS_BASE_ROLE], roles_to_detach=[], ) brand = 'theorchard' with pytest.raises(Exception) as exc_info: user_invite.create_employee( assignee_identity=assignee_identity, admin_identity=admin_identity, tenant_roles_input=tenant_roles_input, brand=brand, ) assert exc_info.value == neo4j_error tx_mock.rollback.assert_called_once() sentry_client_mock.capture_exception.assert_called_once_with(neo4j_error) tx_mock.commit.assert_not_called() add_profiles_mock.assert_not_called() @mock.patch('permissions.logic.user_invite.sentry_client') @mock.patch('permissions.logic.user_invite.add_profiles_to_employee_from_v2_roles') @mock.patch('permissions.logic.user_invite.identity_model') @mock.patch('permissions.logic.user_invite.neo4j_connector') @mock.patch('permissions.logic.user_invite.g', spec=['log']) def test_create_employee_general_exception( g_mock: mock.MagicMock, neo4j_connector_mock: mock.MagicMock, identity_model_mock: mock.MagicMock, add_profiles_mock: mock.MagicMock, sentry_client_mock: mock.MagicMock, ) -> None: """Test employee creation with non-neo4j exception.""" session_mock = mock.Mock() tx_mock = mock.Mock() tx_mock.closed.return_value = False session_mock.begin_transaction.return_value = tx_mock neo4j_connector_mock.db_session.return_value.__enter__.return_value = session_mock some_error = ValueError('🐛') add_profiles_mock.side_effect = some_error created_identity = Identity( id='new-employee-id', first_name='Hunter', last_name='Dog', name='Hunter Dog', email='hunter@pear.tree', auth0_user_id=None, user_types=['employee'], active='Y', default_brand='theorchard', ) identity_model_mock.create_identity.return_value = created_identity # Test data assignee_identity = IdentityInput( first_name='Hunter', last_name='Dog', email='hunter@pear.tree', ) admin_identity = mock.Mock(id='admin-id') tenant_roles_input = TenantRolesInput( tenant=Tenant(tenant_type=TenantType.PARENT_COMPANY, tenant_uuid='parent-uuid'), roles_to_attach=[application.SETTINGS_BASE_ROLE], roles_to_detach=[], ) brand = 'theorchard' # Call function and expect exception with pytest.raises(ValueError) as exc_info: user_invite.create_employee( assignee_identity=assignee_identity, admin_identity=admin_identity, tenant_roles_input=tenant_roles_input, brand=brand, ) assert exc_info.value == some_error # Assertions tx_mock.rollback.assert_called_once() sentry_client_mock.capture_exception.assert_called_once_with(some_error) tx_mock.commit.assert_not_called() @pytest.mark.parametrize('active', ['Y', 'N']) @mock.patch('permissions.logic.user_invite.default_brand') @mock.patch('permissions.logic.user_invite.identity_logic') @mock.patch('permissions.logic.user_invite.update_identity_with_v2_roles') @mock.patch('permissions.connectors.neo4j.db_session') def test_create_or_update_user_update_reactivates_and_updates_default_brand( db_session_mock: mock.MagicMock, _update_identity_mock: mock.MagicMock, identity_logic_mock: mock.MagicMock, default_brand_mock: mock.MagicMock, active: str, ) -> None: """Test create_or_update_user reactivates user and updates default brand when updating.""" admin_id = 'admin-id-123' admin = mock.MagicMock(id=admin_id) identity = mock.MagicMock(id='identity-id-123', active=active) _update_identity_mock.return_value = None identity_logic_mock.reactivate_user_if_needed.return_value = active != 'Y' # Add transaction mock setup transaction_mock = mock.MagicMock() session_mock = mock.MagicMock() session_mock.begin_transaction.return_value = transaction_mock db_session_mock.return_value.__enter__.return_value = session_mock tenant_roles_input = mock.MagicMock(roles_to_attach=['SETTINGS_BASE_ROLE']) user_invite.create_or_update_user( admin_identity=admin, assignee_identity=identity, tenant_roles_input=tenant_roles_input, brand='theorchard', master_contact=False, localization=None, ) if active != 'Y': identity_logic_mock.reactivate_user_if_needed.assert_called_once_with( admin=admin, identity_with_auth0=identity, ) default_brand_mock.update_default_brand_if_needed.assert_called_once_with( identity=identity, admin_identity_id=admin_id, company_brand='theorchard', ) else: identity_logic_mock.reactivate_user_if_needed.assert_called_once_with( admin=admin, identity_with_auth0=identity, ) default_brand_mock.update_default_brand_if_needed.assert_not_called() @mock.patch('permissions.logic.user_invite.identity_logic') @mock.patch('permissions.logic.user_invite.sentry_client') @mock.patch('permissions.connectors.neo4j.db_session') def test_create_or_update_user_update_raises_if_reactivate_fails( _db_session_mock: mock.MagicMock, sentry_client_mock: mock.MagicMock, identity_logic_mock: mock.MagicMock, ) -> None: """Test create_or_update_user raises if reactivating user fails when updating.""" identity = mock.MagicMock(id='identity-id-123', active='N') # Simulate reactivation failure by raising RuntimeError identity_logic_mock.reactivate_user_if_needed.side_effect = RuntimeError( 'Failed to reactivate user during update' ) tenant_roles_input = mock.MagicMock(roles_to_attach=['SETTINGS_BASE_ROLE']) with pytest.raises(RuntimeError) as e: user_invite.create_or_update_user( admin_identity=mock.MagicMock(), assignee_identity=identity, tenant_roles_input=tenant_roles_input, brand='theorchard', master_contact=False, localization=None, ) assert str(e.value) == 'Failed to reactivate user during update' sentry_client_mock.capture_exception.assert_called_once() @mock.patch('permissions.logic.user_invite.g', spec=['log']) @mock.patch('permissions.logic.user_invite.default_brand') @mock.patch('permissions.logic.user_invite.identity_logic') @mock.patch('permissions.logic.user_invite.update_identity_with_v2_roles') @mock.patch('permissions.connectors.neo4j.db_session') def test_create_or_update_user_update_does_not_raise_if_default_brand_fails( _db_session_mock: mock.MagicMock, _update_identity_mock: mock.MagicMock, identity_logic_mock: mock.MagicMock, default_brand_mock: mock.MagicMock, g_mock: mock.MagicMock, ): """Test create_or_update_user does not raise if updating default brand fails when updating.""" identity = mock.MagicMock(id='identity-id-123', active='N') _update_identity_mock.return_value = None identity_logic_mock.reactivate_user_if_needed.return_value = True default_brand_mock.update_default_brand_if_needed.side_effect = Exception('🫩') tenant_roles_input = mock.MagicMock(roles_to_attach=['SETTINGS_BASE_ROLE']) user_invite.create_or_update_user( admin_identity=mock.MagicMock(), assignee_identity=identity, tenant_roles_input=tenant_roles_input, brand='theorchard', master_contact=False, localization=None, ) g_mock.log.error.assert_called_once_with( 'Error updating default brand after reactivation--user is still reactivated', resources={ 'identity_id': 'identity-id-123', 'error': '🫩', }, )