"""Tests for Identity endpoints.""" import uuid from typing import Any, Dict from unittest.mock import AsyncMock, MagicMock, patch import pytest from cerbos.sdk.client import AsyncCerbosClient from fastapi import FastAPI from fastapi.testclient import TestClient from splitio.client.factory import Client as SplitioClient from pdp.connectors.ows_account import OwsAccountClient from pdp.connectors.ows_participant import OwsParticipantClient from pdp.connectors.redis_client import RedisConnector from pdp.constants.constants import ( AUTH_EFFECT_ALLOW, AUTH_EFFECT_DENY, USER_TYPE_HUMAN, USER_TYPE_MACHINE, TenantType, ) from pdp.fastapi.auth import ( get_principal_pdp_tenant_roles_from_scope, impersonated_by_identity_uuid_from_scope, user_type_from_scope, ) from pdp.fastapi.schemas.cache import ( IdentityAllowedTenantsCacheObject, IdentityRolesResponseCacheObject, ) from pdp.fastapi.schemas.check_resource_type_actions import ( CheckResourceTypeAction, CheckResourceTypeActionResult, CheckResourceTypeActionsRequest, CheckResourceTypeActionsResponse, ) from pdp.fastapi.schemas.check_resources import CheckResourcesRequest from pdp.fastapi.schemas.deactivation import ( DeactivateAllResponse, DeactivateOneResponse, DeactivationSummary, ) from pdp.fastapi.schemas.get_allowed_tenants import ( AllowedTenant, GetAllowedTenantsRequest, GetAllowedTenantsResponse, ) from pdp.fastapi.schemas.identity import ( PaginationCursor, Role, RolesResponse, TenantRoles, TenantRolesMapValidator, ) from pdp.fastapi.schemas.principal import Principal @pytest.fixture() def roles_by_identity_response(tenant_1_uuid_as_string: str) -> Dict[str, Any]: """Reusable response representing an identity's tenants/roles.""" return { "cursor": {"cursor": None, "shorthand": None}, "tenants": { tenant_1_uuid_as_string: { "tenant_type": "account", "tenant_uuid": tenant_1_uuid_as_string, "roles": [{"role": "settings_admin"}], } }, } @patch.object(CheckResourcesRequest, "update_with_id_to_uuid_exchange") @patch("pdp.fastapi.routers.identity.cerbos.check_resources") @patch("pdp.fastapi.routers.identity.BooleanFeature") def test_check_my_resources_when_send_impersonated_by_ff_disabled( mock_boolean_feature: AsyncMock, mock_cerbos_check_resources: AsyncMock, mock_update_with_id_to_uuid_exchange: AsyncMock, mock_ows_account_client: OwsAccountClient, mock_ows_participant_client: OwsParticipantClient, mock_redis_connector: RedisConnector, test_client: TestClient, identity_uuid: str, identity_uuid_as_uuid: uuid.UUID, mock_pdp_tenant_roles: Dict[uuid.UUID, TenantRoles], mock_async_cerbos_client: AsyncCerbosClient, mock_splitio_client: MagicMock, ) -> None: """Test /identity/self/check/resources/ handler when the pp_send_impersonated_by_identity_uuid feature is disabled. """ mock_boolean_feature_instance = mock_boolean_feature.return_value mock_boolean_feature_instance.is_on_for_identity.return_value = False mock_request_body = { "resources": [ { "resource": { "resource_id": "123", "resource_type": "fan_data_list", "attributes": {"tenant_type": "account", "tenant_uuid": "fail now"}, }, "action": "view", } ] } mock_cerbos_check_resources_response = { "request_id": "hello", "resources": [ { "resource": { "resource_id": "123", "resource_type": "Identity", "attributes": {"attribute": "1"}, }, "action": "view", "effect": "deny", "errors": {"validation_errors": []}, } ], } mock_cerbos_check_resources.return_value = mock_cerbos_check_resources_response response = test_client.post( "/identity/self/check/resources/", json=mock_request_body ) assert response.status_code == 200 assert response.json() == mock_cerbos_check_resources_response mock_boolean_feature.assert_called_once_with( client=mock_splitio_client, feature_name="pp_send_impersonated_by_identity_uuid", ) mock_boolean_feature_instance.is_on_for_identity.assert_called_once_with( identity_uuid, ) mock_cerbos_check_resources.assert_awaited_once_with( identity_uuid=identity_uuid, check_resources_request=CheckResourcesRequest(**mock_request_body), pdp_tenant_roles=mock_pdp_tenant_roles, cerbos_client=mock_async_cerbos_client, include_resource_attributes=False, authenticated_identity_uuid=identity_uuid_as_uuid, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, redis_connector=mock_redis_connector, splitio_client=mock_splitio_client, ) mock_update_with_id_to_uuid_exchange.assert_awaited_once_with( redis_connector=mock_redis_connector, ows_account_client=mock_ows_account_client, ) @patch.object(CheckResourcesRequest, "update_with_id_to_uuid_exchange") @patch("pdp.fastapi.routers.identity.cerbos.check_resources") @patch("pdp.fastapi.routers.identity.BooleanFeature") def test_check_my_resource( mock_boolean_feature: AsyncMock, mock_cerbos_check_resources: AsyncMock, mock_update_with_id_to_uuid_exchange: AsyncMock, mock_ows_account_client: OwsAccountClient, mock_ows_participant_client: OwsParticipantClient, mock_redis_connector: RedisConnector, test_client: TestClient, identity_uuid: str, identity_uuid_as_uuid: uuid.UUID, mock_pdp_tenant_roles: Dict[uuid.UUID, TenantRoles], mock_async_cerbos_client: AsyncCerbosClient, mock_splitio_client: MagicMock, ) -> None: """Test /identity/self/check/resources/ handler.""" mock_boolean_feature_instance = mock_boolean_feature.return_value mock_boolean_feature_instance.is_on_for_identity.return_value = True mock_request_body = { "resources": [ { "resource": { "resource_id": "123", "resource_type": "fan_data_list", "attributes": {"tenant_type": "account", "tenant_uuid": "fail now"}, }, "action": "view", } ] } mock_cerbos_check_resources_response = { "request_id": "hello", "resources": [ { "resource": { "resource_id": "123", "resource_type": "Identity", "attributes": {"attribute": "1"}, }, "action": "view", "effect": "deny", "errors": {"validation_errors": []}, } ], } mock_cerbos_check_resources.return_value = mock_cerbos_check_resources_response response = test_client.post( "/identity/self/check/resources/", json=mock_request_body ) assert response.status_code == 200 assert response.json() == mock_cerbos_check_resources_response mock_boolean_feature.assert_called_once_with( client=mock_splitio_client, feature_name="pp_send_impersonated_by_identity_uuid", ) mock_boolean_feature_instance.is_on_for_identity.assert_called_once_with( identity_uuid, ) mock_cerbos_check_resources.assert_awaited_once_with( identity_uuid=identity_uuid, check_resources_request=CheckResourcesRequest(**mock_request_body), pdp_tenant_roles=mock_pdp_tenant_roles, cerbos_client=mock_async_cerbos_client, include_resource_attributes=False, authenticated_identity_uuid=identity_uuid_as_uuid, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, redis_connector=mock_redis_connector, principal=Principal( identity_uuid=identity_uuid_as_uuid, user_type="human", pdp_tenant_roles=mock_pdp_tenant_roles, ), splitio_client=mock_splitio_client, ) mock_update_with_id_to_uuid_exchange.assert_awaited_once_with( redis_connector=mock_redis_connector, ows_account_client=mock_ows_account_client, ) @patch.object(CheckResourcesRequest, "update_with_id_to_uuid_exchange") @patch("pdp.fastapi.routers.identity.cerbos.check_resources") @patch("pdp.fastapi.routers.identity.BooleanFeature") def test_check_my_resource_uses_impersonated_by_identity_and_user_type( mock_boolean_feature: AsyncMock, mock_cerbos_check_resources: AsyncMock, mock_update_with_id_to_uuid_exchange: AsyncMock, mock_ows_account_client: OwsAccountClient, mock_ows_participant_client: OwsParticipantClient, mock_redis_connector: RedisConnector, app: FastAPI, test_client: TestClient, identity_uuid: str, identity_uuid_as_uuid: uuid.UUID, mock_pdp_tenant_roles: Dict[uuid.UUID, TenantRoles], mock_async_cerbos_client: AsyncCerbosClient, mock_splitio_client: MagicMock, ) -> None: """Test /identity/self/check/resources/ handler when the pp_send_impersonated_by_identity_uuid feature is enabled and an impersonated_by_identity_uuid is present. """ impersonated_by_identity_uuid = uuid.uuid4() app.dependency_overrides[impersonated_by_identity_uuid_from_scope] = ( lambda: impersonated_by_identity_uuid ) app.dependency_overrides[user_type_from_scope] = lambda: USER_TYPE_MACHINE mock_boolean_feature_instance = mock_boolean_feature.return_value mock_boolean_feature_instance.is_on_for_identity.return_value = True mock_request_body = { "resources": [ { "resource": { "resource_id": "123", "resource_type": "fan_data_list", "attributes": {"tenant_type": "account", "tenant_uuid": "fail now"}, }, "action": "view", } ] } mock_cerbos_check_resources_response = { "request_id": "hello", "resources": [ { "resource": { "resource_id": "123", "resource_type": "Identity", "attributes": {"attribute": "1"}, }, "action": "view", "effect": "deny", "errors": {"validation_errors": []}, } ], } mock_cerbos_check_resources.return_value = mock_cerbos_check_resources_response response = test_client.post( "/identity/self/check/resources/", json=mock_request_body ) assert response.status_code == 200 assert response.json() == mock_cerbos_check_resources_response mock_boolean_feature.assert_called_once_with( client=mock_splitio_client, feature_name="pp_send_impersonated_by_identity_uuid", ) mock_boolean_feature_instance.is_on_for_identity.assert_called_once_with( str(impersonated_by_identity_uuid), ) mock_cerbos_check_resources.assert_awaited_once_with( identity_uuid=identity_uuid, check_resources_request=CheckResourcesRequest(**mock_request_body), pdp_tenant_roles=mock_pdp_tenant_roles, cerbos_client=mock_async_cerbos_client, include_resource_attributes=False, authenticated_identity_uuid=identity_uuid_as_uuid, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, redis_connector=mock_redis_connector, principal=Principal( identity_uuid=identity_uuid_as_uuid, user_type=USER_TYPE_MACHINE, pdp_tenant_roles=mock_pdp_tenant_roles, impersonated_by_identity_uuid=impersonated_by_identity_uuid, ), splitio_client=mock_splitio_client, ) mock_update_with_id_to_uuid_exchange.assert_awaited_once_with( redis_connector=mock_redis_connector, ows_account_client=mock_ows_account_client, ) @patch("pdp.fastapi.routers.identity.identity.check_resource_type_actions") @patch("pdp.fastapi.routers.identity.BooleanFeature") async def test_check_my_resource_type_actions_when_send_impersonated_by_ff_disabled( mock_boolean_feature: AsyncMock, mock_check_resource_type_actions: AsyncMock, test_client: TestClient, mock_pdp_tenant_roles: Dict[uuid.UUID, TenantRoles], identity_uuid: str, identity_uuid_as_uuid: uuid.UUID, mock_async_cerbos_client: AsyncCerbosClient, mock_ows_account_client: OwsAccountClient, mock_ows_participant_client: OwsParticipantClient, mock_redis_connector: RedisConnector, mock_splitio_client: MagicMock, ) -> None: """Test /identity/self/check/resource-type-actions/ handler.""" mock_boolean_feature_instance = mock_boolean_feature.return_value mock_boolean_feature_instance.is_on_for_identity.return_value = False mock_request_body_json = { "resource_type_actions": [ { "resource_type": "audience", "action": "create", }, { "resource_type": "audience", "action": "view", }, ] } mock_request_body_pydantic = CheckResourceTypeActionsRequest( resource_type_actions=[ CheckResourceTypeAction( resource_type="audience", action="create", ), CheckResourceTypeAction( resource_type="audience", action="view", ), ] ) mock_check_resource_type_actions.return_value = CheckResourceTypeActionsResponse( resource_type_actions=[ CheckResourceTypeActionResult( resource_type="audience", action="create", effect=AUTH_EFFECT_ALLOW, errors={}, ), CheckResourceTypeActionResult( resource_type="audience", action="view", effect=AUTH_EFFECT_DENY, errors={}, ), ] ) response = test_client.post( "identity/self/check/resource-type-actions/", json=mock_request_body_json, ) assert response.status_code == 200 mock_boolean_feature.assert_called_once_with( client=mock_splitio_client, feature_name="pp_send_impersonated_by_identity_uuid", ) mock_boolean_feature_instance.is_on_for_identity.assert_called_once_with( identity_uuid, ) mock_check_resource_type_actions.assert_awaited_once_with( request=mock_request_body_pydantic, identity_uuid=identity_uuid, pdp_tenant_roles=mock_pdp_tenant_roles, cerbos_client=mock_async_cerbos_client, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, redis_connector=mock_redis_connector, splitio_client=mock_splitio_client, authenticated_identity_uuid=identity_uuid_as_uuid, ) assert response.json() == { "resource_type_actions": [ { "resource_type": "audience", "action": "create", "effect": "allow", "errors": {}, }, { "resource_type": "audience", "action": "view", "effect": "deny", "errors": {}, }, ] } @patch("pdp.fastapi.routers.identity.identity.check_resource_type_actions") @patch("pdp.fastapi.routers.identity.BooleanFeature") async def test_check_my_resource_type_actions( mock_boolean_feature: AsyncMock, mock_check_resource_type_actions: AsyncMock, test_client: TestClient, mock_pdp_tenant_roles: Dict[uuid.UUID, TenantRoles], identity_uuid: str, identity_uuid_as_uuid: uuid.UUID, mock_async_cerbos_client: AsyncCerbosClient, mock_ows_account_client: OwsAccountClient, mock_ows_participant_client: OwsParticipantClient, mock_redis_connector: RedisConnector, mock_splitio_client: MagicMock, ) -> None: """Test /identity/self/check/resource-type-actions/ handler.""" mock_boolean_feature_instance = mock_boolean_feature.return_value mock_boolean_feature_instance.is_on_for_identity.return_value = True mock_request_body_json = { "resource_type_actions": [ { "resource_type": "audience", "action": "create", }, { "resource_type": "audience", "action": "view", }, ] } mock_request_body_pydantic = CheckResourceTypeActionsRequest( resource_type_actions=[ CheckResourceTypeAction( resource_type="audience", action="create", ), CheckResourceTypeAction( resource_type="audience", action="view", ), ] ) mock_check_resource_type_actions.return_value = CheckResourceTypeActionsResponse( resource_type_actions=[ CheckResourceTypeActionResult( resource_type="audience", action="create", effect=AUTH_EFFECT_ALLOW, errors={}, ), CheckResourceTypeActionResult( resource_type="audience", action="view", effect=AUTH_EFFECT_DENY, errors={}, ), ] ) response = test_client.post( "identity/self/check/resource-type-actions/", json=mock_request_body_json, ) assert response.status_code == 200 mock_boolean_feature.assert_called_once_with( client=mock_splitio_client, feature_name="pp_send_impersonated_by_identity_uuid", ) mock_boolean_feature_instance.is_on_for_identity.assert_called_once_with( identity_uuid, ) mock_check_resource_type_actions.assert_awaited_once_with( request=mock_request_body_pydantic, identity_uuid=identity_uuid, pdp_tenant_roles=mock_pdp_tenant_roles, cerbos_client=mock_async_cerbos_client, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, redis_connector=mock_redis_connector, splitio_client=mock_splitio_client, authenticated_identity_uuid=identity_uuid_as_uuid, principal=Principal( identity_uuid=identity_uuid_as_uuid, user_type="human", pdp_tenant_roles=mock_pdp_tenant_roles, ), ) assert response.json() == { "resource_type_actions": [ { "resource_type": "audience", "action": "create", "effect": "allow", "errors": {}, }, { "resource_type": "audience", "action": "view", "effect": "deny", "errors": {}, }, ] } @patch("pdp.fastapi.routers.identity.identity.check_resource_type_actions") @patch("pdp.fastapi.routers.identity.BooleanFeature") async def test_check_my_resource_type_actions_uses_impersonated_by_identity( mock_boolean_feature: AsyncMock, mock_check_resource_type_actions: AsyncMock, app: FastAPI, test_client: TestClient, mock_pdp_tenant_roles: Dict[uuid.UUID, TenantRoles], identity_uuid: str, identity_uuid_as_uuid: uuid.UUID, mock_async_cerbos_client: AsyncCerbosClient, mock_ows_account_client: OwsAccountClient, mock_ows_participant_client: OwsParticipantClient, mock_redis_connector: RedisConnector, mock_splitio_client: MagicMock, ) -> None: """Test /identity/self/check/resource-type-actions/ handler.""" impersonated_by_identity_uuid = uuid.uuid4() app.dependency_overrides[impersonated_by_identity_uuid_from_scope] = ( lambda: impersonated_by_identity_uuid ) mock_boolean_feature_instance = mock_boolean_feature.return_value mock_boolean_feature_instance.is_on_for_identity.return_value = True mock_request_body_json = { "resource_type_actions": [ { "resource_type": "audience", "action": "create", }, { "resource_type": "audience", "action": "view", }, ] } mock_request_body_pydantic = CheckResourceTypeActionsRequest( resource_type_actions=[ CheckResourceTypeAction( resource_type="audience", action="create", ), CheckResourceTypeAction( resource_type="audience", action="view", ), ] ) mock_check_resource_type_actions.return_value = CheckResourceTypeActionsResponse( resource_type_actions=[ CheckResourceTypeActionResult( resource_type="audience", action="create", effect=AUTH_EFFECT_ALLOW, errors={}, ), CheckResourceTypeActionResult( resource_type="audience", action="view", effect=AUTH_EFFECT_DENY, errors={}, ), ] ) response = test_client.post( "identity/self/check/resource-type-actions/", json=mock_request_body_json, ) assert response.status_code == 200 mock_boolean_feature.assert_called_once_with( client=mock_splitio_client, feature_name="pp_send_impersonated_by_identity_uuid", ) mock_boolean_feature_instance.is_on_for_identity.assert_called_once_with( str(impersonated_by_identity_uuid), ) mock_check_resource_type_actions.assert_awaited_once_with( request=mock_request_body_pydantic, identity_uuid=identity_uuid, pdp_tenant_roles=mock_pdp_tenant_roles, cerbos_client=mock_async_cerbos_client, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, redis_connector=mock_redis_connector, splitio_client=mock_splitio_client, authenticated_identity_uuid=identity_uuid_as_uuid, principal=Principal( identity_uuid=identity_uuid_as_uuid, user_type="human", pdp_tenant_roles=mock_pdp_tenant_roles, impersonated_by_identity_uuid=impersonated_by_identity_uuid, ), ) assert response.json() == { "resource_type_actions": [ { "resource_type": "audience", "action": "create", "effect": "allow", "errors": {}, }, { "resource_type": "audience", "action": "view", "effect": "deny", "errors": {}, }, ] } @patch("pdp.fastapi.routers.identity.identity.check_resource_type_actions") @patch("pdp.fastapi.routers.identity.BooleanFeature") async def test_check_my_resource_type_actions_uses_user_type( mock_boolean_feature: AsyncMock, mock_check_resource_type_actions: AsyncMock, app: FastAPI, test_client: TestClient, mock_pdp_tenant_roles: Dict[uuid.UUID, TenantRoles], identity_uuid: str, identity_uuid_as_uuid: uuid.UUID, mock_async_cerbos_client: AsyncCerbosClient, mock_ows_account_client: OwsAccountClient, mock_ows_participant_client: OwsParticipantClient, mock_redis_connector: RedisConnector, mock_splitio_client: MagicMock, ) -> None: """Test /identity/self/check/resource-type-actions/ handler.""" app.dependency_overrides[user_type_from_scope] = lambda: "machine" mock_boolean_feature_instance = mock_boolean_feature.return_value mock_boolean_feature_instance.is_on_for_identity.return_value = True mock_request_body_json = { "resource_type_actions": [ { "resource_type": "audience", "action": "create", }, { "resource_type": "audience", "action": "view", }, ] } mock_request_body_pydantic = CheckResourceTypeActionsRequest( resource_type_actions=[ CheckResourceTypeAction( resource_type="audience", action="create", ), CheckResourceTypeAction( resource_type="audience", action="view", ), ] ) mock_check_resource_type_actions.return_value = CheckResourceTypeActionsResponse( resource_type_actions=[ CheckResourceTypeActionResult( resource_type="audience", action="create", effect=AUTH_EFFECT_ALLOW, errors={}, ), CheckResourceTypeActionResult( resource_type="audience", action="view", effect=AUTH_EFFECT_DENY, errors={}, ), ] ) response = test_client.post( "identity/self/check/resource-type-actions/", json=mock_request_body_json, ) assert response.status_code == 200 mock_boolean_feature.assert_called_once_with( client=mock_splitio_client, feature_name="pp_send_impersonated_by_identity_uuid", ) mock_boolean_feature_instance.is_on_for_identity.assert_called_once_with( identity_uuid, ) mock_check_resource_type_actions.assert_awaited_once_with( request=mock_request_body_pydantic, identity_uuid=identity_uuid, pdp_tenant_roles=mock_pdp_tenant_roles, cerbos_client=mock_async_cerbos_client, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, redis_connector=mock_redis_connector, splitio_client=mock_splitio_client, authenticated_identity_uuid=identity_uuid_as_uuid, principal=Principal( identity_uuid=identity_uuid_as_uuid, user_type="machine", pdp_tenant_roles=mock_pdp_tenant_roles, ), ) assert response.json() == { "resource_type_actions": [ { "resource_type": "audience", "action": "create", "effect": "allow", "errors": {}, }, { "resource_type": "audience", "action": "view", "effect": "deny", "errors": {}, }, ] } @pytest.mark.parametrize( "mock_get_object_from_cache_response, " "expect_mock_update_with_uuid_to_id_exchange_called, " "expect_get_allowed_tenants_called", [ pytest.param( None, True, True, id="""Cache is empty, continue to call get_allowed_tenants, and perform uuid to id exchange. """, ), pytest.param( GetAllowedTenantsResponse( resource_type="audience", action="create", tenants=[ AllowedTenant(tenant_type="account", tenant_uuid=str(uuid.uuid4())), ], ), False, False, id="""Cache returns a value. This should be returned and get_allowed_tenants function should not be called.""", ), ], ) @patch("pdp.fastapi.routers.identity.PydanticSchemaSerializer") @patch("pdp.fastapi.routers.identity.save_cache_object") @patch("pdp.fastapi.routers.identity.get_object_from_cache") @patch("pdp.fastapi.routers.identity.identity") @patch("pdp.fastapi.routers.identity.BooleanFeature") @patch.object(GetAllowedTenantsResponse, "update_with_uuid_to_id_exchange") async def test_get_my_allowed_tenants_when_send_impersonated_by_ff_disabled( mock_update_with_uuid_to_id_exchange: AsyncMock, mock_boolean_feature: AsyncMock, mock_identity: MagicMock, mock_get_object_from_cache: AsyncMock, mock_save_cache_object: AsyncMock, mock_serializer: MagicMock, mock_get_object_from_cache_response: GetAllowedTenantsResponse | None, expect_mock_update_with_uuid_to_id_exchange_called: bool, expect_get_allowed_tenants_called: bool, test_client: TestClient, identity_uuid: str, identity_uuid_as_uuid: uuid.UUID, mock_pdp_tenant_roles: Dict[uuid.UUID, TenantRoles], mock_ows_account_client: OwsAccountClient, mock_ows_participant_client: OwsParticipantClient, mock_async_cerbos_client: AsyncCerbosClient, mock_redis_connector: RedisConnector, mock_splitio_client: SplitioClient, ) -> None: """Test /identity/self/allowed-tenants/ handler.""" mock_boolean_feature_instance = mock_boolean_feature.return_value mock_boolean_feature_instance.is_on_for_identity.return_value = False request_body = { "resource_type": "audience", "action": "create", } mock_get_allowed_tenants_response = { "resource_type": "audience", "action": "create", "tenants": [], } mock_get_object_from_cache.return_value = mock_get_object_from_cache_response mock_identity.get_allowed_tenants = AsyncMock( return_value=GetAllowedTenantsResponse.model_validate( mock_get_allowed_tenants_response ) ) response = test_client.post( "identity/self/allowed-tenants/", json=request_body, ) mock_get_object_from_cache.assert_called_once_with( f"allowed_tenants@action#create|identity_uuid#{identity_uuid}|resource_type#audience", # noqa: E501 redis_connector=mock_redis_connector, serializer=mock_serializer(IdentityAllowedTenantsCacheObject), cache_attribute_name="get_allowed_tenants_response", ) if expect_get_allowed_tenants_called: mock_identity.get_allowed_tenants.assert_called_once_with( request=GetAllowedTenantsRequest(**request_body), identity_uuid=identity_uuid, pdp_tenant_roles=mock_pdp_tenant_roles, cerbos_client=mock_async_cerbos_client, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, redis_connector=mock_redis_connector, authenticated_identity_uuid=identity_uuid_as_uuid, splitio_client=mock_splitio_client, ) assert response.json() == mock_get_allowed_tenants_response else: mock_identity.get_allowed_tenants.assert_not_called() assert mock_get_object_from_cache_response assert response.json() == mock_get_object_from_cache_response.model_dump( mode="json", ) if expect_mock_update_with_uuid_to_id_exchange_called: mock_update_with_uuid_to_id_exchange.assert_awaited_once_with( redis_connector=mock_redis_connector, ows_account_client=mock_ows_account_client, ) else: mock_update_with_uuid_to_id_exchange.assert_not_called() if mock_get_object_from_cache_response is None: mock_save_cache_object.assert_called_once_with( cache_object=IdentityAllowedTenantsCacheObject( identity_uuid=identity_uuid, resource_type="audience", action="create", get_allowed_tenants_response=GetAllowedTenantsResponse( resource_type="audience", action="create", tenants=[], ), ), cache_model_type=IdentityAllowedTenantsCacheObject, redis_connector=mock_redis_connector, ) else: mock_save_cache_object.assert_not_called() @patch("pdp.fastapi.routers.identity.PydanticSchemaSerializer") @patch("pdp.fastapi.routers.identity.save_cache_object") @patch("pdp.fastapi.routers.identity.get_object_from_cache") @patch("pdp.fastapi.routers.identity.identity") @patch("pdp.fastapi.routers.identity.BooleanFeature") @patch.object(GetAllowedTenantsResponse, "update_with_uuid_to_id_exchange") async def test_get_my_allowed_tenants( mock_update_with_uuid_to_id_exchange: AsyncMock, mock_boolean_feature: AsyncMock, mock_identity: MagicMock, mock_get_object_from_cache: AsyncMock, mock_save_cache_object: AsyncMock, mock_serializer: MagicMock, test_client: TestClient, identity_uuid: str, identity_uuid_as_uuid: uuid.UUID, mock_pdp_tenant_roles: Dict[uuid.UUID, TenantRoles], mock_ows_account_client: OwsAccountClient, mock_ows_participant_client: OwsParticipantClient, mock_async_cerbos_client: AsyncCerbosClient, mock_redis_connector: RedisConnector, mock_splitio_client: SplitioClient, ) -> None: """Test /identity/self/allowed-tenants/ handler.""" mock_boolean_feature_instance = mock_boolean_feature.return_value mock_boolean_feature_instance.is_on_for_identity.return_value = True request_body = { "resource_type": "audience", "action": "create", } mock_get_allowed_tenants_response = { "resource_type": "audience", "action": "create", "tenants": [], } mock_get_object_from_cache.return_value = None mock_identity.get_allowed_tenants = AsyncMock( return_value=GetAllowedTenantsResponse.model_validate( mock_get_allowed_tenants_response ) ) response = test_client.post( "identity/self/allowed-tenants/", json=request_body, ) mock_get_object_from_cache.assert_called_once_with( f"allowed_tenants@action#create|identity_uuid#{identity_uuid}|resource_type#audience", # noqa: E501 redis_connector=mock_redis_connector, serializer=mock_serializer(IdentityAllowedTenantsCacheObject), cache_attribute_name="get_allowed_tenants_response", ) mock_identity.get_allowed_tenants.assert_called_once_with( request=GetAllowedTenantsRequest(**request_body), identity_uuid=identity_uuid, pdp_tenant_roles=mock_pdp_tenant_roles, cerbos_client=mock_async_cerbos_client, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, redis_connector=mock_redis_connector, authenticated_identity_uuid=identity_uuid_as_uuid, splitio_client=mock_splitio_client, principal=Principal( identity_uuid=identity_uuid_as_uuid, user_type="human", pdp_tenant_roles=mock_pdp_tenant_roles, ), ) assert response.json() == mock_get_allowed_tenants_response mock_update_with_uuid_to_id_exchange.assert_awaited_once_with( redis_connector=mock_redis_connector, ows_account_client=mock_ows_account_client, ) mock_save_cache_object.assert_called_once_with( cache_object=IdentityAllowedTenantsCacheObject( identity_uuid=identity_uuid, resource_type="audience", action="create", get_allowed_tenants_response=GetAllowedTenantsResponse( resource_type="audience", action="create", tenants=[], ), ), cache_model_type=IdentityAllowedTenantsCacheObject, redis_connector=mock_redis_connector, ) mock_boolean_feature_instance.is_on_for_identity.assert_called_once_with( identity_uuid ) @patch("pdp.fastapi.routers.identity.PydanticSchemaSerializer") @patch("pdp.fastapi.routers.identity.save_cache_object") @patch("pdp.fastapi.routers.identity.get_object_from_cache") @patch("pdp.fastapi.routers.identity.identity") @patch("pdp.fastapi.routers.identity.BooleanFeature") @patch.object(GetAllowedTenantsResponse, "update_with_uuid_to_id_exchange") async def test_get_my_allowed_tenants_uses_impersonated_by_identity( mock_update_with_uuid_to_id_exchange: AsyncMock, mock_boolean_feature: AsyncMock, mock_identity: MagicMock, mock_get_object_from_cache: AsyncMock, mock_save_cache_object: AsyncMock, mock_serializer: MagicMock, app: FastAPI, test_client: TestClient, identity_uuid: str, identity_uuid_as_uuid: uuid.UUID, mock_pdp_tenant_roles: Dict[uuid.UUID, TenantRoles], mock_ows_account_client: OwsAccountClient, mock_ows_participant_client: OwsParticipantClient, mock_async_cerbos_client: AsyncCerbosClient, mock_redis_connector: RedisConnector, mock_splitio_client: SplitioClient, ) -> None: """Test /identity/self/allowed-tenants/ handler.""" impersonated_by_identity_uuid = uuid.uuid4() app.dependency_overrides[impersonated_by_identity_uuid_from_scope] = ( lambda: impersonated_by_identity_uuid ) mock_boolean_feature_instance = mock_boolean_feature.return_value mock_boolean_feature_instance.is_on_for_identity.return_value = True request_body = { "resource_type": "audience", "action": "create", } mock_get_allowed_tenants_response = { "resource_type": "audience", "action": "create", "tenants": [], } mock_get_object_from_cache.return_value = None mock_identity.get_allowed_tenants = AsyncMock( return_value=GetAllowedTenantsResponse.model_validate( mock_get_allowed_tenants_response ) ) response = test_client.post( "identity/self/allowed-tenants/", json=request_body, ) mock_get_object_from_cache.assert_called_once_with( f"allowed_tenants@action#create|identity_uuid#{identity_uuid}|resource_type#audience", # noqa: E501 redis_connector=mock_redis_connector, serializer=mock_serializer(IdentityAllowedTenantsCacheObject), cache_attribute_name="get_allowed_tenants_response", ) mock_identity.get_allowed_tenants.assert_called_once_with( request=GetAllowedTenantsRequest(**request_body), identity_uuid=identity_uuid, pdp_tenant_roles=mock_pdp_tenant_roles, cerbos_client=mock_async_cerbos_client, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, redis_connector=mock_redis_connector, authenticated_identity_uuid=identity_uuid_as_uuid, splitio_client=mock_splitio_client, principal=Principal( identity_uuid=identity_uuid_as_uuid, user_type="human", pdp_tenant_roles=mock_pdp_tenant_roles, impersonated_by_identity_uuid=impersonated_by_identity_uuid, ), ) assert response.json() == mock_get_allowed_tenants_response mock_update_with_uuid_to_id_exchange.assert_awaited_once_with( redis_connector=mock_redis_connector, ows_account_client=mock_ows_account_client, ) mock_save_cache_object.assert_called_once_with( cache_object=IdentityAllowedTenantsCacheObject( identity_uuid=identity_uuid, resource_type="audience", action="create", get_allowed_tenants_response=GetAllowedTenantsResponse( resource_type="audience", action="create", tenants=[], ), ), cache_model_type=IdentityAllowedTenantsCacheObject, redis_connector=mock_redis_connector, ) @pytest.mark.parametrize( "mock_get_object_from_cache_response, " "expect_get_roles_by_identity_called, " "expected", [ pytest.param( None, True, RolesResponse(), id="Cache miss, so fetch and return from dynamodb", ), pytest.param( RolesResponse(), False, RolesResponse(), id="Cache hit, so return what is stored in cache", ), ], ) @patch("pdp.fastapi.routers.identity.PydanticSchemaSerializer") @patch("pdp.fastapi.routers.identity.save_cache_object") @patch("pdp.fastapi.routers.identity.get_object_from_cache") @patch("pdp.fastapi.routers.identity.identity") def test_get_my_roles( mock_identity: MagicMock, mock_get_object_from_cache: AsyncMock, mock_save_cache_object: AsyncMock, mock_serializer: MagicMock, mock_get_object_from_cache_response: RolesResponse | None, expect_get_roles_by_identity_called: bool, expected: RolesResponse, mock_identity_ddb_connector: MagicMock, mock_redis_connector: RedisConnector, test_client: TestClient, roles_by_identity_response: Dict[str, Any], identity_uuid: str, ) -> None: """Test /identity/self/roles/ handler.""" mock_get_object_from_cache.return_value = mock_get_object_from_cache_response mock_identity.get_roles_by_identity.return_value = RolesResponse.model_validate( roles_by_identity_response ) ows_response = test_client.get( "/identity/self/roles/", headers={"Authorization": "dummy"} ) mock_get_object_from_cache.assert_called_once_with( f"list_tenant_roles@principal_identity_uuid#{identity_uuid}|listed_identity_uuid#{identity_uuid}|cursor#", redis_connector=mock_redis_connector, serializer=mock_serializer(IdentityRolesResponseCacheObject), cache_attribute_name="roles_response", ) if not mock_get_object_from_cache_response: mock_save_cache_object.assert_called_once_with( cache_object=IdentityRolesResponseCacheObject( principal_identity_uuid=identity_uuid, listed_identity_uuid=identity_uuid, roles_response=RolesResponse.model_validate(roles_by_identity_response), ), cache_model_type=IdentityRolesResponseCacheObject, redis_connector=mock_redis_connector, ) else: mock_save_cache_object.assert_not_called() assert ows_response.status_code == 200 data = ows_response.json() assert data["errors"] == {} assert data["cursor"] == roles_by_identity_response["cursor"] if expect_get_roles_by_identity_called: mock_identity.get_roles_by_identity.assert_called_with( identity_uuid, cursor=None, identity_ddb_connector=mock_identity_ddb_connector, ) assert data["tenants"] == roles_by_identity_response["tenants"] else: mock_identity.get_roles_by_identity.assert_not_called() assert data["tenants"] == {} @patch("pdp.fastapi.routers.identity.PydanticSchemaSerializer") @patch("pdp.fastapi.routers.identity.save_cache_object") @patch("pdp.fastapi.routers.identity.get_object_from_cache") @patch("pdp.fastapi.routers.identity.filter_for_identity_tenants") @patch("pdp.fastapi.routers.identity.identity") async def test_get_roles_by_identity_with_cursor( mock_identity: MagicMock, mock_filter_for_identity_tenants: AsyncMock, mock_get_object_from_cache: AsyncMock, mock_save_cache_object: AsyncMock, mock_serializer: MagicMock, mock_identity_ddb_connector: MagicMock, identity_uuid: str, identity_uuid_as_uuid: uuid.UUID, mock_my_adminable_tenant_roles: Dict[uuid.UUID, TenantRoles], mock_pdp_tenant_roles: Dict[uuid.UUID, TenantRoles], test_client: TestClient, roles_by_identity_response: Dict[str, Any], mock_async_cerbos_client: AsyncCerbosClient, mock_ows_account_client: OwsAccountClient, mock_ows_participant_client: OwsParticipantClient, mock_redis_connector: RedisConnector, mock_splitio_client: SplitioClient, ) -> None: """Test /identity//roles/ handler.""" mock_identity.get_roles_by_identity.return_value = RolesResponse.model_validate( roles_by_identity_response ) mock_filter_for_identity_tenants.return_value = ( TenantRolesMapValidator.validate_python(roles_by_identity_response["tenants"]) ) requested_identity_uuid = uuid.uuid4() mock_get_object_from_cache.return_value = None ows_response = test_client.get( f"/identity/{requested_identity_uuid}/roles/?cursor=nextpagepls" ) mock_get_object_from_cache.assert_awaited_once_with( f"list_tenant_roles@principal_identity_uuid#{identity_uuid}|listed_identity_uuid#{requested_identity_uuid}|cursor#nextpagepls", # noqa: E501 redis_connector=mock_redis_connector, serializer=mock_serializer(IdentityRolesResponseCacheObject), cache_attribute_name="roles_response", ) mock_save_cache_object.assert_awaited_once_with( cache_object=IdentityRolesResponseCacheObject( principal_identity_uuid=identity_uuid, listed_identity_uuid=requested_identity_uuid, cursor="nextpagepls", roles_response=RolesResponse.model_validate(roles_by_identity_response), ), cache_model_type=IdentityRolesResponseCacheObject, redis_connector=mock_redis_connector, ) mock_identity.get_roles_by_identity.assert_called_with( str(requested_identity_uuid), cursor="nextpagepls", identity_ddb_connector=mock_identity_ddb_connector, ) mock_filter_for_identity_tenants.assert_called_once() mock_filter_for_identity_tenants.assert_called_once_with( requested_identity_uuid, TenantRolesMapValidator.validate_python(roles_by_identity_response["tenants"]), action="view", principal_identity_uuid=identity_uuid_as_uuid, pdp_tenant_roles=mock_pdp_tenant_roles, ows_permissions_tenant_roles=mock_my_adminable_tenant_roles, cerbos_client=mock_async_cerbos_client, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, redis_connector=mock_redis_connector, impersonated_by_identity_uuid=None, splitio_client=mock_splitio_client, user_type=USER_TYPE_HUMAN, ) assert ows_response.status_code == 200 data = ows_response.json() assert data assert data["errors"] == {} assert data["cursor"] == roles_by_identity_response["cursor"] assert data["tenants"] == roles_by_identity_response["tenants"] @patch("pdp.fastapi.routers.identity.PydanticSchemaSerializer") @patch("pdp.fastapi.routers.identity.save_cache_object") @patch("pdp.fastapi.routers.identity.get_object_from_cache") @patch("pdp.fastapi.routers.identity.filter_for_identity_tenants") @patch("pdp.fastapi.routers.identity.identity") async def test_get_roles_by_identity( mock_identity: MagicMock, mock_filter_for_identity_tenants: AsyncMock, mock_get_object_from_cache: AsyncMock, mock_save_cache_object: AsyncMock, mock_serializer: MagicMock, mock_identity_ddb_connector: MagicMock, identity_uuid: str, identity_uuid_as_uuid: uuid.UUID, mock_my_adminable_tenant_roles: Dict[uuid.UUID, TenantRoles], test_client: TestClient, roles_by_identity_response: Dict[str, Any], mock_pdp_tenant_roles: Dict[uuid.UUID, TenantRoles], mock_async_cerbos_client: AsyncCerbosClient, mock_ows_account_client: OwsAccountClient, mock_ows_participant_client: OwsParticipantClient, mock_redis_connector: RedisConnector, mock_splitio_client: MagicMock, ) -> None: """Test /identity//roles/ handler. This test asserts the application tries to get the response from cache, but since it is empty, must continue to make calls to DynamoDB, and filter the results with cerbos. """ mock_identity.get_roles_by_identity.return_value = RolesResponse.model_validate( roles_by_identity_response ) mock_filter_for_identity_tenants.return_value = ( TenantRolesMapValidator.validate_python(roles_by_identity_response["tenants"]) ) requested_identity_uuid = uuid.uuid4() mock_get_object_from_cache.return_value = None # Cache was empty response = test_client.get(f"/identity/{requested_identity_uuid}/roles/") mock_get_object_from_cache.assert_awaited_once_with( f"list_tenant_roles@principal_identity_uuid#{identity_uuid}|listed_identity_uuid#{requested_identity_uuid}|cursor#", # noqa: E501 redis_connector=mock_redis_connector, serializer=mock_serializer(IdentityRolesResponseCacheObject), cache_attribute_name="roles_response", ) mock_save_cache_object.assert_awaited_once_with( cache_object=IdentityRolesResponseCacheObject( principal_identity_uuid=identity_uuid, listed_identity_uuid=requested_identity_uuid, roles_response=RolesResponse.model_validate(roles_by_identity_response), ), cache_model_type=IdentityRolesResponseCacheObject, redis_connector=mock_redis_connector, ) mock_identity.get_roles_by_identity.assert_called_with( str(requested_identity_uuid), cursor=None, identity_ddb_connector=mock_identity_ddb_connector, ) mock_filter_for_identity_tenants.assert_called_once() mock_filter_for_identity_tenants.assert_called_once_with( requested_identity_uuid, TenantRolesMapValidator.validate_python(roles_by_identity_response["tenants"]), action="view", principal_identity_uuid=identity_uuid_as_uuid, pdp_tenant_roles=mock_pdp_tenant_roles, ows_permissions_tenant_roles=mock_my_adminable_tenant_roles, cerbos_client=mock_async_cerbos_client, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, redis_connector=mock_redis_connector, impersonated_by_identity_uuid=None, splitio_client=mock_splitio_client, user_type=USER_TYPE_HUMAN, ) assert response.status_code == 200 data = response.json() assert data["errors"] == {} assert data["cursor"] == roles_by_identity_response["cursor"] assert data["tenants"] == roles_by_identity_response["tenants"] @patch("pdp.fastapi.routers.identity.PydanticSchemaSerializer") @patch("pdp.fastapi.routers.identity.save_cache_object") @patch("pdp.fastapi.routers.identity.get_object_from_cache") @patch("pdp.fastapi.routers.identity.filter_for_identity_tenants") @patch("pdp.fastapi.routers.identity.identity") async def test_get_roles_by_identity_uses_cached_data( mock_identity: MagicMock, mock_filter_for_identity_tenants: AsyncMock, mock_get_object_from_cache: AsyncMock, mock_save_cache_object: AsyncMock, mock_serializer: MagicMock, identity_uuid: str, test_client: TestClient, roles_by_identity_response: Dict[str, Any], mock_redis_connector: RedisConnector, ) -> None: """Test /identity//roles/ handler. This test asserts the application tries to get the response from cache, and since it is a valid RolesResponse, returns immediately. Thus, no calls are made to fetch from DynamoDB or to filter results using cerbos. """ mock_identity.get_roles_by_identity.return_value = RolesResponse.model_validate( roles_by_identity_response ) mock_filter_for_identity_tenants.return_value = ( TenantRolesMapValidator.validate_python(roles_by_identity_response["tenants"]) ) requested_identity_uuid = uuid.uuid4() mock_get_object_from_cache.return_value = RolesResponse( tenants={}, cursor=PaginationCursor(cursor="go", shorthand="here") ) response = test_client.get(f"/identity/{requested_identity_uuid}/roles/") mock_get_object_from_cache.assert_awaited_once_with( f"list_tenant_roles@principal_identity_uuid#{identity_uuid}|listed_identity_uuid#{requested_identity_uuid}|cursor#", # noqa: E501 redis_connector=mock_redis_connector, serializer=mock_serializer(IdentityRolesResponseCacheObject), cache_attribute_name="roles_response", ) mock_identity.get_roles_by_identity.assert_not_called() mock_filter_for_identity_tenants.assert_not_called() mock_save_cache_object.assert_not_called() assert response.status_code == 200 data = response.json() assert data["errors"] == {} assert data["cursor"] == {"cursor": "go", "shorthand": "here"} assert data["tenants"] == {} @pytest.mark.parametrize( "impersonated_by_identity_uuid,user_type,expected_is_on_for_identity_uuid,expected_principal", [ pytest.param( None, "human", "ab123456-1234-4c2b-9c23-123ab4000a1b", Principal( identity_uuid="ab123456-1234-4c2b-9c23-123ab4000a1b", pdp_tenant_roles={}, user_type="human", ), id="Principal is not impersonated", ), pytest.param( None, "machine", "ab123456-1234-4c2b-9c23-123ab4000a1b", Principal( identity_uuid="ab123456-1234-4c2b-9c23-123ab4000a1b", pdp_tenant_roles={}, user_type="machine", ), id="Principal is a machine", ), pytest.param( uuid.UUID("7e1ba04d-b15f-4811-96ed-4320a599f595"), "machine", "7e1ba04d-b15f-4811-96ed-4320a599f595", Principal( identity_uuid="ab123456-1234-4c2b-9c23-123ab4000a1b", pdp_tenant_roles={}, user_type="machine", impersonated_by_identity_uuid="7e1ba04d-b15f-4811-96ed-4320a599f595", ), id="Principal is a machine impersonating", ), ], ) @patch("fastapi.BackgroundTasks.add_task") @patch("pdp.fastapi.routers.identity.bust_identity_caches") @patch("pdp.fastapi.routers.identity.identity") @patch("pdp.fastapi.routers.identity.BooleanFeature") async def test_attach_detach_roles_by_identity_tenant( mock_boolean_feature: AsyncMock, mock_identity: MagicMock, mock_bust_identity_caches: AsyncMock, mock_background_tasks: MagicMock, impersonated_by_identity_uuid: uuid.UUID | None, user_type: str, expected_is_on_for_identity_uuid: str, expected_principal: Principal, identity_uuid: str, tenant_1_uuid_as_string: str, tenant_1_uuid: uuid.UUID, mock_identity_ddb_connector: MagicMock, app: FastAPI, test_client: TestClient, mock_redis_connector: RedisConnector, mock_splitio_client: MagicMock, ) -> None: """Test PUT /identity//tenant//attach-and-detach/roles/ handler.""" # noqa: E501 app.dependency_overrides[impersonated_by_identity_uuid_from_scope] = ( lambda: impersonated_by_identity_uuid ) app.dependency_overrides[user_type_from_scope] = lambda: user_type app.dependency_overrides[get_principal_pdp_tenant_roles_from_scope] = lambda: {} mock_boolean_feature_instance = mock_boolean_feature.return_value mock_boolean_feature_instance.is_on_for_identity.return_value = True body = { "tenant_uuid": tenant_1_uuid_as_string, "tenant_type": "account", "roles_to_attach": [{"role": "settings_admin"}], "roles_to_detach": [], } operating_on_identity_uuid = uuid.uuid4() mock_identity.attach_and_detach_roles = AsyncMock( return_value={ "identity_uuid": identity_uuid, "tenant_uuid": tenant_1_uuid_as_string, "tenant_type": "account", "roles": [], "version": "0", } ) ows_response = test_client.put( f"/identity/{operating_on_identity_uuid}/tenant/{tenant_1_uuid_as_string}/attach-and-detach/roles/", # noqa: E501 json=body, ) assert ows_response.status_code == 200 data = ows_response.json() assert data == { "cursor": {"cursor": None, "shorthand": None}, "errors": {}, "tenants": { tenant_1_uuid_as_string: { "tenant_uuid": tenant_1_uuid_as_string, "tenant_type": "account", "roles": [], }, }, } mock_boolean_feature.assert_called_once_with( client=mock_splitio_client, feature_name="pp_send_impersonated_by_identity_uuid", ) mock_boolean_feature_instance.is_on_for_identity.assert_called_once_with( expected_is_on_for_identity_uuid, ) mock_identity.attach_and_detach_roles.assert_called_once_with( identity_uuid=str(operating_on_identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type=TenantType.TENANT_TYPE_ACCOUNT, roles_to_attach=[Role(**{"role": "settings_admin"})], roles_to_detach=[], authenticated_identity_uuid=identity_uuid, identity_ddb_connector=mock_identity_ddb_connector, principal=expected_principal, ) mock_background_tasks.assert_called_once_with( mock_bust_identity_caches, [operating_on_identity_uuid], redis_connector=mock_redis_connector, ) mock_bust_identity_caches.assert_not_called() @patch("fastapi.BackgroundTasks.add_task") @patch("pdp.fastapi.routers.identity.bust_identity_caches") @patch("pdp.fastapi.routers.identity.identity") @patch("pdp.fastapi.routers.identity.BooleanFeature") async def test_attach_detach_roles_by_identity_tenant_impersonation_ff_disabled( mock_boolean_feature: AsyncMock, mock_identity: MagicMock, mock_bust_identity_caches: AsyncMock, mock_background_tasks: MagicMock, identity_uuid: str, tenant_1_uuid_as_string: str, tenant_1_uuid: uuid.UUID, mock_identity_ddb_connector: MagicMock, test_client: TestClient, mock_redis_connector: RedisConnector, mock_splitio_client: MagicMock, ) -> None: """Test PUT /identity//tenant//attach-and-detach/roles/ handler.""" # noqa: E501 mock_boolean_feature_instance = mock_boolean_feature.return_value mock_boolean_feature_instance.is_on_for_identity.return_value = False body = { "tenant_uuid": tenant_1_uuid_as_string, "tenant_type": "account", "roles_to_attach": [{"role": "settings_admin"}], "roles_to_detach": [], } operating_on_identity_uuid = uuid.uuid4() mock_identity.attach_and_detach_roles = AsyncMock( return_value={ "identity_uuid": identity_uuid, "tenant_uuid": tenant_1_uuid_as_string, "tenant_type": "account", "roles": [], "version": "0", } ) ows_response = test_client.put( f"/identity/{operating_on_identity_uuid}/tenant/{tenant_1_uuid_as_string}/attach-and-detach/roles/", # noqa: E501 json=body, ) assert ows_response.status_code == 200 data = ows_response.json() assert data == { "cursor": {"cursor": None, "shorthand": None}, "errors": {}, "tenants": { tenant_1_uuid_as_string: { "tenant_uuid": tenant_1_uuid_as_string, "tenant_type": "account", "roles": [], }, }, } mock_boolean_feature.assert_called_once_with( client=mock_splitio_client, feature_name="pp_send_impersonated_by_identity_uuid", ) mock_boolean_feature_instance.is_on_for_identity.assert_called_once_with( identity_uuid, ) mock_identity.attach_and_detach_roles.assert_called_once_with( identity_uuid=str(operating_on_identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type=TenantType.TENANT_TYPE_ACCOUNT, roles_to_attach=[Role(**{"role": "settings_admin"})], roles_to_detach=[], authenticated_identity_uuid=identity_uuid, identity_ddb_connector=mock_identity_ddb_connector, ) mock_background_tasks.assert_called_once_with( mock_bust_identity_caches, [operating_on_identity_uuid], redis_connector=mock_redis_connector, ) mock_bust_identity_caches.assert_not_called() @patch("pdp.fastapi.routers.identity.identity") def test_attach_detach_roles_by_identity_tenant_mismatch_tenant_uuid( mock_identity: MagicMock, tenant_1_uuid_as_string: str, tenant_2_uuid_as_string: str, test_client: TestClient, ) -> None: """Test PUT /identity//tenant//attach-and-detach/roles/ handler.""" # noqa: E501 body = { "tenant_uuid": tenant_2_uuid_as_string, "tenant_type": "account", "roles_to_attach": [{"role": "settings_admin"}], "roles_to_detach": [], } operating_on_identity_uuid = uuid.uuid4() ows_response = test_client.put( f"/identity/{operating_on_identity_uuid}/tenant/{tenant_1_uuid_as_string}/attach-and-detach/roles/", # noqa: E501 json=body, ) assert ows_response.status_code == 400 assert ows_response.json() == { "code": "bad_request", "message": "Inconsistent request", } mock_identity.attach_and_detach_roles.assert_not_called() @patch.object(CheckResourcesRequest, "update_with_id_to_uuid_exchange") @patch("pdp.fastapi.auth.cerbos.check_resources") @patch("pdp.fastapi.routers.identity.get_identity_pdp_tenant_roles_from_scope") def test_check_identity_resources( mock_get_pdp_tenant_roles: AsyncMock, mock_cerbos_check_resources: AsyncMock, mock_update_with_id_to_uuid_exchange: AsyncMock, test_client: TestClient, mock_pdp_tenant_roles: Dict[uuid.UUID, TenantRoles], mock_async_cerbos_client: AsyncCerbosClient, identity_uuid_as_uuid: uuid.UUID, # The authenticated identity uuid from conftest mock_ows_account_client: OwsAccountClient, mock_ows_participant_client: OwsParticipantClient, mock_redis_connector: RedisConnector, mock_splitio_client: MagicMock, ) -> None: """Test /identity//check/resources/ handler.""" mock_request_body = { "resources": [ { "resource": { "resource_id": "123", "resource_type": "fan_data_list", "attributes": {"tenant_type": "account", "tenant_uuid": "fail now"}, }, "action": "view", } ] } mock_get_pdp_tenant_roles.return_value = mock_pdp_tenant_roles mock_cerbos_check_resources_response = { "request_id": "hello2", "resources": [ { "resource": { "resource_id": "123", "resource_type": "Identity", "attributes": {"attribute": "1"}, }, "action": "view", "effect": "deny", "errors": {"validation_errors": []}, } ], } mock_cerbos_check_resources.return_value = mock_cerbos_check_resources_response identity_uuid_to_check = uuid.uuid4() response = test_client.post( f"/identity/{identity_uuid_to_check}/check/resources/", json=mock_request_body, ) mock_cerbos_check_resources.assert_awaited_once_with( identity_uuid=str(identity_uuid_to_check), check_resources_request=CheckResourcesRequest(**mock_request_body), pdp_tenant_roles=mock_pdp_tenant_roles, cerbos_client=mock_async_cerbos_client, include_resource_attributes=False, authenticated_identity_uuid=identity_uuid_as_uuid, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, redis_connector=mock_redis_connector, splitio_client=mock_splitio_client, ) mock_update_with_id_to_uuid_exchange.assert_awaited_once_with( redis_connector=mock_redis_connector, ows_account_client=mock_ows_account_client, ) assert response.json() == mock_cerbos_check_resources_response @pytest.mark.parametrize( "impersonated_by_identity_uuid,user_type,expected_is_on_for_identity_uuid,expected_principal", [ pytest.param( None, "human", "ab123456-1234-4c2b-9c23-123ab4000a1b", Principal( identity_uuid="ab123456-1234-4c2b-9c23-123ab4000a1b", pdp_tenant_roles={}, user_type="human", ), id="Principal is not impersonated", ), pytest.param( None, "machine", "ab123456-1234-4c2b-9c23-123ab4000a1b", Principal( identity_uuid="ab123456-1234-4c2b-9c23-123ab4000a1b", pdp_tenant_roles={}, user_type="machine", ), id="Principal is a machine", ), pytest.param( uuid.UUID("7e1ba04d-b15f-4811-96ed-4320a599f595"), "machine", "7e1ba04d-b15f-4811-96ed-4320a599f595", Principal( identity_uuid="ab123456-1234-4c2b-9c23-123ab4000a1b", pdp_tenant_roles={}, user_type="machine", impersonated_by_identity_uuid="7e1ba04d-b15f-4811-96ed-4320a599f595", ), id="Principal is a machine impersonating", ), ], ) @patch("fastapi.BackgroundTasks.add_task") @patch("pdp.fastapi.routers.identity.bust_identity_caches") @patch("pdp.fastapi.routers.identity.identity.deactivate_many") @patch("pdp.fastapi.routers.identity.filter_for_identity_tenants") @patch("pdp.fastapi.routers.identity.get_identity_pdp_tenant_roles_from_scope") @patch("pdp.fastapi.routers.identity.BooleanFeature") async def test_deactivate_all( mock_boolean_feature: AsyncMock, mock_get_identity_pdp_tenant_roles: AsyncMock, mock_filter_for_identity_tenants: AsyncMock, mock_deactivate_many: AsyncMock, mock_bust_identity_caches: AsyncMock, mock_background_tasks: MagicMock, impersonated_by_identity_uuid: uuid.UUID | None, user_type: str, expected_is_on_for_identity_uuid: str, expected_principal: Principal, mock_identity_ddb_connector: MagicMock, identity_uuid: str, identity_uuid_as_uuid: uuid.UUID, mock_pdp_tenant_roles: Dict[uuid.UUID, TenantRoles], mock_my_adminable_tenant_roles: Dict[uuid.UUID, TenantRoles], app: FastAPI, test_client: TestClient, mock_async_cerbos_client: AsyncCerbosClient, mock_ows_account_client: OwsAccountClient, mock_ows_participant_client: OwsParticipantClient, mock_redis_connector: RedisConnector, mock_splitio_client: MagicMock, tenant_1_uuid: uuid.UUID, tenant_2_uuid: uuid.UUID, ) -> None: """Test the deactivate all endpoint.""" # noqa: E501 app.dependency_overrides[impersonated_by_identity_uuid_from_scope] = ( lambda: impersonated_by_identity_uuid ) app.dependency_overrides[user_type_from_scope] = lambda: user_type mock_boolean_feature_instance = mock_boolean_feature.return_value mock_boolean_feature_instance.is_on_for_identity.return_value = True requested_identity_uuid = uuid.uuid4() # Less copy-paste for the parametrize'd expected_principal expected_principal.pdp_tenant_roles = mock_pdp_tenant_roles mock_get_identity_pdp_tenant_roles.return_value = mock_pdp_tenant_roles allowed_tenants = { tenant_1_uuid: TenantRoles( tenant_type="account", tenant_uuid=tenant_1_uuid, roles=[], ), tenant_2_uuid: TenantRoles( tenant_type="subaccount", tenant_uuid=tenant_2_uuid, roles=[], ), } mock_filter_for_identity_tenants.return_value = allowed_tenants mock_deactivate_many.return_value = DeactivationSummary(deleted=2, remaining=0) response = test_client.delete(f"/identity/{requested_identity_uuid}/deactivate/") mock_filter_for_identity_tenants.assert_awaited_once_with( requested_identity_uuid, mock_pdp_tenant_roles, action="deactivate", principal_identity_uuid=identity_uuid_as_uuid, pdp_tenant_roles=mock_pdp_tenant_roles, ows_permissions_tenant_roles=mock_my_adminable_tenant_roles, cerbos_client=mock_async_cerbos_client, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, redis_connector=mock_redis_connector, impersonated_by_identity_uuid=impersonated_by_identity_uuid, splitio_client=mock_splitio_client, user_type=user_type, ) mock_boolean_feature.assert_called_once_with( client=mock_splitio_client, feature_name="pp_send_impersonated_by_identity_uuid", ) mock_boolean_feature_instance.is_on_for_identity.assert_called_once_with( expected_is_on_for_identity_uuid, ) mock_deactivate_many.assert_awaited_once_with( identity_uuid=str(requested_identity_uuid), tenants=allowed_tenants, identity_ddb_connector=mock_identity_ddb_connector, authenticated_identity_uuid=identity_uuid, principal=expected_principal, ) mock_background_tasks.assert_called_once_with( mock_bust_identity_caches, [requested_identity_uuid], redis_connector=mock_redis_connector, ) mock_bust_identity_caches.assert_not_called() assert DeactivateAllResponse.model_validate( response.json() ) == DeactivateAllResponse(summary=DeactivationSummary(deleted=2, remaining=0)) @patch("fastapi.BackgroundTasks.add_task") @patch("pdp.fastapi.routers.identity.bust_identity_caches") @patch("pdp.fastapi.routers.identity.identity.deactivate_many") @patch("pdp.fastapi.routers.identity.filter_for_identity_tenants") @patch("pdp.fastapi.routers.identity.get_identity_pdp_tenant_roles_from_scope") @patch("pdp.fastapi.routers.identity.BooleanFeature") async def test_deactivate_all_impersonation_ff_disabled( mock_boolean_feature: AsyncMock, mock_get_identity_pdp_tenant_roles: AsyncMock, mock_filter_for_identity_tenants: AsyncMock, mock_deactivate_many: AsyncMock, mock_bust_identity_caches: AsyncMock, mock_background_tasks: MagicMock, mock_identity_ddb_connector: MagicMock, identity_uuid: str, identity_uuid_as_uuid: uuid.UUID, mock_pdp_tenant_roles: Dict[uuid.UUID, TenantRoles], mock_my_adminable_tenant_roles: Dict[uuid.UUID, TenantRoles], test_client: TestClient, mock_async_cerbos_client: AsyncCerbosClient, mock_ows_account_client: OwsAccountClient, mock_ows_participant_client: OwsParticipantClient, mock_redis_connector: RedisConnector, mock_splitio_client: MagicMock, tenant_1_uuid: uuid.UUID, tenant_2_uuid: uuid.UUID, ) -> None: """Test the deactivate all endpoint, pp_send_impersonated_by_identity_uuid disabled.""" # noqa: E501 mock_boolean_feature_instance = mock_boolean_feature.return_value mock_boolean_feature_instance.is_on_for_identity.return_value = False requested_identity_uuid = uuid.uuid4() mock_get_identity_pdp_tenant_roles.return_value = mock_pdp_tenant_roles allowed_tenants = { tenant_1_uuid: TenantRoles( tenant_type="account", tenant_uuid=tenant_1_uuid, roles=[], ), tenant_2_uuid: TenantRoles( tenant_type="subaccount", tenant_uuid=tenant_2_uuid, roles=[], ), } mock_filter_for_identity_tenants.return_value = allowed_tenants mock_deactivate_many.return_value = DeactivationSummary(deleted=2, remaining=0) response = test_client.delete(f"/identity/{requested_identity_uuid}/deactivate/") mock_filter_for_identity_tenants.assert_awaited_once_with( requested_identity_uuid, mock_pdp_tenant_roles, action="deactivate", principal_identity_uuid=identity_uuid_as_uuid, pdp_tenant_roles=mock_pdp_tenant_roles, ows_permissions_tenant_roles=mock_my_adminable_tenant_roles, cerbos_client=mock_async_cerbos_client, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, redis_connector=mock_redis_connector, impersonated_by_identity_uuid=None, splitio_client=mock_splitio_client, user_type=USER_TYPE_HUMAN, ) mock_boolean_feature.assert_called_once_with( client=mock_splitio_client, feature_name="pp_send_impersonated_by_identity_uuid", ) mock_boolean_feature_instance.is_on_for_identity.assert_called_once_with( identity_uuid, ) mock_deactivate_many.assert_awaited_once_with( identity_uuid=str(requested_identity_uuid), tenants=allowed_tenants, identity_ddb_connector=mock_identity_ddb_connector, authenticated_identity_uuid=identity_uuid, ) mock_background_tasks.assert_called_once_with( mock_bust_identity_caches, [requested_identity_uuid], redis_connector=mock_redis_connector, ) mock_bust_identity_caches.assert_not_called() assert DeactivateAllResponse.model_validate( response.json() ) == DeactivateAllResponse(summary=DeactivationSummary(deleted=2, remaining=0)) @patch("pdp.fastapi.routers.identity.identity.deactivate_many") @patch("pdp.fastapi.routers.identity.filter_for_identity_tenants") @patch("pdp.fastapi.routers.identity.get_identity_pdp_tenant_roles_from_scope") async def test_deactivate_all_no_attached_tenants( mock_get_identity_pdp_tenant_roles: AsyncMock, mock_filter_for_identity_tenants: AsyncMock, mock_deactivate_many: AsyncMock, test_client: TestClient, ) -> None: """Test the deactivate all endpoint 200s when identity has no attached tenants.""" requested_identity_uuid = uuid.uuid4() mock_get_identity_pdp_tenant_roles.return_value = {} response = test_client.delete(f"/identity/{requested_identity_uuid}/deactivate/") mock_filter_for_identity_tenants.assert_not_called() mock_deactivate_many.assert_not_called() assert response.status_code == 200 assert DeactivateAllResponse.model_validate( response.json() ) == DeactivateAllResponse(summary=DeactivationSummary(deleted=0, remaining=0)) @patch("pdp.fastapi.routers.identity.identity.deactivate_many") @patch("pdp.fastapi.routers.identity.filter_for_identity_tenants") @patch("pdp.fastapi.routers.identity.get_identity_pdp_tenant_roles_from_scope") async def test_deactivate_all_no_allowed_tenants( mock_get_identity_pdp_tenant_roles: AsyncMock, mock_filter_for_identity_tenants: AsyncMock, mock_deactivate_many: AsyncMock, identity_uuid_as_uuid: uuid.UUID, mock_pdp_tenant_roles: Dict[uuid.UUID, TenantRoles], mock_my_adminable_tenant_roles: Dict[uuid.UUID, TenantRoles], test_client: TestClient, mock_async_cerbos_client: AsyncCerbosClient, mock_ows_account_client: OwsAccountClient, mock_ows_participant_client: OwsParticipantClient, mock_redis_connector: RedisConnector, mock_splitio_client: SplitioClient, ) -> None: """Test the deactivate all endpoint 403s when there are no allowed tenants.""" requested_identity_uuid = uuid.uuid4() mock_get_identity_pdp_tenant_roles.return_value = mock_pdp_tenant_roles mock_filter_for_identity_tenants.return_value = {} response = test_client.delete(f"/identity/{requested_identity_uuid}/deactivate/") mock_filter_for_identity_tenants.assert_awaited_once_with( requested_identity_uuid, mock_pdp_tenant_roles, action="deactivate", principal_identity_uuid=identity_uuid_as_uuid, pdp_tenant_roles=mock_pdp_tenant_roles, ows_permissions_tenant_roles=mock_my_adminable_tenant_roles, cerbos_client=mock_async_cerbos_client, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, redis_connector=mock_redis_connector, impersonated_by_identity_uuid=None, splitio_client=mock_splitio_client, user_type=USER_TYPE_HUMAN, ) mock_deactivate_many.assert_not_called() assert response.status_code == 403 assert response.json() == { "code": "bad_request", "message": f"Principal {identity_uuid_as_uuid} not authorized to deactivate all tenants from {requested_identity_uuid}.", # noqa: E501 } @pytest.mark.parametrize( "impersonated_by_identity_uuid,user_type,expected_is_on_for_identity_uuid,expected_principal", [ pytest.param( None, "human", "ab123456-1234-4c2b-9c23-123ab4000a1b", Principal( identity_uuid="ab123456-1234-4c2b-9c23-123ab4000a1b", pdp_tenant_roles={}, user_type="human", ), id="Principal is not impersonated", ), pytest.param( None, "machine", "ab123456-1234-4c2b-9c23-123ab4000a1b", Principal( identity_uuid="ab123456-1234-4c2b-9c23-123ab4000a1b", pdp_tenant_roles={}, user_type="machine", ), id="Principal is a machine", ), pytest.param( uuid.UUID("7e1ba04d-b15f-4811-96ed-4320a599f595"), "machine", "7e1ba04d-b15f-4811-96ed-4320a599f595", Principal( identity_uuid="ab123456-1234-4c2b-9c23-123ab4000a1b", pdp_tenant_roles={}, user_type="machine", impersonated_by_identity_uuid="7e1ba04d-b15f-4811-96ed-4320a599f595", ), id="Principal is a machine impersonating", ), ], ) @patch("fastapi.BackgroundTasks.add_task") @patch("pdp.fastapi.routers.identity.bust_identity_caches") @patch("pdp.fastapi.routers.identity.identity.deactivate_many") @patch("pdp.fastapi.routers.identity.identity.is_tenant_assigned") @patch("pdp.fastapi.routers.identity.BooleanFeature") def test_deactivate_one( mock_boolean_feature: AsyncMock, mock_is_tenant_assigned: MagicMock, mock_deactivate_many: AsyncMock, mock_bust_identity_caches: AsyncMock, mock_background_tasks: MagicMock, impersonated_by_identity_uuid: uuid.UUID | None, user_type: str, expected_is_on_for_identity_uuid: str, expected_principal: Principal, tenant_1_uuid: uuid.UUID, tenant_1_uuid_as_string: str, app: FastAPI, test_client: TestClient, mock_identity_ddb_connector: MagicMock, mock_redis_connector: MagicMock, mock_splitio_client: MagicMock, identity_uuid: str, ) -> None: """Test deactivate_one endpoint.""" app.dependency_overrides[impersonated_by_identity_uuid_from_scope] = ( lambda: impersonated_by_identity_uuid ) app.dependency_overrides[user_type_from_scope] = lambda: user_type app.dependency_overrides[get_principal_pdp_tenant_roles_from_scope] = lambda: {} mock_boolean_feature_instance = mock_boolean_feature.return_value mock_boolean_feature_instance.is_on_for_identity.return_value = True mock_is_tenant_assigned.return_value = True mock_deactivate_many.return_value = DeactivationSummary(deleted=0, remaining=1) expected = DeactivateOneResponse( summary=DeactivationSummary(deleted=0, remaining=1) ) requested_identity_uuid = uuid.uuid4() actual = test_client.post( f"/identity/{requested_identity_uuid}/tenant/{tenant_1_uuid_as_string}/deactivate/", json={"tenant_type": "account", "tenant_uuid": tenant_1_uuid_as_string}, ) mock_boolean_feature.assert_called_once_with( client=mock_splitio_client, feature_name="pp_send_impersonated_by_identity_uuid", ) mock_boolean_feature_instance.is_on_for_identity.assert_called_once_with( expected_is_on_for_identity_uuid, ) mock_is_tenant_assigned.assert_called_once_with( identity_uuid=str(requested_identity_uuid), tenant_uuid=tenant_1_uuid, identity_ddb_connector=mock_identity_ddb_connector, ) mock_deactivate_many.assert_awaited_once_with( identity_uuid=str(requested_identity_uuid), tenants={ tenant_1_uuid: TenantRoles.model_validate( { "tenant_type": "account", "tenant_uuid": tenant_1_uuid_as_string, "roles": [], } ) }, identity_ddb_connector=mock_identity_ddb_connector, authenticated_identity_uuid=identity_uuid, principal=expected_principal, ) mock_background_tasks.assert_called_once_with( mock_bust_identity_caches, [requested_identity_uuid], redis_connector=mock_redis_connector, ) mock_bust_identity_caches.assert_not_called() assert DeactivateOneResponse.model_validate(actual.json()) == expected @patch("fastapi.BackgroundTasks.add_task") @patch("pdp.fastapi.routers.identity.bust_identity_caches") @patch("pdp.fastapi.routers.identity.identity.deactivate_many") @patch("pdp.fastapi.routers.identity.identity.is_tenant_assigned") @patch("pdp.fastapi.routers.identity.BooleanFeature") def test_deactivate_one_impersonation_ff_disabled( mock_boolean_feature: AsyncMock, mock_is_tenant_assigned: MagicMock, mock_deactivate_many: AsyncMock, mock_bust_identity_caches: AsyncMock, mock_background_tasks: MagicMock, tenant_1_uuid: uuid.UUID, tenant_1_uuid_as_string: str, test_client: TestClient, mock_identity_ddb_connector: MagicMock, mock_redis_connector: MagicMock, mock_splitio_client: MagicMock, identity_uuid: str, ) -> None: """Test deactivate_one endpoint, pp_send_impersonated_by_identity_uuid disabled.""" mock_boolean_feature_instance = mock_boolean_feature.return_value mock_boolean_feature_instance.is_on_for_identity.return_value = False mock_is_tenant_assigned.return_value = True mock_deactivate_many.return_value = DeactivationSummary(deleted=0, remaining=1) expected = DeactivateOneResponse( summary=DeactivationSummary(deleted=0, remaining=1) ) requested_identity_uuid = uuid.uuid4() actual = test_client.post( f"/identity/{requested_identity_uuid}/tenant/{tenant_1_uuid_as_string}/deactivate/", json={"tenant_type": "account", "tenant_uuid": tenant_1_uuid_as_string}, ) mock_boolean_feature.assert_called_once_with( client=mock_splitio_client, feature_name="pp_send_impersonated_by_identity_uuid", ) mock_boolean_feature_instance.is_on_for_identity.assert_called_once_with( identity_uuid, ) mock_is_tenant_assigned.assert_called_once_with( identity_uuid=str(requested_identity_uuid), tenant_uuid=tenant_1_uuid, identity_ddb_connector=mock_identity_ddb_connector, ) mock_deactivate_many.assert_awaited_once_with( identity_uuid=str(requested_identity_uuid), tenants={ tenant_1_uuid: TenantRoles.model_validate( { "tenant_type": "account", "tenant_uuid": tenant_1_uuid_as_string, "roles": [], } ) }, identity_ddb_connector=mock_identity_ddb_connector, authenticated_identity_uuid=identity_uuid, ) mock_background_tasks.assert_called_once_with( mock_bust_identity_caches, [requested_identity_uuid], redis_connector=mock_redis_connector, ) mock_bust_identity_caches.assert_not_called() assert DeactivateOneResponse.model_validate(actual.json()) == expected @pytest.mark.parametrize( "identity_uuid, tenant_uuid", [ pytest.param( str(uuid.uuid1()), str(uuid.uuid4()), id="identity_uuid cannot be uuid1", ), pytest.param( "some-string-we-pretend-is-unique", str(uuid.uuid4()), id="identity_uuid cannot be some string", ), pytest.param( str(uuid.uuid4()), "no-this-is-not-a-uuid", id="tenant_uuid cannot be some string", ), ], ) @patch("pdp.fastapi.routers.identity.identity.deactivate_many") def test_deactivate_one_validates_uuids( mock_deactivate_many: AsyncMock, identity_uuid: str, tenant_uuid: str, test_client: TestClient, ) -> None: """Test deactivate_one endpoint validates the path parameters.""" actual = test_client.post( f"/identity/{identity_uuid}/tenant/{tenant_uuid}/deactivate/", json={"tenant_type": "account", "tenant_uuid": tenant_uuid}, ) mock_deactivate_many.assert_not_called() assert actual.status_code == 422 def test_deactivate_one_validates_tenant_uuid_matches(test_client: TestClient) -> None: """Test deactivate one checks the tenant_uuids provided match.""" actual = test_client.post( f"/identity/{uuid.uuid4()}/tenant/{uuid.uuid1()}/deactivate/", json={"tenant_type": "account", "tenant_uuid": str(uuid.uuid4())}, ) assert actual.status_code == 400 @patch("pdp.fastapi.routers.identity.identity.deactivate_many") @patch("pdp.fastapi.routers.identity.identity.is_tenant_assigned") def test_deactivate_one_validates_is_tenant_assigned_for_identity( mock_is_tenant_assigned: MagicMock, mock_deactivate_many: AsyncMock, mock_identity_ddb_connector: MagicMock, test_client: TestClient, ) -> None: """Test deactivate one checks the tenant exists for identity.""" mock_is_tenant_assigned.return_value = False identity_uuid = uuid.uuid4() tenant_uuid = uuid.uuid1() actual = test_client.post( f"/identity/{identity_uuid}/tenant/{tenant_uuid}/deactivate/", json={"tenant_type": "account", "tenant_uuid": str(tenant_uuid)}, ) mock_is_tenant_assigned.assert_called_once_with( identity_uuid=str(identity_uuid), tenant_uuid=tenant_uuid, identity_ddb_connector=mock_identity_ddb_connector, ) mock_deactivate_many.assert_not_awaited() assert actual.status_code == 403