"""Tests for Identity v2 Handlers.""" import json import uuid from typing import Any from unittest.mock import MagicMock, Mock, call, patch import flask.testing import pytest from flask.app import AppContext from flask.testing import FlaskClient from owsresponse import response from pytest_mock import MockerFixture from pythonfeatures import pythonfeatures from pythonfeatures.constants import split as split_constants import permissions from permissions.constants import constants from permissions.types import ( AccessibleTenant, AdminIdentity, Identity, IdentityTenantsWithV2Roles, ProfileInfo, Tenant, TenantRolesInput, TenantType, TenantWithV2Roles, ) VALID_SETTINGS_PROFILE = { 'profile_type': 'SettingsProfile', 'profile_id': 3333, 'roles': ['admin'], 'uuid': '555df6b6-661b-4d5b-bc0f-df83cf46eb75', } VALID_ACCESSIBLE_TENANT = AccessibleTenant( tenant_uuid='890df6b6-661b-4d5b-bc0f-df83cf46eb00', tenant_type=TenantType.ACCOUNT, access=True, ) @pytest.mark.parametrize( ( 'identity_uuid', 'get_identity_settings_profile', 'accessible_tenants', 'brand', 'status_code', ), [ pytest.param( '555df6b6-661b-4d5b-bc0f-df83cf46eb75', VALID_SETTINGS_PROFILE, [VALID_ACCESSIBLE_TENANT], 'awal', 200, id='Success', ), pytest.param( None, VALID_SETTINGS_PROFILE, [VALID_ACCESSIBLE_TENANT], 'awal', 401, id='Unauthorized error', ), pytest.param( '555df6b6-661b-4d5b-bc0f-df83cf46eb75', None, [VALID_ACCESSIBLE_TENANT], 'awal', 403, id='Forbidden user error', ), pytest.param( '555df6b6-661b-4d5b-bc0f-df83cf46eb75', VALID_SETTINGS_PROFILE, [ AccessibleTenant( tenant_uuid='890df6b6-661b-4d5b-bc0f-df83cf46eb00', tenant_type=TenantType.ACCOUNT, access=False, ) ], 'knr', 403, id='Forbidden access to the tenant', ), pytest.param( '555df6b6-661b-4d5b-bc0f-df83cf46eb75', None, [VALID_ACCESSIBLE_TENANT], None, 403, id='No company brand is associated with the tenant', ), ], ) @patch.object(pythonfeatures, 'get_single_feature') @patch('permissions.utils.api_utils.g') @patch('permissions.handlers.identity_v2_handlers.g') def test_v2_create_identity( mock_g: MagicMock, mock_g2: MagicMock, mock_feature: MagicMock, identity_uuid: str, get_identity_settings_profile: ProfileInfo, accessible_tenants: list[AccessibleTenant], brand: str, status_code: int, app_context: AppContext, mocker: MockerFixture, fixture_client: FlaskClient, ) -> None: """Test create a new identity v2.""" headers = { 'Orchard-Identity-Id': 'admin-uuid', 'Orchard-Profile-Id': 888, 'Orchard-Profile-Type': 'SettingsProfile', } data = { 'first_name': 'Foo', 'last_name': 'Bar', 'email': 'foo@bar.com', 'roles_to_attach': ['BANKING_TAX_BASE_ROLE'], 'tenant': {'tenant_type': 'account', 'tenant_uuid': 'dffedd4d-b88d-444d-a9eb-6ce89aa4d2f6'}, } expected_response_body = {'id': 'test_id'} mock_g.request_context.jwt_identity_id = identity_uuid mock_g2.request_context.jwt_identity_id = identity_uuid mock_feature.return_value = response.Response(message='control') mocker.patch( 'permissions.models.identity.get_identity_settings_profile', return_value=get_identity_settings_profile, ) mocker.patch( 'permissions.logic.tenant.check_admin_access_to_tenants', return_value=accessible_tenants ) mocker.patch( 'permissions.models.ows_account.get_enabled_tenant_applications', return_value=[{'roles': ['BANKING_TAX_BASE_ROLE']}], ) mocker.patch( 'permissions.models.tenant.get_parent_company_brand_for_tenant', return_value='awal' ) mocker.patch('permissions.logic.tenant.does_tenant_exist', return_value=True) mocker.patch( 'permissions.models.identity.get_identity_by_id_new', return_value=Identity( id='555df6b6-661b-4d5b-bc0f-df83cf46eb75', email='admin@test.com', first_name='Test', last_name='Test', name='Test Test', auth0_user_id='auth0|test', user_types=['label'], active='Y', default_brand='awal', ), ) mocker.patch('permissions.models.identity.get_identity_with_auth0', return_value=None) mocker.patch( 'permissions.logic.user_invite.create_or_update_user', return_value=(Mock(id='test_id'), None), ) mocker.patch('permissions.handlers.identity_v2_handlers.user_notify') handler_response = fixture_client.post('/v2/identities', headers=headers, json=data) assert handler_response assert handler_response.status_code == status_code if status_code == 200: assert handler_response.json == expected_response_body @pytest.mark.parametrize( ( 'edit_super_admins', 'get_identity_settings_profile', 'accessible_tenants', 'valid_roles', 'email', 'expected_status_code', 'expected_response_body', ), [ pytest.param( 'enabled', VALID_SETTINGS_PROFILE, [ AccessibleTenant( tenant_uuid='053a1a75-acc5-4cd8-9206-a194335d2afa', tenant_type=TenantType.ACCOUNT, access=True, ) ], True, 'foo@awal.com', 200, {'id': 'test_id'}, id='User can assign vendor* because all access checks pass', ), pytest.param( 'enabled', VALID_SETTINGS_PROFILE, [ AccessibleTenant( tenant_uuid='053a1a75-acc5-4cd8-9206-a194335d2afa', tenant_type=TenantType.ACCOUNT, access=True, ) ], False, 'foo@awal.com', 422, { 'code': 'Unable to process the request because it contains invalid data.', 'message': 'Error creating identity.', }, id='User is not able to create an identity because of incorrect roles.', ), pytest.param( 'control', VALID_SETTINGS_PROFILE, [ AccessibleTenant( tenant_uuid='053a1a75-acc5-4cd8-9206-a194335d2afa', tenant_type=TenantType.ACCOUNT, access=True, ) ], True, 'foo@awal.com', 403, {'code': 'bad_request', 'message': 'Cannot assign resource to this identity'}, id='User cannot assign vendor* because edit_super_admins is not enabled', ), pytest.param( 'enabled', VALID_SETTINGS_PROFILE, [ AccessibleTenant( tenant_uuid='053a1a75-acc5-4cd8-9206-a194335d2afa', tenant_type=TenantType.ACCOUNT, access=False, ) ], True, 'foo@awal.com', 403, {'code': 'User is forbidden', 'message': 'There is no access to the tenant.'}, id='User cannot assign vendor* because it is not an AccessibleTenant', ), pytest.param( 'enabled', VALID_SETTINGS_PROFILE, [ AccessibleTenant( tenant_uuid='053a1a75-acc5-4cd8-9206-a194335d2afa', tenant_type=TenantType.ACCOUNT, access=True, ) ], True, 'foo@bar.com', 403, {'code': 'bad_request', 'message': 'Cannot assign resource to this identity'}, id='User cannot assign vendor* because target identity email domain is not allowed', ), ], ) @patch.object(pythonfeatures, 'get_single_feature') @patch('permissions.utils.api_utils.g') @patch('permissions.handlers.identity_v2_handlers.g') def test_v2_create_identity_with_vendor_star( mock_g: MagicMock, mock_g2: MagicMock, mock_feature: MagicMock, edit_super_admins: str, get_identity_settings_profile: ProfileInfo, accessible_tenants: list[AccessibleTenant], valid_roles: bool, email: str, expected_status_code: int, expected_response_body: dict[str, Any], app_context: AppContext, mocker: MockerFixture, fixture_client: FlaskClient, ) -> None: """Test create a new identity v2.""" identity_uuid = str(uuid.uuid4()) headers = { 'Orchard-Identity-Id': 'admin-uuid', 'Orchard-Profile-Id': 888, 'Orchard-Profile-Type': 'SettingsProfile', } data = { 'first_name': 'Foo', 'last_name': 'Bar', 'email': email, 'roles_to_attach': ['BANKING_TAX_BASE_ROLE'], 'tenant': {'tenant_type': 'account', 'tenant_uuid': '053a1a75-acc5-4cd8-9206-a194335d2afa'}, } mock_g.request_context.jwt_identity_id = identity_uuid mock_g2.request_context.jwt_identity_id = identity_uuid mock_feature.side_effect = lambda feature_name, _: ( response.Response(message=edit_super_admins) if feature_name == constants.EDIT_SUPER_ADMINS else response.Response(message='control') ) mocker.patch( 'permissions.models.identity.get_identity_settings_profile', return_value=get_identity_settings_profile, ) mocker.patch( 'permissions.logic.tenant.check_admin_access_to_tenants', return_value=accessible_tenants ) mocker.patch( 'permissions.models.tenant.get_parent_company_brand_for_tenant', return_value='awal' ) mocker.patch('permissions.logic.tenant.does_tenant_exist', return_value=True) mocker.patch( 'permissions.models.ows_account.get_enabled_tenant_applications', return_value=[{'roles': ['BANKING_TAX_BASE_ROLE']}], ) mocker.patch( 'permissions.logic.tenant.check_compatibility_with_tenant_configuration', return_value=valid_roles, ) mocker.patch( 'permissions.models.identity.get_identity_by_id_new', return_value=Identity( id='555df6b6-661b-4d5b-bc0f-df83cf46eb75', email='admin@test.com', first_name='Test', last_name='Test', name='Test Test', auth0_user_id='auth0|test', user_types=['label'], active='Y', default_brand='awal', ), ) mocker.patch('permissions.models.identity.get_identity_with_auth0', return_value=None) mocker.patch( 'permissions.logic.user_invite.create_or_update_user', return_value=(Mock(id='test_id'), None), ) mocker.patch('permissions.handlers.identity_v2_handlers.user_notify') handler_response = fixture_client.post('/v2/identities', headers=headers, json=data) assert handler_response assert handler_response.status_code == expected_status_code, handler_response.text assert handler_response.json == expected_response_body, handler_response.text @patch.object(pythonfeatures, 'get_single_feature') @patch('permissions.utils.api_utils.g') @patch('permissions.handlers.identity_v2_handlers.g') def test_v2_create_identity_delegates_to_create_or_update_user( mock_g: MagicMock, mock_g2: MagicMock, mock_single_feature: MagicMock, app_context: AppContext, mocker: MockerFixture, fixture_client: FlaskClient, ) -> None: """Test create identity v2 delegates notification responsibility to logic layer.""" headers = { 'Orchard-Identity-Id': 'admin-uuid', 'Orchard-Profile-Id': 888, 'Orchard-Profile-Type': 'SettingsProfile', } data = { 'first_name': 'Foo', 'last_name': 'Bar', 'email': 'foo@bar.com', 'roles_to_attach': ['BANKING_TAX_BASE_ROLE'], 'tenant': {'tenant_type': 'account', 'tenant_uuid': 'dffedd4d-b88d-444d-a9eb-6ce89aa4d2f6'}, } identity_id = '555df6b6-661b-4d5b-bc0f-df83cf46eb75' settings_profile = ProfileInfo( **VALID_SETTINGS_PROFILE, ) identity = Identity( id='555df6b6-661b-4d5b-bc0f-df83cf46eb75', email='admin@test.com', first_name='Test', last_name='Test', name='Test Test', auth0_user_id='auth0|test', user_types=['label'], active='Y', default_brand='awal', ) mock_g.request_context.jwt_identity_id = identity_id mock_g2.request_context.jwt_identity_id = identity_id mock_single_feature.return_value = response.Response(message=split_constants.FEATURE_DISABLED) mocker.patch( 'permissions.models.identity.get_identity_settings_profile', return_value=settings_profile, ) mocker.patch( 'permissions.logic.tenant.check_admin_access_to_tenants', return_value=[VALID_ACCESSIBLE_TENANT], ) mocker.patch( 'permissions.models.ows_account.get_enabled_tenant_applications', return_value=[{'roles': ['BANKING_TAX_BASE_ROLE']}], ) mocker.patch( 'permissions.models.tenant.get_parent_company_brand_for_tenant', return_value='awal' ) mocker.patch('permissions.logic.tenant.does_tenant_exist', return_value=True) mocker.patch( 'permissions.models.identity.get_identity_by_id_new', return_value=identity, ) mocker.patch('permissions.models.identity.get_identity_with_auth0', return_value=None) create_or_update_mock = mocker.patch( 'permissions.logic.user_invite.create_or_update_user', return_value=(Mock(id='new_id'), None), ) mocker.patch('permissions.handlers.identity_v2_handlers.user_notify') handler_response = fixture_client.post('/v2/identities', headers=headers, json=data) assert handler_response.status_code == 200 assert handler_response.json == {'id': 'new_id'} create_or_update_mock.assert_called_once() @patch('permissions.utils.api_utils.g') @patch('permissions.handlers.identity_v2_handlers.g') def test_v2_create_identity_tenant_not_found( mock_g: MagicMock, mock_g2: MagicMock, mocker: MockerFixture, app_context: AppContext, fixture_client: FlaskClient, ) -> None: """Test create identity returns 404 when tenant doesn't exist.""" mock_g.request_context.jwt_identity_id = mock_g2.request_context.jwt_identity_id = 'admin-id' mocker.patch('permissions.handlers.identity_v2_handlers.pythonfeatures.get_single_feature') mocker.patch('permissions.models.identity.get_identity_by_id_new') mocker.patch('permissions.models.identity.get_identity_settings_profile') mocker.patch('permissions.logic.tenant.check_admin_access_to_tenant') mocker.patch('permissions.logic.tenant.does_tenant_exist', return_value=False) result = fixture_client.post( '/v2/identities', json={ 'email': 'test@example.com', 'first_name': 'Test', 'last_name': 'User', 'roles_to_attach': ['SETTINGS_BASE_ROLE'], 'tenant': {'tenant_type': 'account', 'tenant_uuid': 'uuid'}, }, ) assert result.status_code == 404 assert result.json['message'] == 'Tenant not found.' VALID_PATCH_DATA = { 'roles_to_attach': ['BANKING_TAX_BASE_ROLE'], 'roles_to_detach': [], 'tenant': {'tenant_type': 'account', 'tenant_uuid': 'dffedd4d-b88d-444d-a9eb-6ce89aa4d2f6'}, } VENDOR_STAR_TO_UPDATE = { 'roles_to_attach': ['BANKING_TAX_BASE_ROLE'], 'roles_to_detach': [], 'tenant': {'tenant_type': 'account', 'tenant_uuid': '053a1a75-acc5-4cd8-9206-a194335d2afa'}, } @pytest.mark.parametrize( ('identity_uuid', 'admin_id', 'profile_info', 'tenant', 'status_code', 'error_message'), [ pytest.param( '555df6b6-661b-4d5b-bc0f-df83cf46eb75', None, None, None, 401, 'Request context has no identity uuid.', id='Unauthorized error', ), pytest.param( '555df6b6-661b-4d5b-bc0f-df83cf46eb75', '345d1cf7-3b20-465c-ab13-69fd46ec4bcb', None, None, 403, 'user is forbidden', id='Forbidden user error', ), pytest.param( '555df6b6-661b-4d5b-bc0f-df83cf46eb75', '345d1cf7-3b20-465c-ab13-69fd46ec4bcb', VALID_SETTINGS_PROFILE, AccessibleTenant( tenant_uuid='890df6b6-661b-4d5b-bc0f-df83cf46eb00', tenant_type=TenantType.ACCOUNT, access=False, ), 403, 'There is no access to the tenant.', id='Forbidden access to the tenant', ), ], ) @patch.object(pythonfeatures, 'get_single_feature') @patch('permissions.utils.api_utils.g') @patch('permissions.handlers.identity_v2_handlers.g') def test_v2_update_identity_access_checking( mock_g: MagicMock, mock_g2: MagicMock, mock_feature: MagicMock, identity_uuid: str, admin_id: str, profile_info: ProfileInfo, tenant: AccessibleTenant, status_code: int, error_message: str, app_context: AppContext, mocker: MockerFixture, fixture_client: FlaskClient, ): """Test access checks in the PATCH v2/identities/ endpoint.""" headers = { 'Orchard-Identity-Id': admin_id, 'Orchard-Profile-Id': 888, 'Orchard-Profile-Type': 'SettingsProfile', } mock_g.request_context.jwt_identity_id = admin_id mock_g2.request_context.jwt_identity_id = admin_id mock_feature.return_value = response.Response(message='control') mocker.patch( 'permissions.models.identity.get_identity_settings_profile', return_value=profile_info, ) mocker.patch( 'permissions.logic.tenant.check_admin_access_to_tenants', return_value=[tenant], ) mocker.patch('permissions.logic.tenant.does_tenant_exist', return_value=True) handler_response = fixture_client.patch( f'/v2/identities/{identity_uuid}', headers=headers, json=VALID_PATCH_DATA, ) assert handler_response.status_code == status_code assert handler_response.json['message'] == error_message @pytest.mark.parametrize( ( 'profile_info', 'edit_super_admins', 'tenant', 'has_valid_roles', 'is_update_removing_last_tenant', 'email', 'expected_status_code', 'expected_response_body', ), [ pytest.param( ProfileInfo(**VALID_SETTINGS_PROFILE), 'control', AccessibleTenant( tenant_uuid='053a1a75-acc5-4cd8-9206-a194335d2afa', tenant_type=TenantType.ACCOUNT, access=True, ), True, True, 'foo@theorchard.com', 403, {'code': 'bad_request', 'message': 'Cannot assign resource to this identity'}, id='User cannot assign vendor* because edit_super_admins is not enabled', ), pytest.param( ProfileInfo(**VALID_SETTINGS_PROFILE), 'enabled', AccessibleTenant( tenant_uuid='053a1a75-acc5-4cd8-9206-a194335d2afa', tenant_type=TenantType.ACCOUNT, access=False, ), True, True, 'foo@theorchard.com', 403, {'code': 'User is forbidden', 'message': 'There is no access to the tenant.'}, id='User cannot assign vendor* because it is not an AccessibleTenant', ), pytest.param( ProfileInfo(**VALID_SETTINGS_PROFILE), 'enabled', AccessibleTenant( tenant_uuid='053a1a75-acc5-4cd8-9206-a194335d2afa', tenant_type=TenantType.ACCOUNT, access=True, ), True, False, 'foo@bar.com', 403, {'code': 'bad_request', 'message': 'Cannot assign resource to this identity'}, id='User cannot assign vendor* because target identity email domain is not allowed', ), pytest.param( ProfileInfo(**VALID_SETTINGS_PROFILE), 'enabled', AccessibleTenant( tenant_uuid='053a1a75-acc5-4cd8-9206-a194335d2afa', tenant_type=TenantType.ACCOUNT, access=True, ), True, False, 'foo@theorchard.com', 200, {'id': 'c86c941b-5b28-42bd-b3ab-d6756c31d195'}, id='User can update an identity with vendor*', ), pytest.param( ProfileInfo(**VALID_SETTINGS_PROFILE), 'enabled', AccessibleTenant( tenant_uuid='053a1a75-acc5-4cd8-9206-a194335d2afa', tenant_type=TenantType.ACCOUNT, access=True, ), False, False, 'foo@theorchard.com', 422, { 'code': 'Unable to process the request because it contains invalid data.', 'message': 'Error updating identity.', }, id='User is not able to update an identity because of invalid roles.', ), pytest.param( ProfileInfo(**VALID_SETTINGS_PROFILE), 'enabled', AccessibleTenant( tenant_uuid='053a1a75-acc5-4cd8-9206-a194335d2afa', tenant_type=TenantType.ACCOUNT, access=True, ), True, True, 'foo@theorchard.com', 422, { 'code': 'Unable to process the request because it contains invalid data.', 'message': 'You are trying to detach access from the last tenant.', }, id='User is not able to detach access from the last tenant..', ), pytest.param( ProfileInfo(**VALID_SETTINGS_PROFILE), 'enabled', AccessibleTenant( tenant_uuid='053a1a75-acc5-4cd8-9206-a194335d2afa', tenant_type=TenantType.ACCOUNT, access=True, ), True, ValueError, 'foo@theorchard.com', 400, { 'code': 'bad_request', 'message': 'You are not enable proceed with updating, because tenant for ' 'update 123-abdc and accessible tenant ' '053a1a75-acc5-4cd8-9206-a194335d2afa are not matching.', }, id='Tenant for update and accessible tenant are not matching.', ), ], ) @patch.object(pythonfeatures, 'get_single_feature') @patch('permissions.utils.api_utils.g') @patch('permissions.handlers.identity_v2_handlers.g') def test_v2_update_identity_with_vendor_star( mock_g: MagicMock, mock_g2: MagicMock, mock_feature: MagicMock, profile_info: ProfileInfo, edit_super_admins: str, tenant: AccessibleTenant, has_valid_roles: bool, is_update_removing_last_tenant: bool, email: str, expected_status_code: int, expected_response_body: dict[str, Any], app_context: AppContext, mocker: MockerFixture, fixture_client: FlaskClient, ): """Test access checks in the PATCH v2/identities/ endpoint.""" identity_uuid = 'c86c941b-5b28-42bd-b3ab-d6756c31d195' admin_identity_uuid = uuid.uuid4() headers = { 'Orchard-Identity-Id': admin_identity_uuid, 'Orchard-Profile-Id': 888, 'Orchard-Profile-Type': 'SettingsProfile', } mock_g.request_context.jwt_identity_id = admin_identity_uuid mock_g2.request_context.jwt_identity_id = admin_identity_uuid mock_feature.side_effect = lambda feature_name, _: ( response.Response(message=edit_super_admins) if feature_name == constants.EDIT_SUPER_ADMINS else response.Response(message='control') ) mocker.patch( 'permissions.models.identity.get_identity_settings_profile', return_value=profile_info, ) mocker.patch( 'permissions.logic.tenant.check_admin_access_to_tenants', return_value=[tenant], ) mocker.patch('permissions.logic.tenant.does_tenant_exist', return_value=True) mocker.patch( 'permissions.models.identity.get_identity_by_id_new', side_effect=[ (Mock(email=email)), ( Mock( id=admin_identity_uuid, first_name='Guy', last_name='Doing Business', name='Guy Doing Business', email='guy@theorchard.com', auth0_user_id='auth0|numbers', active='Y', user_types=[], settings_profile={}, ) ), ], ) mocker.patch( 'permissions.models.ows_account.get_enabled_tenant_applications', return_value=[{'roles': ['BANKING_TAX_BASE_ROLE']}], ) mocker.patch( 'permissions.logic.tenant.check_compatibility_with_tenant_configuration', return_value=has_valid_roles, ) mocker.patch( 'permissions.logic.tenant.is_update_removing_last_tenant', return_value=is_update_removing_last_tenant, ) mocker.patch( 'permissions.logic.tenant.get_parent_company_brand_for_tenant', return_value='awal', ) mocker.patch( 'permissions.models.identity.get_identity_with_auth0_for_existing_identity', return_value=Mock(), ) mocker.patch('permissions.logic.user_update.update_user') if is_update_removing_last_tenant is ValueError: mocker.patch( 'permissions.logic.tenant.is_update_removing_last_tenant', side_effect=ValueError( 'You are not enable proceed with updating, because tenant' ' for update {} and accessible tenant {} are not matching.'.format( '123-abdc', tenant.tenant_uuid ) ), ) VENDOR_STAR_PATCH_DATA = VENDOR_STAR_TO_UPDATE if not is_update_removing_last_tenant: VENDOR_STAR_PATCH_DATA['roles_to_attach'] = [] VENDOR_STAR_PATCH_DATA['roles_to_detach'] = ['BANKING_TAX_BASE_ROLE'] handler_response = fixture_client.patch( f'/v2/identities/{identity_uuid}', headers=headers, json=VENDOR_STAR_PATCH_DATA, ) assert handler_response.status_code == expected_status_code, handler_response.text assert handler_response.json == expected_response_body, handler_response.text if expected_status_code == 200 or expected_status_code == 422: permissions.logic.tenant.check_compatibility_with_tenant_configuration.assert_called_with( Tenant( tenant_type=tenant.tenant_type, tenant_uuid=tenant.tenant_uuid, ), VENDOR_STAR_PATCH_DATA['roles_to_attach'], ) @patch('permissions.handlers.identity_v2_handlers.tenant_logic') @patch('permissions.handlers.identity_v2_handlers.identity_model') @patch.object(pythonfeatures, 'get_single_feature') @patch('permissions.utils.api_utils.g') @patch('permissions.handlers.identity_v2_handlers.g') def test_v2_update_identity_identity_check( mock_g: MagicMock, mock_g2: MagicMock, mock_feature: MagicMock, identity_model_mock: MagicMock, _, app_context: AppContext, fixture_client: FlaskClient, ): """Test the identity check in the PATCH v2/identities/ endpoint.""" identity_uuid = '555df6b6-661b-4d5b-bc0f-df83cf46eb75' admin_id = '345d1cf7-3b20-465c-ab13-69fd46ec4bcb' identity_model_mock.get_identity_by_id_new.return_value = None mock_g.request_context.jwt_identity_id = admin_id mock_g2.request_context.jwt_identity_id = admin_id mock_feature.return_value = response.Response(message='control') result = fixture_client.patch(f'/v2/identities/{identity_uuid}', json=VALID_PATCH_DATA) assert result.status_code == 404 @patch('permissions.handlers.identity_v2_handlers.tenant_logic') @patch('permissions.handlers.identity_v2_handlers.identity_model') @patch.object(pythonfeatures, 'get_single_feature') @patch('permissions.utils.api_utils.g') @patch('permissions.handlers.identity_v2_handlers.g') def test_v2_update_identity_brand_check( mock_g: MagicMock, mock_g2: MagicMock, mock_feature: MagicMock, _, tenant_logic_mock: MagicMock, app_context: AppContext, fixture_client: FlaskClient, ): """Test the brand check in the PATCH v2/identities/ endpoint.""" identity_uuid = '555df6b6-661b-4d5b-bc0f-df83cf46eb75' admin_id = '345d1cf7-3b20-465c-ab13-69fd46ec4bcb' mock_g.request_context.jwt_identity_id = admin_id mock_g2.request_context.jwt_identity_id = admin_id mock_feature.return_value = response.Response(message='control') tenant_logic_mock.get_parent_company_brand_for_tenant.return_value = None result = fixture_client.patch(f'/v2/identities/{identity_uuid}', json=VALID_PATCH_DATA) assert result.status_code == 403 assert result.json['message'] == 'No one company brand is associated with the tenant.' @patch('permissions.handlers.identity_v2_handlers.user_notify') @patch('permissions.handlers.identity_v2_handlers.user_update') @patch('permissions.handlers.identity_v2_handlers.tenant_logic') @patch('permissions.handlers.identity_v2_handlers.identity_model') @patch('permissions.utils.api_utils.g') @patch('permissions.handlers.identity_v2_handlers.g') @patch.object(pythonfeatures, 'get_single_feature_by_attributes') @patch.object(pythonfeatures, 'get_single_feature') def test_v2_update_identity_method_calls( mock_single_feature: MagicMock, mock_feature_by_attr: MagicMock, mock_g: MagicMock, mock_g2: MagicMock, identity_model_mock: MagicMock, tenant_logic_mock: MagicMock, user_update_mock: MagicMock, _user_notify_mock: MagicMock, fixture_client: FlaskClient, app_context: AppContext, ) -> None: """Test the method calls in the PATCH v2/identities/ endpoint.""" # Given... identity_uuid = '555df6b6-661b-4d5b-bc0f-df83cf46eb75' admin_id = '345d1cf7-3b20-465c-ab13-69fd46ec4bcb' mock_single_feature.return_value = response.Response(message=split_constants.FEATURE_DISABLED) mock_feature_by_attr.return_value = response.Response(message=split_constants.FEATURE_ENABLED) mock_g.request_context.jwt_identity_id = admin_id mock_g2.request_context.jwt_identity_id = admin_id expected_tenant = Tenant( tenant_type=TenantType.ACCOUNT, tenant_uuid=VALID_PATCH_DATA['tenant']['tenant_uuid'] ) brand = 'theorchard' tenant_logic_mock.get_parent_company_brand_for_tenant.return_value = brand user_identity = MagicMock(name='user identity', is_pending=lambda: False) admin_identity = MagicMock(name='admin identity') identity_model_mock.get_identity_by_id_new.side_effect = [user_identity, admin_identity] expected_admin = AdminIdentity( id=admin_identity.id, email=admin_identity.email, first_name=admin_identity.first_name, last_name=admin_identity.last_name, name=admin_identity.name, auth0_user_id=admin_identity.auth0_user_id, active=admin_identity.active, user_types=admin_identity.user_types, default_brand=admin_identity.default_brand, settings_profile=identity_model_mock.get_identity_settings_profile.return_value, ) identity_model_mock.get_identity_with_auth0_for_existing_identity.return_value = user_identity # When... fixture_client.patch(f'/v2/identities/{identity_uuid}', json=VALID_PATCH_DATA) # Then... identity_model_mock.get_identity_settings_profile.assert_called_once_with(admin_id) tenant_logic_mock.check_admin_access_to_tenant.assert_called_once_with( identity_uuid=admin_id, tenant=expected_tenant, settings_profile=identity_model_mock.get_identity_settings_profile.return_value, ) identity_model_mock.get_identity_by_id_new.assert_has_calls( [call(identity_uuid), call(admin_id)] ) tenant_logic_mock.check_compatibility_with_tenant_configuration.assert_called_once_with( expected_tenant, VALID_PATCH_DATA['roles_to_attach'] ) tenant_logic_mock.get_parent_company_brand_for_tenant.assert_called_once_with(expected_tenant) tenant_logic_mock.does_tenant_exist.assert_called_once_with(expected_tenant) identity_model_mock.get_identity_with_auth0_for_existing_identity.assert_called_once_with( admin=expected_admin, existing_identity=user_identity, email=user_identity.email, brand=brand, ) user_update_mock.update_user.assert_called_once_with( admin=expected_admin, identity_with_auth0=( identity_model_mock.get_identity_with_auth0_for_existing_identity.return_value ), tenant_roles_input=TenantRolesInput( roles_to_attach=VALID_PATCH_DATA['roles_to_attach'], roles_to_detach=VALID_PATCH_DATA['roles_to_detach'], tenant=expected_tenant, ), brand=brand, ) @patch('permissions.utils.api_utils.g') @patch('permissions.handlers.identity_v2_handlers.g') def test_v2_update_identity_tenant_not_found( mock_g: MagicMock, mock_g2: MagicMock, mocker: MockerFixture, app_context: AppContext, fixture_client: FlaskClient, ) -> None: """Test update identity returns 404 when tenant doesn't exist.""" mock_g.request_context.jwt_identity_id = mock_g2.request_context.jwt_identity_id = 'admin-id' mocker.patch('permissions.handlers.identity_v2_handlers.pythonfeatures.get_single_feature') mocker.patch('permissions.models.identity.get_identity_by_id_new') mocker.patch('permissions.models.identity.get_identity_settings_profile') mocker.patch('permissions.logic.tenant.check_admin_access_to_tenant') mocker.patch('permissions.logic.tenant.does_tenant_exist', return_value=False) result = fixture_client.patch( '/v2/identities/test-identity-id', json=VALID_PATCH_DATA, ) assert result.status_code == 404 assert result.json['message'] == 'Tenant not found.' @pytest.mark.parametrize( ('admin_id', 'get_identity_settings_profile', 'accessible_tenants', 'status_code'), [ pytest.param( None, VALID_SETTINGS_PROFILE, [VALID_ACCESSIBLE_TENANT], 401, id='Unauthorized error', ), pytest.param( '555df6b6-661b-4d5b-bc0f-df83cf46eb75', None, [VALID_ACCESSIBLE_TENANT], 403, id='Forbidden user error', ), pytest.param( '555df6b6-661b-4d5b-bc0f-df83cf46eb75', None, [ AccessibleTenant( tenant_uuid='890df6b6-661b-4d5b-bc0f-df83cf46eb00', tenant_type=TenantType.ACCOUNT, access=False, ) ], 403, id='Forbidden access to tenant', ), ], ) @patch('permissions.utils.api_utils.g') @patch('permissions.handlers.identity_v2_handlers.g') def test_v2_revoke_access_to_single_tenant_for_identity_access( mock_g: MagicMock, mock_g2: MagicMock, admin_id: str, get_identity_settings_profile: ProfileInfo, accessible_tenants: list[AccessibleTenant], status_code: int, app_context: AppContext, mocker: MockerFixture, fixture_client: FlaskClient, ) -> None: """Test revoke all access to a single tenant for identity.""" headers = { 'Orchard-Identity-Id': 'admin-uuid', 'Orchard-Profile-Id': 888, 'Orchard-Profile-Type': 'SettingsProfile', } tenant_type = 'account' tenant_uuid = 'dffedd4d-b88d-444d-a9eb-6ce89aa4d2f6' identity_uuid = '555df6b6-661b-4d5b-bc0f-df83cf46eb75' mock_g.request_context.jwt_identity_id = admin_id mock_g2.request_context.jwt_identity_id = admin_id mocker.patch( 'permissions.models.identity.get_identity_settings_profile', return_value=get_identity_settings_profile, ) mocker.patch( 'permissions.logic.tenant.check_admin_access_to_tenants', return_value=accessible_tenants ) mocker.patch('permissions.logic.tenant.does_tenant_exist', return_value=True) mocker.patch( 'permissions.logic.user_revoke.revoke_access_to_tenant_for_identity', return_value=response.Response(message={'IdentityId': identity_uuid}), ) handler_response = fixture_client.delete( f'v2/identities/{identity_uuid}/tenants/{tenant_type}/{tenant_uuid}', headers=headers ) assert handler_response assert handler_response.status_code == status_code @patch('permissions.handlers.identity_v2_handlers.user_revoke') @patch('permissions.handlers.identity_v2_handlers.tenant_logic') @patch('permissions.handlers.identity_v2_handlers.identity_model') @patch('permissions.utils.api_utils.g') @patch('permissions.handlers.identity_v2_handlers.g') def test_v2_revoke_access_to_single_tenant_success( mock_g: MagicMock, mock_g2: MagicMock, identity_model_mock: MagicMock, tenant_logic_mock: MagicMock, user_revoke_mock: MagicMock, fixture_client: FlaskClient, app_context: AppContext, ) -> None: """Test DELETE v2/identities//tenants/tenant_type/.""" admin_id = '4af0a0c4-1ad0-4a34-8b33-f8df7f08e9d3' mock_g.request_context.jwt_identity_id = admin_id mock_g2.request_context.jwt_identity_id = admin_id admin_profile_id = 11238 settings_profile_mock = MagicMock(profile_id=admin_profile_id) identity_model_mock.get_identity_settings_profile.return_value = settings_profile_mock mock_identity = MagicMock() identity_model_mock.get_identity_by_id_new.return_value = mock_identity tenant_type = 'account' tenant_uuid = 'dffedd4d-b88d-444d-a9eb-6ce89aa4d2f6' expected_tenant = Tenant(tenant_type=tenant_type, tenant_uuid=tenant_uuid) identity_id = '36b548f0-4574-43c5-bbf9-847568cca83a' headers = { 'Orchard-Identity-Id': admin_id, 'Orchard-Profile-Id': admin_profile_id, 'Orchard-Profile-Type': 'SettingsProfile', } user_revoke_mock.revoke_access_to_tenant_for_identity.return_value = response.Response() res = fixture_client.delete( f'v2/identities/{identity_id}/tenants/{tenant_type}/{tenant_uuid}', headers=headers ) assert res.status_code == 204 identity_model_mock.get_identity_settings_profile.assert_called_with(admin_id) tenant_logic_mock.check_admin_access_to_tenant.assert_called_with( identity_uuid=admin_id, tenant=expected_tenant, settings_profile=settings_profile_mock, ) tenant_logic_mock.does_tenant_exist.assert_called_with(expected_tenant) user_revoke_mock.revoke_access_to_tenant_for_identity.assert_called_with( admin_identity_id=admin_id, admin_profile_id=admin_profile_id, identity_id=identity_id, tenant=expected_tenant, ) @patch('permissions.utils.api_utils.g') @patch('permissions.handlers.identity_v2_handlers.g') def test_v2_revoke_access_to_single_tenant_tenant_not_found( mock_g: MagicMock, mock_g2: MagicMock, mocker: MockerFixture, app_context: AppContext, fixture_client: FlaskClient, ) -> None: """Test revoke access returns 404 when tenant doesn't exist.""" mock_g.request_context.jwt_identity_id = mock_g2.request_context.jwt_identity_id = 'admin-id' mocker.patch('permissions.models.identity.get_identity_settings_profile') mocker.patch('permissions.logic.tenant.check_admin_access_to_tenant') mocker.patch('permissions.logic.tenant.does_tenant_exist', return_value=False) result = fixture_client.delete('/v2/identities/test-identity-id/tenants/account/test-uuid') assert result.status_code == 404 assert result.json['message'] == 'Tenant not found.' @pytest.mark.parametrize( ('admin_id', 'settings_profile', 'identity', 'status_code'), [ pytest.param(None, None, None, 401, id='Failed jwt check'), pytest.param('a-uuid', None, None, 403, id='Failed settings profile check'), pytest.param('a-uuid', MagicMock(), None, 404, id='Failed identity existence check'), ], ) @patch('permissions.handlers.identity_v2_handlers.identity_model') @patch('permissions.utils.api_utils.g') @patch('permissions.handlers.identity_v2_handlers.g') def test_v2_revoke_all_access_for_identity_checks( mock_g: MagicMock, mock_g2: MagicMock, identity_model_mock: MagicMock, admin_id: str | None, settings_profile: MagicMock | None, identity: None, status_code: int, fixture_client: FlaskClient, app_context: AppContext, ): """Test DELETE v2/identities/ endpoint validations.""" headers = { 'Orchard-Identity-Id': 'admin_id', 'Orchard-Profile-Id': 999, 'Orchard-Profile-Type': 'SettingsProfile', } mock_g.request_context.jwt_identity_id = admin_id mock_g2.request_context.jwt_identity_id = admin_id identity_model_mock.get_identity_settings_profile.return_value = settings_profile identity_model_mock.get_identity_by_id_new.return_value = identity response = fixture_client.delete('v2/identities/an-identity-id', headers=headers) assert response.status_code == status_code @patch('permissions.logic.user_revoke.revoke_access_to_all_tenants_for_identity') @patch('permissions.handlers.identity_v2_handlers.identity_model') @patch('permissions.utils.api_utils.g') @patch('permissions.handlers.identity_v2_handlers.g') def test_v2_revoke_all_access_for_identity( mock_g: MagicMock, mock_g2: MagicMock, identity_model_mock: MagicMock, revoke_access_mock: MagicMock, fixture_client: FlaskClient, app_context: AppContext, ): """Test DELETE v2/identities/ endpoint.""" admin_id = 'admin-id' mock_g.request_context.jwt_identity_id = admin_id mock_g2.request_context.jwt_identity_id = admin_id admin_profile_id = 11238 profile_mock = MagicMock(profile_id=admin_profile_id) identity_model_mock.get_identity_settings_profile.return_value = profile_mock mock_identity = MagicMock() identity_model_mock.get_identity_by_id_new.return_value = mock_identity identity_id = 'different-persons-id' headers = { 'Orchard-Identity-Id': 'something not matching', 'Orchard-Profile-Id': admin_profile_id, 'Orchard-Profile-Type': 'SettingsProfile', } response = fixture_client.delete(f'v2/identities/{identity_id}', headers=headers) assert response.status_code == 204 identity_model_mock.get_identity_settings_profile.assert_called_with(admin_id) identity_model_mock.get_identity_by_id_new.assert_called_with(identity_id) revoke_access_mock.assert_called_with( admin_id=admin_id, admin_profile_id=admin_profile_id, identity=mock_identity ) @patch('permissions.utils.api_utils.g') @patch('permissions.handlers.identity_v2_handlers.g') def test_get_adminable_tenants_for_identity_authentication( mock_g, mock_g2, app_context, fixture_client ): """Test authentication for get_adminable_tenants_for_identity endpoint.""" mock_g.request_context.jwt_identity_id = None mock_g2.request_context.jwt_identity_id = None assert fixture_client.get('/v2/identity/abc/tenants').status_code == 401 @patch('permissions.models.identity.get_identity_settings_profile') @patch('permissions.utils.api_utils.g') @patch('permissions.handlers.identity_v2_handlers.g') def test_get_adminable_tenants_for_identity_authorization( mock_g, mock_g2, get_profile_mock, app_context, fixture_client ): """Test authorization for get_adminable_tenants_for_identity endpoint.""" mock_g.request_context.jwt_identity_id = 'some-uuid' mock_g2.request_context.jwt_identity_id = 'some-uuid' get_profile_mock.return_value = None assert fixture_client.get('/v2/identity/abc/tenants').status_code == 403 @patch('permissions.logic.tenant.get_adminable_tenants_for_identity') @patch('permissions.models.identity.get_identity_settings_profile') @patch('permissions.utils.api_utils.g') @patch('permissions.handlers.identity_v2_handlers.g') def test_get_adminable_tenants_for_identity( mock_g, mock_g2, get_profile_mock, get_tenants_mock, app_context, fixture_client ): """Test get_adminable_tenants_for_identity endpoint.""" admin_uuid = 'some-uuid' mock_g.request_context.jwt_identity_id = admin_uuid mock_g2.request_context.jwt_identity_id = admin_uuid profile_id = 1331 get_profile_mock.return_value = ProfileInfo( profile_id=profile_id, profile_type='SettingsProfile', roles=[], uuid=admin_uuid ) tenant_list = [ TenantWithV2Roles( tenant_uuid='uuid-of-a-tenant', tenant_type=TenantType.ACCOUNT, roles=['INSIGHTS_BASE_ROLE'], ) ] get_tenants_mock.return_value = tenant_list user_uuid = 'a-different-uuid' result = fixture_client.get(f'/v2/identity/{user_uuid}/tenants') assert result.status_code == 200 assert json.loads(result.data)['tenants'] == [ { 'tenant_uuid': 'uuid-of-a-tenant', 'tenant_type': 'account', 'roles': ['INSIGHTS_BASE_ROLE'], } ] get_tenants_mock.assert_called_with( admin_context={'identity_id': admin_uuid, 'profile_id': profile_id}, identity_id=user_uuid, ) @pytest.mark.parametrize( ('jwt_identity_id', 'return_settings_profile', 'status_code'), [ ('test-identity-id-123', {'profile_type': 'SettingsProfile', 'profile_id': 1234}, 200), (None, None, 401), ('test-identity-id-123', None, 403), ], ) @patch('permissions.utils.api_utils.g') @patch('permissions.handlers.identity_v2_handlers.g') def test_get_my_adminable_tenant_types( mock_g, mock_g2, app_context, mocker, fixture_client, jwt_identity_id, return_settings_profile, status_code, ): """Test GET /v2/identity/self/tenant-types.""" mock_g.request_context.jwt_identity_id = jwt_identity_id mock_g2.request_context.jwt_identity_id = jwt_identity_id expected_response = response.Response( message=[ {'tenant_type': 'account', 'tenant_count': 1}, {'tenant_type': 'label_participant', 'tenant_count': 2}, {'tenant_type': 'collaborator', 'tenant_count': 2}, ] ) mocker.patch( 'permissions.models.identity.get_identity_settings_profile', return_value=return_settings_profile, ) mocker.patch( 'permissions.logic.tenant.get_my_adminable_tenant_types', return_value=expected_response ) handler_response = fixture_client.get('/v2/identity/self/tenant-types') assert handler_response.status_code == status_code if status_code == 200: assert handler_response.json == expected_response.message @patch('permissions.utils.api_utils.g') @patch('permissions.handlers.identity_v2_handlers.g') @patch('permissions.models.identity.get_identity_settings_profile') @patch('permissions.logic.tenant.check_admin_access_to_tenants') def test_check_my_access_to_tenants_success( check_access_to_tenants, get_identity_settings_profile, mock_g, mock_g2, app_context, fixture_client, ): """Test POST /v2/identity/self/tenant-access success.""" mock_g.request_context.jwt_identity_id = 'admin-uuid' mock_g.request_context.profile_type = 'SettingsProfile' mock_g2.request_context.jwt_identity_id = 'admin-uuid' mock_g2.request_context.profile_type = 'SettingsProfile' get_identity_settings_profile.return_value = ( ('test-identity-id-123', {'profile_type': 'SettingsProfile', 'profile_id': 1234}, 200), ) check_access_to_tenants.return_value = [ AccessibleTenant(tenant_type=TenantType.ACCOUNT, tenant_uuid='valid-uuid', access=True) ] payload = {'tenants': [{'tenant_type': 'account', 'tenant_uuid': 'valid-uuid'}]} handler_response = fixture_client.post( '/v2/identity/self/tenant-access', json=payload, ) assert handler_response.status_code == 200 @patch('permissions.utils.api_utils.g') @patch('permissions.handlers.identity_v2_handlers.g') def test_check_my_access_to_tenants_unauthorized(mock_g, mock_g2, app_context, fixture_client): """Test POST /v2/identity/self/tenant-access unauthorized.""" mock_g.request_context.jwt_identity_id = None mock_g2.request_context.jwt_identity_id = None payload = {'tenants': [{'tenant_type': 'account', 'tenant_uuid': 'valid-uuid'}]} handler_response = fixture_client.post( '/v2/identity/self/tenant-access', json=payload, ) assert handler_response.status_code == 401 assert handler_response.json == { 'code': 'authorization_error', 'message': 'Request context has no identity uuid.', } @patch('permissions.utils.api_utils.g') @patch('permissions.handlers.identity_v2_handlers.g') @patch('permissions.models.identity.get_identity_settings_profile') def test_check_my_access_to_tenants_forbidden( get_identity_settings_profile, mock_g, mock_g2, app_context, fixture_client ): """Test POST /v2/identity/self/tenant-access forbidden.""" mock_g.request_context.jwt_identity_id = 'admin-uuid' mock_g2.request_context.profile_type = None get_identity_settings_profile.return_value = None payload = {'tenants': [{'tenant_type': 'account', 'tenant_uuid': 'valid-uuid'}]} handler_response = fixture_client.post('/v2/identity/self/tenant-access', json=payload) assert handler_response.status_code == 403 assert handler_response.json == {'code': 'authorization_error', 'message': 'user is forbidden'} @pytest.mark.parametrize( ['identity_id', 'settings_profile', 'is_super_admin', 'status_code'], [ pytest.param(None, None, None, 401, id='no identity id'), pytest.param('a-uuid', None, None, 403, id='no settings profile'), pytest.param('a-uuid', Mock(), False, 200, id='user is not super admin'), pytest.param('a-uuid', Mock(), True, 200, id='user is super admin'), ], ) @patch('permissions.logic.identity.is_identity_super_admin') @patch('permissions.models.identity.get_identity_settings_profile') @patch('permissions.handlers.identity_v2_handlers.g') @patch('permissions.utils.api_utils.g') def test_check_my_all_label_access( mock_g: MagicMock, mock_g2: MagicMock, get_settings_profile_mock: MagicMock, is_super_admin_mock: MagicMock, identity_id: str | None, settings_profile: Mock | None, is_super_admin: bool | None, status_code: int, app_context: flask.ctx.AppContext, fixture_client: flask.testing.FlaskClient, ) -> None: """Test the GET /identity/admin/all-label-access endpoint.""" mock_g.request_context.jwt_identity_id = identity_id mock_g2.request_context.jwt_identity_id = identity_id get_settings_profile_mock.return_value = settings_profile is_super_admin_mock.return_value = is_super_admin res = fixture_client.get('/v2/identity/self/all-label-access') assert res.status_code == status_code if identity_id: get_settings_profile_mock.assert_called_with(identity_id) else: get_settings_profile_mock.assert_not_called() if settings_profile: is_super_admin_mock.assert_called_with( request_context=mock_g2.request_context, identity_id=identity_id, ) assert res.json['has_access'] == is_super_admin else: is_super_admin_mock.assert_not_called() @patch('permissions.utils.api_utils.g') @patch('permissions.handlers.identity_v2_handlers.g') def test_get_adminable_tenants_for_identities_dataloader_authentication( mock_g, mock_g2, app_context, fixture_client ): """Test authentication for get_adminable_tenants_for_identities_dataloader endpoint.""" mock_g.request_context.jwt_identity_id = None mock_g2.request_context.jwt_identity_id = None payload = {'identity_uuids': [str(uuid.uuid4())]} assert fixture_client.post('/v2/identities/tenants/dataloader', json=payload).status_code == 401 @patch('permissions.utils.authorization.pdp_authorize_manage_employee') @patch('permissions.models.identity.get_identity_settings_profile') @patch('permissions.utils.api_utils.g') @patch('permissions.handlers.identity_v2_handlers.g') def test_get_adminable_tenants_for_identities_dataloader_authorization( mock_g, mock_g2, get_profile_mock, pdp_mock, app_context, fixture_client, ): """Test authorization: no SEAT role and no settings profile returns 403.""" mock_g.request_context.jwt_identity_id = 'some-uuid' mock_g2.request_context.jwt_identity_id = 'some-uuid' pdp_mock.return_value = False get_profile_mock.return_value = None payload = {'identity_uuids': [str(uuid.uuid4())]} assert fixture_client.post('/v2/identities/tenants/dataloader', json=payload).status_code == 403 @patch('permissions.logic.tenant.get_adminable_tenants_for_identities_dataloader') @patch('permissions.utils.authorization.pdp_authorize_manage_employee') @patch('permissions.models.identity.get_identity_settings_profile') @patch('permissions.utils.api_utils.g') @patch('permissions.handlers.identity_v2_handlers.g') def test_get_adminable_tenants_for_identities_dataloader_success( mock_g, mock_g2, get_profile_mock, pdp_mock, get_tenants_mock, app_context, fixture_client, ): """Test settings profile path returns data.""" admin_uuid = 'admin-uuid-123' mock_g.request_context.jwt_identity_id = admin_uuid mock_g2.request_context.jwt_identity_id = admin_uuid profile_id = 1331 pdp_mock.return_value = False get_profile_mock.return_value = ProfileInfo( profile_id=profile_id, profile_type='SettingsProfile', roles=[], uuid=admin_uuid ) user_uuid1 = str(uuid.uuid4()) user_uuid2 = str(uuid.uuid4()) tenant_uuid1 = str(uuid.uuid4()) tenant_uuid2 = str(uuid.uuid4()) # Logic layer returns list of IdentityTenantsWithV2Roles (or None for each identity) get_tenants_mock.return_value = [ IdentityTenantsWithV2Roles( identity_uuid=user_uuid1, tenants=[ TenantWithV2Roles( tenant_uuid=tenant_uuid1, tenant_type=TenantType.ACCOUNT, roles=['INSIGHTS_BASE_ROLE'], ) ], ), IdentityTenantsWithV2Roles( identity_uuid=user_uuid2, tenants=[ TenantWithV2Roles( tenant_uuid=tenant_uuid2, tenant_type=TenantType.SUBACCOUNT, roles=['WORKSTATION_ANALYTICS_BASE_ROLE'], ) ], ), ] payload = {'identity_uuids': [user_uuid1, user_uuid2]} result = fixture_client.post('/v2/identities/tenants/dataloader', json=payload) assert result.status_code == 200 response_data = json.loads(result.data) assert 'identities' in response_data assert len(response_data['identities']) == 2 # Each element has identity_uuid and tenants assert response_data['identities'][0]['identity_uuid'] == user_uuid1 assert response_data['identities'][0]['tenants'][0]['tenant_uuid'] == tenant_uuid1 assert response_data['identities'][1]['identity_uuid'] == user_uuid2 assert response_data['identities'][1]['tenants'][0]['tenant_uuid'] == tenant_uuid2 get_tenants_mock.assert_called_once() call_args = get_tenants_mock.call_args assert call_args[1]['admin_context'] == {'identity_id': admin_uuid, 'profile_id': profile_id} assert call_args[1]['is_seater'] is False # Verify identity_uuids are UUID objects assert all(isinstance(uid, uuid.UUID) for uid in call_args[1]['identity_uuids']) @patch('permissions.logic.tenant.get_adminable_tenants_for_identities_dataloader') @patch('permissions.utils.authorization.pdp_authorize_manage_employee') @patch('permissions.models.identity.get_identity_settings_profile') @patch('permissions.utils.api_utils.g') @patch('permissions.handlers.identity_v2_handlers.g') def test_get_adminable_tenants_for_identities_dataloader_with_none_values( mock_g, mock_g2, get_profile_mock, pdp_mock, get_tenants_mock, app_context, fixture_client, ): """Test dataloader endpoint returns None for identities without access.""" admin_uuid = 'admin-uuid-456' mock_g.request_context.jwt_identity_id = admin_uuid mock_g2.request_context.jwt_identity_id = admin_uuid profile_id = 2468 pdp_mock.return_value = False get_profile_mock.return_value = ProfileInfo( profile_id=profile_id, profile_type='SettingsProfile', roles=[], uuid=admin_uuid ) user_uuid1 = str(uuid.uuid4()) user_uuid2 = str(uuid.uuid4()) # Logic layer returns list of IdentityTenantsWithV2Roles (or None for each identity) get_tenants_mock.return_value = [ IdentityTenantsWithV2Roles( identity_uuid=user_uuid1, tenants=[ TenantWithV2Roles( tenant_uuid=str(uuid.uuid4()), tenant_type=TenantType.ACCOUNT, roles=['INSIGHTS_BASE_ROLE'], ) ], ), None, # No access to second identity ] payload = {'identity_uuids': [user_uuid1, user_uuid2]} result = fixture_client.post('/v2/identities/tenants/dataloader', json=payload) assert result.status_code == 200 response_data = json.loads(result.data) assert len(response_data['identities']) == 2 assert response_data['identities'][0] is not None assert response_data['identities'][0]['identity_uuid'] == user_uuid1 assert response_data['identities'][1] is None @patch('permissions.logic.tenant.get_adminable_tenants_for_identities_dataloader') @patch('permissions.utils.authorization.pdp_authorize_manage_employee') @patch('permissions.models.identity.get_identity_settings_profile') @patch('permissions.utils.api_utils.g') @patch('permissions.handlers.identity_v2_handlers.g') def test_get_adminable_tenants_for_identities_dataloader_empty_list( mock_g, mock_g2, get_profile_mock, pdp_mock, get_tenants_mock, app_context, fixture_client, ): """Test dataloader endpoint with empty identity list.""" admin_uuid = 'admin-uuid-789' mock_g.request_context.jwt_identity_id = admin_uuid mock_g2.request_context.jwt_identity_id = admin_uuid profile_id = 9876 pdp_mock.return_value = False get_profile_mock.return_value = ProfileInfo( profile_id=profile_id, profile_type='SettingsProfile', roles=[], uuid=admin_uuid ) get_tenants_mock.return_value = [] payload = {'identity_uuids': []} result = fixture_client.post('/v2/identities/tenants/dataloader', json=payload) assert result.status_code == 200 response_data = json.loads(result.data) assert response_data['identities'] == [] @patch('permissions.utils.api_utils.g') @patch('permissions.handlers.identity_v2_handlers.g') def test_get_adminable_tenants_for_identities_dataloader_invalid_uuid( mock_g, mock_g2, app_context, fixture_client ): """Test dataloader endpoint with invalid UUID format.""" mock_g.request_context.jwt_identity_id = 'admin-uuid' mock_g2.request_context.jwt_identity_id = 'admin-uuid' payload = {'identity_uuids': ['not-a-valid-uuid']} result = fixture_client.post('/v2/identities/tenants/dataloader', json=payload) assert result.status_code == 400 assert 'message' in result.json @patch('permissions.utils.api_utils.g') @patch('permissions.handlers.identity_v2_handlers.g') def test_get_adminable_tenants_for_identities_dataloader_missing_field( mock_g, mock_g2, app_context, fixture_client ): """Test dataloader endpoint with missing required field.""" mock_g.request_context.jwt_identity_id = 'admin-uuid' mock_g2.request_context.jwt_identity_id = 'admin-uuid' payload = {} result = fixture_client.post('/v2/identities/tenants/dataloader', json=payload) assert result.status_code == 400 assert 'message' in result.json # --- SEAT role (employee management) tests --- @patch('permissions.logic.tenant.get_adminable_tenants_for_identities_dataloader') @patch('permissions.utils.authorization.pdp_authorize_manage_employee') @patch('permissions.utils.api_utils.g') @patch('permissions.handlers.identity_v2_handlers.g') def test_dataloader_seat_role_returns_data_for_all_identities( mock_g, mock_g2, pdp_mock, get_tenants_mock, app_context, fixture_client, ): """SEAT role user returns data for all requested identities.""" admin_uuid = 'seat-admin-uuid' mock_g.request_context.jwt_identity_id = admin_uuid mock_g2.request_context.jwt_identity_id = admin_uuid pdp_mock.return_value = True user_uuid1 = str(uuid.uuid4()) user_uuid2 = str(uuid.uuid4()) tenant_uuid1 = str(uuid.uuid4()) tenant_uuid2 = str(uuid.uuid4()) get_tenants_mock.return_value = [ IdentityTenantsWithV2Roles( identity_uuid=user_uuid1, tenants=[ TenantWithV2Roles( tenant_uuid=tenant_uuid1, tenant_type=TenantType.ACCOUNT, roles=['ROLE_A'], ) ], ), IdentityTenantsWithV2Roles( identity_uuid=user_uuid2, tenants=[ TenantWithV2Roles( tenant_uuid=tenant_uuid2, tenant_type=TenantType.ACCOUNT, roles=['ROLE_B'], ) ], ), ] payload = {'identity_uuids': [user_uuid1, user_uuid2]} result = fixture_client.post('/v2/identities/tenants/dataloader', json=payload) assert result.status_code == 200 data = json.loads(result.data) assert len(data['identities']) == 2 assert data['identities'][0]['identity_uuid'] == user_uuid1 assert data['identities'][1]['identity_uuid'] == user_uuid2 # Verify is_seater=True was passed call_args = get_tenants_mock.call_args assert call_args[1]['is_seater'] is True assert call_args[1]['admin_context'] == {'identity_id': admin_uuid, 'profile_id': None} @patch('permissions.logic.tenant.get_adminable_tenants_for_identities_dataloader') @patch('permissions.utils.authorization.pdp_authorize_manage_employee') @patch('permissions.utils.api_utils.g') @patch('permissions.handlers.identity_v2_handlers.g') def test_dataloader_seat_role_returns_data_for_non_employees( mock_g, mock_g2, pdp_mock, get_tenants_mock, app_context, fixture_client, ): """SEAT role user returns data for non-employee identities (no filtering).""" admin_uuid = 'seat-admin-uuid' mock_g.request_context.jwt_identity_id = admin_uuid mock_g2.request_context.jwt_identity_id = admin_uuid pdp_mock.return_value = True user_uuid1 = str(uuid.uuid4()) user_uuid2 = str(uuid.uuid4()) tenant_uuid1 = str(uuid.uuid4()) tenant_uuid2 = str(uuid.uuid4()) get_tenants_mock.return_value = [ IdentityTenantsWithV2Roles( identity_uuid=user_uuid1, tenants=[ TenantWithV2Roles( tenant_uuid=tenant_uuid1, tenant_type=TenantType.ACCOUNT, roles=['ROLE_A'], ) ], ), IdentityTenantsWithV2Roles( identity_uuid=user_uuid2, tenants=[ TenantWithV2Roles( tenant_uuid=tenant_uuid2, tenant_type=TenantType.ACCOUNT, roles=['ROLE_B'], ) ], ), ] payload = {'identity_uuids': [user_uuid1, user_uuid2]} result = fixture_client.post('/v2/identities/tenants/dataloader', json=payload) assert result.status_code == 200 data = json.loads(result.data) assert len(data['identities']) == 2 assert data['identities'][0]['identity_uuid'] == user_uuid1 assert data['identities'][0]['tenants'][0]['tenant_uuid'] == tenant_uuid1 assert data['identities'][1]['identity_uuid'] == user_uuid2 assert data['identities'][1]['tenants'][0]['tenant_uuid'] == tenant_uuid2 @patch('permissions.logic.tenant.get_adminable_tenants_for_identities_dataloader') @patch('permissions.utils.authorization.pdp_authorize_manage_employee') @patch('permissions.models.identity.get_identity_settings_profile') @patch('permissions.utils.api_utils.g') @patch('permissions.handlers.identity_v2_handlers.g') def test_dataloader_settings_profile_returns_all_identities( mock_g, mock_g2, get_profile_mock, pdp_mock, get_tenants_mock, app_context, fixture_client, ): """Settings profile path returns data for all identities without employee filtering.""" admin_uuid = 'settings-admin-uuid' mock_g.request_context.jwt_identity_id = admin_uuid mock_g2.request_context.jwt_identity_id = admin_uuid profile_id = 5678 pdp_mock.return_value = False get_profile_mock.return_value = ProfileInfo( profile_id=profile_id, profile_type='SettingsProfile', roles=[], uuid=admin_uuid ) employee_uuid = str(uuid.uuid4()) non_employee_uuid = str(uuid.uuid4()) tenant_uuid1 = str(uuid.uuid4()) tenant_uuid2 = str(uuid.uuid4()) get_tenants_mock.return_value = [ IdentityTenantsWithV2Roles( identity_uuid=employee_uuid, tenants=[ TenantWithV2Roles( tenant_uuid=tenant_uuid1, tenant_type=TenantType.ACCOUNT, roles=['ROLE_A'], ) ], ), IdentityTenantsWithV2Roles( identity_uuid=non_employee_uuid, tenants=[ TenantWithV2Roles( tenant_uuid=tenant_uuid2, tenant_type=TenantType.ACCOUNT, roles=['ROLE_B'], ) ], ), ] payload = {'identity_uuids': [employee_uuid, non_employee_uuid]} result = fixture_client.post('/v2/identities/tenants/dataloader', json=payload) assert result.status_code == 200 data = json.loads(result.data) assert len(data['identities']) == 2 assert data['identities'][0]['identity_uuid'] == employee_uuid assert data['identities'][0]['tenants'][0]['tenant_uuid'] == tenant_uuid1 assert data['identities'][1]['identity_uuid'] == non_employee_uuid assert data['identities'][1]['tenants'][0]['tenant_uuid'] == tenant_uuid2