"""Tests for the user_revoke logic module.""" from unittest import mock from unittest.mock import MagicMock, patch import flask import pymysql import pytest import sqlalchemy from neo4j import exceptions as neo4j_exceptions from permissions.connectors import mysql, neo4j from permissions.constants import constants from permissions.exceptions.incomplete_result_error import IncompleteResultError from permissions.logic import default_brand, user_revoke, vend_contact as vend_contact_logic from permissions.models import ( auth0 as auth0_model, identity as identity_model, tenant as tenant_model, vend_contact as vend_contact_model, ) from permissions.types import ( AdminableTenant, Tenant, TenantType, ) from tests.unit.conftest import get_transactional_session_mock @pytest.mark.parametrize( 'tenant_mock', [ Tenant(tenant_uuid='abc-123', tenant_type=TenantType.ACCOUNT), Tenant(tenant_uuid='abc-123', tenant_type=TenantType.SUBACCOUNT), Tenant(tenant_uuid='abc-123', tenant_type=TenantType.LABEL_PARTICIPANT), Tenant(tenant_uuid='abc-123', tenant_type=TenantType.COLLABORATOR), ], ) def test_revoke_access_to_tenant_for_identity(tenant_mock): """Test revoke access to a single tenant for identity.""" identity_id = 'test-user-123' admin_identity_id = 'admin-user-123' admin_profile_id = 100100 auth0_user_id = 'auth0-user-123' session_mock = get_transactional_session_mock() mysql_session_mock = MagicMock() identity_mock = MagicMock(id=identity_id, auth0_user_id=auth0_user_id) with patch.object(neo4j, 'db_session', return_value=session_mock), patch.object( mysql, 'db_session', return_value=mysql_session_mock ), patch.object( identity_model, 'get_identity_by_id_new', return_value=identity_mock, ), patch.object(vend_contact_model, 'deactivate_by_tenant_for_identity'), patch.object( tenant_model, 'soft_delete_access_to_tenant_for_identity' ), patch.object(vend_contact_logic, 'set_new_auth0_vend_contact_id'), patch.object( tenant_model, 'get_identity_tenant_count', return_value=1 ), patch.object(default_brand, 'update_default_brand_if_needed'): user_revoke.revoke_access_to_tenant_for_identity( admin_identity_id, admin_profile_id, identity_id, tenant_mock, ) neo4j_tx_mock = session_mock.__enter__().begin_transaction() identity_model.get_identity_by_id_new.assert_called_with(identity_id) tenant_model.soft_delete_access_to_tenant_for_identity.assert_called_with( tx=neo4j_tx_mock, admin_identity_id=admin_identity_id, identity_id=identity_id, tenant=tenant_mock, ) # Check that defaultBrand logic was called default_brand.update_default_brand_if_needed.assert_called_with( identity_mock, admin_identity_id ) if tenant_mock.tenant_type in (TenantType.ACCOUNT, TenantType.SUBACCOUNT): vend_contact_model.deactivate_by_tenant_for_identity.assert_called_with( tx=mysql_session_mock.__enter__.return_value, identity=identity_mock, tenant=tenant_mock, ) vend_contact_logic.set_new_auth0_vend_contact_id.assert_called_with( session=mysql_session_mock.__enter__.return_value, identity=identity_mock, admin_id=admin_identity_id, admin_profile_id=admin_profile_id, ) else: vend_contact_model.deactivate_by_tenant_for_identity.assert_not_called() vend_contact_logic.set_new_auth0_vend_contact_id.assert_not_called() @mock.patch('permissions.logic.user_revoke.g') def test_revoke_access_to_tenant_for_identity_and_deactivate_user( g_mock: MagicMock, app_context: flask.ctx.AppContext, ) -> None: """Test revoke access to a single tenant for identity and deactivate user.""" identity_id = 'test-user-123' admin_identity_id = 'admin-user-123' admin_profile_id = 100100 auth0_user_id = 'auth0-user123' tenant_uuid = 'abc-123' session_mock = get_transactional_session_mock() mysql_session_mock = MagicMock() tenant_mock = Tenant(tenant_uuid=tenant_uuid, tenant_type=TenantType.ACCOUNT) identity_mock = mock.Mock(auth0_user_id=auth0_user_id, id=identity_id) identity_mock.is_pending.return_value = False with patch.object(neo4j, 'db_session', return_value=session_mock), patch.object( mysql, 'db_session', return_value=mysql_session_mock ), patch.object( identity_model, 'get_identity_by_id_new', return_value=identity_mock, ), patch.object(vend_contact_model, 'deactivate_by_tenant_for_identity'), patch.object( tenant_model, 'soft_delete_access_to_tenant_for_identity' ), patch.object(vend_contact_logic, 'set_new_auth0_vend_contact_id'), patch.object( tenant_model, 'get_identity_tenant_count', return_value=0 ), patch.object(identity_model, 'update_identity_active_status'), patch.object( auth0_model, 'activate_deactivate_user' ), patch.object(default_brand, 'update_default_brand_if_needed'): user_revoke.revoke_access_to_tenant_for_identity( admin_identity_id, admin_profile_id, identity_id, tenant_mock ) neo4j_tx_mock = session_mock.__enter__().begin_transaction() identity_model.get_identity_by_id_new.assert_called_with(identity_id) tenant_model.soft_delete_access_to_tenant_for_identity.assert_called_with( tx=neo4j_tx_mock, admin_identity_id=admin_identity_id, identity_id=identity_id, tenant=tenant_mock, ) vend_contact_model.deactivate_by_tenant_for_identity.assert_called_with( tx=mysql_session_mock.__enter__.return_value, identity=identity_mock, tenant=tenant_mock, ) tenant_model.get_identity_tenant_count.assert_called_with(identity_id) vend_contact_logic.set_new_auth0_vend_contact_id.assert_not_called() g_mock.log.info.assert_called_with( 'Deactivating identity because access to last tenant has been removed', resources={ 'identity_id': identity_id, 'tenant_uuid': tenant_uuid, }, ) identity_model.update_identity_active_status.assert_called_with( session=session_mock.__enter__.return_value, identity_id=identity_id, active='N', audit_user_id=admin_identity_id, ) auth0_model.activate_deactivate_user.assert_called_with( auth0_id=auth0_user_id, active=False ) @mock.patch('permissions.logic.user_revoke.g') def test_revoke_access_to_tenant_updates_default_brand_error( g_mock: MagicMock, app_context: flask.ctx.AppContext, ) -> None: """Test that errors updating defaultBrand are logged but don't fail the revocation.""" identity_id = 'test-user-123' admin_identity_id = 'admin-user-123' admin_profile_id = 100100 auth0_user_id = 'auth0-user-123' tenant_uuid = 'abc-123' tenant_mock = Tenant(tenant_uuid=tenant_uuid, tenant_type=TenantType.ACCOUNT) session_mock = get_transactional_session_mock() mysql_session_mock = MagicMock() identity_mock = MagicMock(id=identity_id, auth0_user_id=auth0_user_id) err = IncompleteResultError(message='Failed to update Identity') with patch.object(neo4j, 'db_session', return_value=session_mock), patch.object( mysql, 'db_session', return_value=mysql_session_mock ), patch.object( identity_model, 'get_identity_by_id_new', return_value=identity_mock ), patch.object(vend_contact_model, 'deactivate_by_tenant_for_identity'), patch.object( tenant_model, 'soft_delete_access_to_tenant_for_identity' ), patch.object(vend_contact_logic, 'set_new_auth0_vend_contact_id'), patch.object( tenant_model, 'get_identity_tenant_count', return_value=1 ), patch.object(default_brand, 'update_default_brand_if_needed', side_effect=err): # Should not raise - error is caught and logged user_revoke.revoke_access_to_tenant_for_identity( admin_identity_id, admin_profile_id, identity_id, tenant_mock ) # Verify error was logged g_mock.log.error.assert_called_with( 'Error updating default brand after tenant revocation--access has still been removed', resources={ 'identity_id': identity_id, 'tenant_uuid': tenant_uuid, }, ) # Verify the access revocation still happened tenant_model.soft_delete_access_to_tenant_for_identity.assert_called_once() @mock.patch('permissions.logic.user_revoke.g') @mock.patch('permissions.logic.user_revoke.identity_logic') @mock.patch('permissions.logic.user_revoke.tenant_model') @mock.patch('permissions.logic.user_revoke.neo4j_connector.db_session') @mock.patch('permissions.logic.user_revoke.soft_delete_access') @mock.patch('permissions.logic.user_revoke.identity_model') def test_revoke_access_to_tenant_for_identity_and_deactivate_error( identity_model_mock: MagicMock, _: MagicMock, db_session_mock: MagicMock, tenant_model_mock: MagicMock, identity_logic_mock: MagicMock, g_mock: MagicMock, app_context: flask.ctx.AppContext, ) -> None: """Test revoke access to single tenant and deactivate when there's an error deactivating.""" identity_id = 'test-user-123' tenant_uuid = 'abc-123' identity_model_mock.get_identity_by_id_new.return_value = mock.Mock(id=identity_id) tenant_model_mock.get_identity_tenant_count.return_value = 0 db_session_mock.return_value.__enter__.return_value = mock.Mock() error = neo4j_exceptions.Neo4jError('🫥') identity_logic_mock.deactivate_identity.side_effect = error with pytest.raises(neo4j_exceptions.Neo4jError): user_revoke.revoke_access_to_tenant_for_identity( admin_identity_id='admin-user-123', admin_profile_id=100100, identity_id=identity_id, tenant=mock.Mock(tenant_uuid=tenant_uuid, tenant_type=TenantType.ACCOUNT), ) g_mock.log.error.assert_called_with( 'Error deactivating identity during revocation of tenant access' '--access has still been removed', resources={ 'identity_id': identity_id, 'tenant_uuid': tenant_uuid, 'error': str(error), }, ) @mock.patch('permissions.logic.user_revoke.g') @mock.patch('permissions.logic.user_revoke.identity_logic') @mock.patch('permissions.logic.user_revoke.tenant_model') @mock.patch('permissions.logic.user_revoke.neo4j_connector.db_session') @mock.patch('permissions.logic.user_revoke.soft_delete_access') @mock.patch('permissions.logic.user_revoke.identity_model') def test_revoke_access_to_tenant_for_identity_last_tenant_no_deactivate( identity_model_mock: MagicMock, soft_delete_mock: MagicMock, db_session_mock: MagicMock, tenant_model_mock: MagicMock, identity_logic_mock: MagicMock, g_mock: MagicMock, app_context: flask.ctx.AppContext, ) -> None: """When deactivate_if_last_tenant is False, the last tenant is soft-deleted but the identity is left active (no deactivation) even with zero remaining tenants. """ identity_id = 'test-user-123' tenant_uuid = 'abc-123' tenant_mock = Tenant(tenant_uuid=tenant_uuid, tenant_type=TenantType.ACCOUNT) identity_model_mock.get_identity_by_id_new.return_value = mock.Mock(id=identity_id) tenant_model_mock.get_identity_tenant_count.return_value = 0 user_revoke.revoke_access_to_tenant_for_identity( admin_identity_id='admin-user-123', admin_profile_id=100100, identity_id=identity_id, tenant=tenant_mock, deactivate_if_last_tenant=False, ) soft_delete_mock.assert_called_once() identity_logic_mock.deactivate_identity.assert_not_called() db_session_mock.assert_not_called() @patch('permissions.logic.user_revoke.g') def test_revoke_access_to_tenant_for_identity__soft_delete_error( g_mock: MagicMock, app_context: flask.ctx.AppContext, ) -> None: """Test soft delete for revoke access to a single tenant.""" identity_id = 'test-user-123' admin_identity_id = 'admin-user-123' auth0_user_id = 'auth0-user123' tenant_uuid = 'abc-123' tenant_mock = Tenant(tenant_uuid=tenant_uuid, tenant_type=TenantType.ACCOUNT) session_mock = get_transactional_session_mock() mysql_session_mock = MagicMock() err = IncompleteResultError(message='error') with patch.object(neo4j, 'db_session', return_value=session_mock), patch.object( mysql, 'db_session', return_value=mysql_session_mock ), patch.object( identity_model, 'get_identity_by_id_new', return_value=mock.Mock(auth0_user_id=auth0_user_id, id=identity_id), ), patch.object(vend_contact_model, 'deactivate_by_tenant_for_identity'), patch.object( tenant_model, 'soft_delete_access_to_tenant_for_identity', side_effect=err ): with pytest.raises(IncompleteResultError): user_revoke.revoke_access_to_tenant_for_identity( admin_identity_id, 100100, identity_id, tenant_mock ) assert tenant_model.soft_delete_access_to_tenant_for_identity.called vend_contact_model.deactivate_by_tenant_for_identity.assert_not_called() g_mock.log.error.assert_called_with( 'Error soft-deleting access to tenant', resources={ 'identity_id': identity_id, 'tenant_uuid': tenant_uuid, 'error': 'error', }, ) @mock.patch('permissions.logic.user_revoke.g') def test_revoke_access_to_tenant_for_identity__vend_contact_error( g_mock: MagicMock, app_context: flask.ctx.AppContext, ) -> None: """Test vend contact for revoke access to a single tenant.""" identity_id = 'test-user-123' admin_identity_id = 'admin-user-123' auth0_user_id = 'auth0-user123' tenant_uuid = 'abc-123' tenant_mock = Tenant(tenant_uuid=tenant_uuid, tenant_type=TenantType.ACCOUNT) session_mock = get_transactional_session_mock() mysql_session_mock = MagicMock() err = sqlalchemy.exc.SQLAlchemyError('🫠') with patch.object(neo4j, 'db_session', return_value=session_mock), patch.object( mysql, 'db_session', return_value=mysql_session_mock ), patch.object( identity_model, 'get_identity_by_id_new', return_value=mock.Mock(auth0_user_id=auth0_user_id, id=identity_id), ), patch.object( vend_contact_model, 'deactivate_by_tenant_for_identity', side_effect=err ), patch.object(vend_contact_logic, 'set_new_auth0_vend_contact_id'), patch.object( tenant_model, 'soft_delete_access_to_tenant_for_identity' ): with pytest.raises(sqlalchemy.exc.SQLAlchemyError): user_revoke.revoke_access_to_tenant_for_identity( admin_identity_id, 100100, identity_id, tenant_mock ) assert tenant_model.soft_delete_access_to_tenant_for_identity.called assert vend_contact_model.deactivate_by_tenant_for_identity.called session_mock.__enter__().begin_transaction.return_value.rollback.assert_called() vend_contact_logic.set_new_auth0_vend_contact_id.assert_not_called() g_mock.log.error.assert_called_with( 'Error soft-deleting access to tenant', resources={ 'identity_id': identity_id, 'tenant_uuid': tenant_uuid, 'error': '🫠', }, ) @mock.patch('permissions.logic.user_revoke.g') @mock.patch('permissions.logic.user_revoke.vend_contact_logic') @mock.patch('permissions.logic.user_revoke.tenant_model') @mock.patch('permissions.logic.user_revoke.mysql.db_session') @mock.patch('permissions.logic.user_revoke.soft_delete_access') @mock.patch('permissions.logic.user_revoke.identity_model') @mock.patch('permissions.logic.user_revoke.default_brand_logic') def test_revoke_access_to_tenant_for_identity__set_auth0_metadata_error( _default_brand_mock: MagicMock, identity_model_mock: MagicMock, _: MagicMock, db_session_mock: MagicMock, tenant_model_mock: MagicMock, vend_contact_logic_mock: MagicMock, g_mock: MagicMock, app_context: flask.ctx.AppContext, ) -> None: """Test set_auth0_metadata for revoke access to a single tenant when an auth0 error happens.""" identity_id = 'test-user-123' tenant_uuid = 'abc-123' identity_model_mock.get_identity_by_id_new.return_value = mock.Mock(id=identity_id) tenant_model_mock.get_identity_tenant_count.return_value = 1 tenant_model_mock.get_brands_for_identity.return_value = ['theorchard', 'sme'] db_session_mock.return_value.__enter__.return_value = mock.Mock() err = auth0_model.Auth0Error(404, '💔', '😳') vend_contact_logic_mock.set_new_auth0_vend_contact_id.side_effect = err with pytest.raises(auth0_model.Auth0Error): user_revoke.revoke_access_to_tenant_for_identity( admin_identity_id='admin-user-123', admin_profile_id=100100, identity_id=identity_id, tenant=mock.Mock(tenant_uuid=tenant_uuid, tenant_type=TenantType.ACCOUNT), ) g_mock.log.error.assert_called_with( 'Error setting new auth0 vend contact id during revocation of tenant access' '--access has still been removed', resources={ 'identity_id': identity_id, 'tenant_uuid': tenant_uuid, 'error': str(err), }, ) @mock.patch('permissions.logic.user_revoke.g') @mock.patch('permissions.logic.user_revoke.neo4j_connector.db_session') @mock.patch('permissions.logic.user_revoke.tenant_model') def test_revoke_access_to_all_tenants_for_identity_neo_error( tenant_model_mock: MagicMock, neo_session_mock: MagicMock, g_mock: MagicMock, app_context: flask.ctx.AppContext, ) -> None: """Test revoke_access_to_all_tenants_for_identity behavior when neo4j raises an error.""" # we need to mock adminable tenants since the method iterates over them. tenant_model_mock.get_adminable_tenants_for_identity.return_value = [ AdminableTenant( tenant=Tenant(tenant_uuid='a-uuid', tenant_type=TenantType.ACCOUNT), profiles=[] ) ] tx_mock = mock.Mock() neo_session_mock.return_value.__enter__.return_value.begin_transaction.return_value = tx_mock err = neo4j_exceptions.Neo4jError('☹️') tenant_model_mock.soft_delete_access_to_multiple_tenants_for_identity.side_effect = err identity = mock.Mock(id='identity-uuid') with pytest.raises(neo4j_exceptions.Neo4jError): user_revoke.revoke_access_to_all_tenants_for_identity( admin_id='1', admin_profile_id=1, identity=identity ) tx_mock.rollback.assert_called() g_mock.log.error.assert_called_with( 'Error soft-deleting access to all tenants', resources={ 'identity_id': 'identity-uuid', 'tenant_uuids': ['a-uuid'], 'error': str(err), }, ) @mock.patch('permissions.logic.user_revoke.g') @mock.patch('permissions.models.vend_contact.deactivate_by_tenant_for_identity') @mock.patch('permissions.logic.user_revoke.mysql.db_session') @mock.patch('permissions.logic.user_revoke.neo4j_connector.db_session') @mock.patch('permissions.logic.user_revoke.tenant_model') def test_revoke_access_to_all_tenants_for_identity_mysql_error( tenant_model_mock: MagicMock, neo_session_mock: MagicMock, mysql_session_mock: MagicMock, deactivate_vend_contact_mock: MagicMock, g_mock: MagicMock, app_context: flask.ctx.AppContext, ) -> None: """Test revoke_access_to_all_tenants_for_identity behavior when mysql raises an error.""" tenant_model_mock.get_adminable_tenants_for_identity.return_value = [ AdminableTenant( tenant=Tenant(tenant_uuid='a-uuid', tenant_type=TenantType.ACCOUNT), profiles=[mock.Mock(profile_type=constants.LABELPROFILE)], ) ] tx_mock = mock.Mock() neo_session_mock.return_value.__enter__.return_value.begin_transaction.return_value = tx_mock session_mock = mock.Mock() mysql_session_mock.return_value.__enter__.return_value = session_mock err = pymysql.err.DatabaseError('🫠') deactivate_vend_contact_mock.side_effect = err identity = mock.Mock(id='identity-uuid') with pytest.raises(pymysql.err.DatabaseError): user_revoke.revoke_access_to_all_tenants_for_identity( admin_id='1', admin_profile_id=1, identity=identity ) # MySQL rollback happens automatically via context manager tx_mock.commit.assert_not_called() tx_mock.rollback.assert_called_once() g_mock.log.error.assert_called_with( 'Error soft-deleting access to all tenants', resources={ 'identity_id': 'identity-uuid', 'tenant_uuids': ['a-uuid'], 'error': str(err), }, ) @pytest.mark.parametrize('remaining_tenant_count', [0, 1]) @mock.patch('permissions.logic.user_revoke.g') @mock.patch('permissions.logic.identity.deactivate_identity') @mock.patch('permissions.logic.user_revoke.default_brand_logic.update_default_brand_if_needed') @mock.patch('permissions.logic.vend_contact.set_new_auth0_vend_contact_id') @mock.patch('permissions.models.vend_contact.deactivate_by_tenant_for_identity') @mock.patch('permissions.logic.user_revoke.mysql.db_session') @mock.patch('permissions.logic.user_revoke.neo4j_connector.db_session') @mock.patch('permissions.logic.user_revoke.tenant_model') def test_revoke_access_to_all_tenants_for_identity( tenant_model_mock: MagicMock, neo_session_mock: MagicMock, mysql_session_mock: MagicMock, vend_contact_deactivate_mock: MagicMock, set_new_auth0_vend_contact_id_mock: MagicMock, update_default_brand_mock: MagicMock, deactivate_identity_mock: MagicMock, g_mock: MagicMock, remaining_tenant_count: int, app_context: flask.ctx.AppContext, ) -> None: """Test revoke_access_to_all_tenants_for_identity behavior.""" tenant_uuid = 'a-uuid' tenant_type = TenantType.ACCOUNT tenant1 = Tenant(tenant_uuid=tenant_uuid, tenant_type=tenant_type) tenant2 = Tenant(tenant_uuid='other', tenant_type=tenant_type) adminable_tenants = [ AdminableTenant(tenant=tenant1, profiles=[mock.Mock(profile_type=constants.LABELPROFILE)]), AdminableTenant( tenant=tenant2, profiles=[mock.Mock(profile_type=constants.INSIGHTSPROFILE)] ), ] tenant_model_mock.get_adminable_tenants_for_identity.return_value = adminable_tenants tx_mock = mock.Mock() neo_session_mock.return_value.__enter__.return_value.begin_transaction.return_value = tx_mock session_mock = mock.Mock() mysql_session_mock.return_value.__enter__.return_value = session_mock tenant_model_mock.get_identity_tenant_count.return_value = remaining_tenant_count admin_id = 'admin-uuid' admin_profile_id = 1 identity_id = 'identity-uuid' auth0_user_id = 'somehexvalue' mock_identity = mock.Mock(id=identity_id, auth0_user_id=auth0_user_id) user_revoke.revoke_access_to_all_tenants_for_identity( admin_id=admin_id, admin_profile_id=admin_profile_id, identity=mock_identity, ) tenant_model_mock.soft_delete_access_to_multiple_tenants_for_identity.assert_called_with( tx=tx_mock, tenants=[tenant1, tenant2], admin_id=admin_id, identity_id=identity_id ) vend_contact_deactivate_mock.assert_called_once_with( tx=session_mock, identity=mock_identity, tenant=tenant1 ) set_new_auth0_vend_contact_id_mock.assert_called_with( session=session_mock, identity=mock_identity, admin_id=admin_id, admin_profile_id=admin_profile_id, ) if remaining_tenant_count: deactivate_identity_mock.assert_not_called() update_default_brand_mock.assert_called_once_with(mock_identity, admin_id) else: update_default_brand_mock.assert_not_called() g_mock.log.info.assert_called_with( 'Deactivating identity because access to all tenants has been removed', resources={'identity_id': identity_id}, ) deactivate_identity_mock.assert_called_with( session=neo_session_mock.return_value.__enter__.return_value, identity=mock_identity, admin_id=admin_id, ) tx_mock.commit.assert_called() @mock.patch('permissions.logic.user_revoke.g') @mock.patch('permissions.logic.user_revoke.identity_logic') @mock.patch('permissions.logic.user_revoke.vend_contact_logic') @mock.patch('permissions.logic.user_revoke.vend_contact_model') @mock.patch('permissions.logic.user_revoke.neo4j_connector.db_session') @mock.patch('permissions.logic.user_revoke.tenant_model') def test_revoke_access_to_all_tenants_error_deactivating( tenant_model_mock: MagicMock, _neo_session_mock: MagicMock, _vend_contact_model_mock: MagicMock, _vend_contact_logic_mock: MagicMock, identity_logic_mock: MagicMock, g_mock: MagicMock, app_context: flask.ctx.AppContext, ) -> None: """Test revoke_access_to_all_tenants_for_identity when deactivating identity fails.""" tenant_model_mock.get_adminable_tenants_for_identity.return_value = [MagicMock()] tenant_model_mock.get_identity_tenant_count.return_value = 0 err = sqlalchemy.exc.SQLAlchemyError('🤪') identity_logic_mock.deactivate_identity.side_effect = err identity = mock.Mock(id='identity-uuid') with pytest.raises(sqlalchemy.exc.SQLAlchemyError): user_revoke.revoke_access_to_all_tenants_for_identity( admin_id='admin-id', admin_profile_id=100100, identity=identity, ) g_mock.log.error.assert_called_with( 'Error deactivating identity during revocation of all tenant access' '--access has still been removed', resources={'identity_id': 'identity-uuid', 'error': str(err)}, ) @mock.patch('permissions.logic.user_revoke.g') @mock.patch('permissions.logic.identity.deactivate_identity') @mock.patch('permissions.logic.user_revoke.default_brand_logic.update_default_brand_if_needed') @mock.patch('permissions.logic.vend_contact.set_new_auth0_vend_contact_id') @mock.patch('permissions.models.vend_contact.deactivate_by_tenant_for_identity') @mock.patch('permissions.logic.user_revoke.mysql.db_session') @mock.patch('permissions.logic.user_revoke.neo4j_connector.db_session') @mock.patch('permissions.logic.user_revoke.tenant_model') def test_revoke_access_to_all_tenants_includes_parent_companies_for_vendor_star( tenant_model_mock: MagicMock, neo_session_mock: MagicMock, mysql_session_mock: MagicMock, vend_contact_deactivate_mock: MagicMock, set_new_auth0_vend_contact_id_mock: MagicMock, update_default_brand_mock: MagicMock, deactivate_identity_mock: MagicMock, g_mock: MagicMock, app_context: flask.ctx.AppContext, ) -> None: """Test that parent company tenants are included when admin has vendor star.""" from permissions.constants import parent_companies vendor_star_tenant = Tenant( tenant_uuid=constants.VENDOR_STAR_UUID, tenant_type=TenantType.ACCOUNT ) regular_tenant = Tenant(tenant_uuid='other-uuid', tenant_type=TenantType.ACCOUNT) adminable_tenants = [ AdminableTenant( tenant=vendor_star_tenant, profiles=[mock.Mock(profile_type=constants.LABELPROFILE)] ), AdminableTenant( tenant=regular_tenant, profiles=[mock.Mock(profile_type=constants.INSIGHTSPROFILE)] ), ] tenant_model_mock.get_adminable_tenants_for_identity.return_value = adminable_tenants tx_mock = mock.Mock() neo_session_mock.return_value.__enter__.return_value.begin_transaction.return_value = tx_mock session_mock = mock.Mock() mysql_session_mock.return_value.__enter__.return_value = session_mock tenant_model_mock.get_identity_tenant_count.return_value = 1 mock_identity = mock.Mock(id='identity-uuid', auth0_user_id='auth0-id') user_revoke.revoke_access_to_all_tenants_for_identity( admin_id='admin-uuid', admin_profile_id=1, identity=mock_identity, ) # Verify that parent company tenants were added to the list call_args = tenant_model_mock.soft_delete_access_to_multiple_tenants_for_identity.call_args tenants_arg = call_args.kwargs['tenants'] # Should include vendor star, regular tenant, and both parent companies assert len(tenants_arg) == 4 tenant_uuids = {t.tenant_uuid for t in tenants_arg} assert constants.VENDOR_STAR_UUID in tenant_uuids assert 'other-uuid' in tenant_uuids assert parent_companies.ORCHARD_PARENT_COMPANY_UUID in tenant_uuids assert parent_companies.SME_PARENT_COMPANY_UUID in tenant_uuids # Verify parent companies have correct type parent_company_tenants = [t for t in tenants_arg if t.tenant_type == TenantType.PARENT_COMPANY] assert len(parent_company_tenants) == 2 tx_mock.commit.assert_called() @mock.patch('permissions.logic.user_revoke.g', spec=['log']) @mock.patch('permissions.logic.user_revoke.identity_logic') @mock.patch('permissions.logic.user_revoke.neo4j_connector.db_session') @mock.patch('permissions.logic.user_revoke.tenant_model') def test_revoke_all_access_for_employee_identity( tenant_model_mock: MagicMock, neo_session_mock: MagicMock, identity_logic_mock: MagicMock, g_mock: MagicMock, ) -> None: """Test revoke_all_access_for_employee_identity behavior.""" session_mock = mock.Mock() neo_session_mock.return_value.__enter__.return_value = session_mock admin_id = 'admin-uuid' profile_id = '12345' identity_id = 'identity-uuid' identity = mock.Mock(id=identity_id) admin_context = {'identity_id': admin_id, 'profile_id': profile_id} user_revoke.revoke_all_access_for_employee_identity( admin_context=admin_context, identity=identity, ) tenant_model_mock.soft_delete_all_access_to_tenants_for_identity.assert_called_once_with( session=session_mock, identity_id=identity_id, admin_id=admin_id, ) identity_logic_mock.deactivate_identity.assert_called_with( session=session_mock, identity=identity, admin_id=admin_id ) @mock.patch('permissions.logic.user_revoke.g', spec=['log']) @mock.patch('permissions.logic.user_revoke.neo4j_connector.db_session') @mock.patch('permissions.logic.user_revoke.tenant_model') def test_revoke_all_access_for_employee_identity_neo_error( tenant_model_mock: MagicMock, neo_session_mock: MagicMock, g_mock: MagicMock, ) -> None: """Test revoke_all_access_for_employee_identity when neo4j raises an error.""" session_mock = mock.Mock() neo_session_mock.return_value.__enter__.return_value = session_mock err = neo4j_exceptions.Neo4jError('Database connection failed') tenant_model_mock.soft_delete_all_access_to_tenants_for_identity.side_effect = err identity_id = 'identity-uuid' identity = mock.Mock(id=identity_id) admin_context = {'identity_id': 'admin-id', 'profile_id': '12345'} with pytest.raises(neo4j_exceptions.Neo4jError): user_revoke.revoke_all_access_for_employee_identity( admin_context=admin_context, identity=identity, ) g_mock.log.error.assert_called_with( 'Error soft-deleting access to all tenants for employee', resources={ 'identity_id': 'identity-uuid', 'error': str(err), }, ) @mock.patch('permissions.logic.user_revoke.g', spec=['log']) @mock.patch('permissions.logic.user_revoke.identity_logic') @mock.patch('permissions.logic.user_revoke.neo4j_connector.db_session') @mock.patch('permissions.logic.user_revoke.tenant_model') def test_revoke_all_access_for_employee_identity_error_deactivating( tenant_model_mock: MagicMock, neo_session_mock: MagicMock, identity_logic_mock: MagicMock, g_mock: MagicMock, ) -> None: """Test revoke_all_access_for_employee_identity when deactivating identity fails.""" session_mock = mock.Mock() neo_session_mock.return_value.__enter__.return_value = session_mock err = RuntimeError('Deactivation failed') identity_logic_mock.deactivate_identity.side_effect = err identity_id = 'identity-uuid' admin_context = {'identity_id': 'admin-id', 'profile_id': '12345'} with pytest.raises(RuntimeError): user_revoke.revoke_all_access_for_employee_identity( admin_context=admin_context, identity=mock.Mock(id=identity_id), ) g_mock.log.error.assert_called_with( 'Error deactivating identity during revocation of all tenant access' '--access has still been removed', resources={'identity_id': 'identity-uuid', 'error': str(err)}, )