"""Unit tests for the employee-related functions in the user_update logic module.""" from unittest import mock import neo4j.exceptions import pytest from permissions import types from permissions.constants import constants, parent_companies from permissions.exceptions.incomplete_result_error import IncompleteResultError from permissions.logic import user_invite, user_update @mock.patch('permissions.logic.user_invite.add_profiles_to_employee_from_v2_roles') def test_update_employee_with_v2_roles( add_profiles_mock: mock.MagicMock, ) -> None: """Test update_employee_with_v2_roles calls expected functions.""" tx_mock = mock.MagicMock() admin_identity = mock.MagicMock(id='admin-id-123') assignee_identity = mock.MagicMock(id='assignee-id-456') tenant_roles_input = types.TenantRolesInput( roles_to_attach=['SETTINGS_BASE_ROLE'], roles_to_detach=[], tenant=types.Tenant( tenant_type=types.TenantType.PARENT_COMPANY, tenant_uuid=parent_companies.ORCHARD_PARENT_COMPANY_UUID, ), ) result = user_invite.update_employee_with_v2_roles( tx=tx_mock, admin_identity=admin_identity, assignee_identity=assignee_identity, tenant_roles_input=tenant_roles_input, ) 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, ) assert result == assignee_identity.id @pytest.mark.parametrize( ['tenant_roles_input', 'has_roles_to_attach', 'has_roles_to_detach'], [ pytest.param( types.TenantRolesInput( roles_to_attach=['SETTINGS_BASE_ROLE', 'INSIGHTS_BASE_ROLE'], roles_to_detach=[], tenant=types.Tenant( tenant_type=types.TenantType.PARENT_COMPANY, tenant_uuid=parent_companies.ORCHARD_PARENT_COMPANY_UUID, ), ), True, False, id='Roles to attach but not detach', ), pytest.param( types.TenantRolesInput( roles_to_attach=[], roles_to_detach=['SETTINGS_BASE_ROLE'], tenant=types.Tenant( tenant_type=types.TenantType.PARENT_COMPANY, tenant_uuid=parent_companies.ORCHARD_PARENT_COMPANY_UUID, ), ), False, True, id='Roles to detach but not attach', ), pytest.param( types.TenantRolesInput( roles_to_attach=['INSIGHTS_BASE_ROLE'], roles_to_detach=['SETTINGS_BASE_ROLE'], tenant=types.Tenant( tenant_type=types.TenantType.PARENT_COMPANY, tenant_uuid=parent_companies.ORCHARD_PARENT_COMPANY_UUID, ), ), True, True, id='Roles to attach and detach', ), ], ) @mock.patch('permissions.logic.user_update.detach_v2_roles') @mock.patch('permissions.logic.user_invite.update_employee_with_v2_roles') @mock.patch('permissions.connectors.neo4j.db_session') def test_update_employee( db_session_mock: mock.MagicMock, update_employee_with_v2_roles_mock: mock.MagicMock, detach_roles_mock: mock.MagicMock, tenant_roles_input: types.TenantRolesInput, has_roles_to_attach: bool, has_roles_to_detach: bool, ): """Test update_employee behavior when there are/aren't roles to attach/detach.""" admin = mock.MagicMock(id='admin-id-123') identity = mock.MagicMock(id='identity-id-456', active='Y') transaction_mock = mock.MagicMock() enter_mock = mock.MagicMock() enter_mock.begin_transaction.return_value = transaction_mock session_mock = mock.MagicMock() session_mock.__enter__.return_value = enter_mock db_session_mock.return_value = session_mock user_update.update_employee( admin=admin, identity_with_auth0=identity, tenant_roles_input=tenant_roles_input, ) enter_mock.begin_transaction.assert_called_once() transaction_mock.commit.assert_called_once() if has_roles_to_attach: update_employee_with_v2_roles_mock.assert_called_once_with( tx=transaction_mock, admin_identity=admin, assignee_identity=identity, tenant_roles_input=tenant_roles_input, ) else: update_employee_with_v2_roles_mock.assert_not_called() if has_roles_to_detach: detach_roles_mock.assert_called_once_with( tx=transaction_mock, roles=tenant_roles_input.roles_to_detach, identity=identity, admin_identity=admin, tenant=tenant_roles_input.tenant, is_employee=True, ) else: detach_roles_mock.assert_not_called() @pytest.mark.parametrize( ['error', 'error_on_operation', 'roles_to_attach', 'roles_to_detach'], [ pytest.param( neo4j.exceptions.ClientError('🫩'), 'attach', ['SETTINGS_BASE_ROLE'], [], id='ClientError on attach', ), pytest.param( neo4j.exceptions.ClientError('🤬'), 'detach', ['SETTINGS_BASE_ROLE'], ['INSIGHTS_BASE_ROLE'], id='ClientError on detach', ), pytest.param( IncompleteResultError('😱'), 'detach', [], ['INSIGHTS_BASE_ROLE'], id='IncompleteResultError on detach', ), ], ) @mock.patch('permissions.logic.user_update.g', spec=['log']) @mock.patch('permissions.logic.user_update.detach_v2_roles') @mock.patch('permissions.logic.user_invite.update_employee_with_v2_roles') @mock.patch('permissions.connectors.neo4j.db_session') def test_update_employee_error_rollback( db_session_mock: mock.MagicMock, update_employee_with_v2_roles_mock: mock.MagicMock, detach_roles_mock: mock.MagicMock, g_mock: mock.MagicMock, error: Exception, error_on_operation: str, roles_to_attach: list[str], roles_to_detach: list[str], ): """Test update_employee rolls back on various errors.""" if error_on_operation == 'attach': update_employee_with_v2_roles_mock.side_effect = error else: detach_roles_mock.side_effect = error transaction_mock = mock.MagicMock() db_session_mock.return_value.__enter__.return_value.begin_transaction.return_value = ( transaction_mock ) tenant_roles_input = types.TenantRolesInput( roles_to_attach=roles_to_attach, roles_to_detach=roles_to_detach, tenant=types.Tenant( tenant_type=types.TenantType.PARENT_COMPANY, tenant_uuid=parent_companies.ORCHARD_PARENT_COMPANY_UUID, ), ) with pytest.raises(type(error)) as e: user_update.update_employee( admin=mock.MagicMock(), identity_with_auth0=mock.MagicMock(active='Y'), tenant_roles_input=tenant_roles_input, ) transaction_mock.rollback.assert_called_once() transaction_mock.commit.assert_not_called() assert e.value == error @pytest.mark.parametrize( ['active', 'should_call_activate'], [ pytest.param('Y', False, id='Already active user'), pytest.param('N', True, id='Reactivate inactive user'), ], ) @mock.patch('permissions.logic.user_update.default_brand') @mock.patch('permissions.logic.user_update.identity_logic') @mock.patch('permissions.logic.user_invite.update_employee_with_v2_roles') @mock.patch('permissions.connectors.neo4j.db_session') def test_update_employee_reactivation( db_session_mock: mock.MagicMock, _update_employee_with_v2_roles_mock: mock.MagicMock, identity_logic_mock: mock.MagicMock, default_brand_mock: mock.MagicMock, active: str, should_call_activate: bool, ): """Test update_employee reactivation behavior.""" identity_logic_mock.reactivate_user_if_needed.return_value = should_call_activate admin = mock.MagicMock(id='admin-id-123') admin.settings_profile.profile_id = 'profile-id-789' identity = mock.MagicMock(id='identity-id-456', active=active) brand = constants.THEORCHARD_BRAND transaction_mock = mock.MagicMock() enter_mock = mock.MagicMock() enter_mock.begin_transaction.return_value = transaction_mock session_mock = mock.MagicMock() session_mock.__enter__.return_value = enter_mock db_session_mock.return_value = session_mock tenant_roles_input = types.TenantRolesInput( roles_to_attach=['SETTINGS_BASE_ROLE'], roles_to_detach=[], tenant=types.Tenant( tenant_type=types.TenantType.PARENT_COMPANY, tenant_uuid=parent_companies.ORCHARD_PARENT_COMPANY_UUID, ), ) user_update.update_employee( admin=admin, identity_with_auth0=identity, tenant_roles_input=tenant_roles_input, brand=brand, ) if should_call_activate: 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=brand, ) 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() transaction_mock.commit.assert_called_once() @mock.patch('permissions.logic.user_update.identity_logic') @mock.patch('permissions.connectors.neo4j.db_session') def test_update_employee_reactivation_failure( db_session_mock: mock.MagicMock, identity_logic_mock: mock.MagicMock, ): """Test update_employee raises error when reactivation fails.""" identity_logic_mock.reactivate_user_if_needed.side_effect = RuntimeError( 'Failed to reactivate user during update' ) admin = mock.MagicMock(id='admin-id-123') admin.settings_profile.profile_id = 'profile-id-789' identity = mock.MagicMock(id='identity-id-456', active='N') brand = constants.THEORCHARD_BRAND transaction_mock = mock.MagicMock() enter_mock = mock.MagicMock() enter_mock.begin_transaction.return_value = transaction_mock session_mock = mock.MagicMock() session_mock.__enter__.return_value = enter_mock db_session_mock.return_value = session_mock tenant_roles_input = types.TenantRolesInput( roles_to_attach=['SETTINGS_BASE_ROLE'], roles_to_detach=[], tenant=types.Tenant( tenant_type=types.TenantType.PARENT_COMPANY, tenant_uuid=parent_companies.ORCHARD_PARENT_COMPANY_UUID, ), ) with pytest.raises(RuntimeError, match='Failed to reactivate user during update'): user_update.update_employee( admin=admin, identity_with_auth0=identity, tenant_roles_input=tenant_roles_input, brand=brand, ) identity_logic_mock.reactivate_user_if_needed.assert_called_once_with( admin=admin, identity_with_auth0=identity, )