"""Test auth utility methods.""" import logging import uuid from typing import Any, Dict, List from unittest.mock import AsyncMock, MagicMock, call, patch import pytest from cerbos.sdk.client import AsyncCerbosClient from fastapi import Depends, FastAPI, HTTPException from fastapi.middleware import Middleware from fastapi.requests import Request from fastapi.responses import PlainTextResponse from fastapi.routing import APIRoute from fastapi.testclient import TestClient from httpx import HTTPStatusError from jwtauth import JWTAuth from jwtauth.asgi.middleware import JWTAuthenticationMiddleware from splitio.client.factory import Client as SplitioClient from pdp import config from pdp.connectors.features import FEATURE_OFF, FEATURE_ON from pdp.connectors.ows_permissions import OwsPermissionsResource from pdp.constants import constants from pdp.constants.constants import USER_TYPE_HUMAN from pdp.fastapi.auth import ( _build_identity_resources, check_authorization, check_authorization_infra, filter_for_identity_tenants, get_identity_pdp_tenant_roles_from_scope, get_principal_ows_permissions_from_scope, get_principal_pdp_tenant_roles_from_scope, identity_uuid_from_scope, impersonated_by_identity_uuid_from_scope, user_type_from_scope, ) from pdp.fastapi.schemas import check_resources, identity from pdp.fastapi.schemas.cache import PrincipalPdpCacheObject from pdp.fastapi.schemas.principal import Principal from pdp.logic.jwt import CLAIM_ORCHARD_IDENTITY_ID identity_uuid_one = "0a633498-e9cd-402c-9791-36fad00a6afd" identity_uuid_two = "5da22999-bc1a-4f10-a46b-2bde3c1f2a72" @pytest.fixture def app() -> FastAPI: """Fastapi app fixture.""" mock_auth = MagicMock(spec=JWTAuth) mock_auth.aget_token.return_value = { config.JWT_USER_METADATA: {CLAIM_ORCHARD_IDENTITY_ID: identity_uuid_one}, config.JWT_IMPERSONATED_BY: identity_uuid_two, } app = FastAPI( middleware=[ Middleware( JWTAuthenticationMiddleware, enabled=True, auth=mock_auth, ), ] ) @app.get("/test/") async def handler( identity_uuid: uuid.UUID = Depends(identity_uuid_from_scope), ) -> PlainTextResponse: """Fake handler that returns identity uuid.""" return PlainTextResponse(str(identity_uuid)) @app.get("/test/user-type/") async def user_type_handler( user_type: str = Depends(user_type_from_scope), ) -> PlainTextResponse: """Fake handler that returns user type.""" return PlainTextResponse(user_type) @app.get("/test/impersonated-by/") async def impersonated_by_handler( impersonated_by_identity_uuid: uuid.UUID = Depends( impersonated_by_identity_uuid_from_scope ), ) -> PlainTextResponse: """Fake handler that returns impersonated by identity uuid.""" return PlainTextResponse(str(impersonated_by_identity_uuid)) return app def test_identity_uuid_from_scope(app: FastAPI) -> None: """Verify endpoint get identity_uuid from jwt.""" with TestClient(app) as test_client: response = test_client.get("/test/", headers={"authorization": "bearer token"}) assert response.text == identity_uuid_one @pytest.mark.parametrize( "bad_identity_uuid, error_message, description", [ ( None, "No identity found", "Missing identity uuid from token claims should be 401", ), ( "not-a-uuid-4", "Identity UUID is not valid", "Invalid identity uuid from token claims should be 401", ), ], ) @patch("pdp.fastapi.auth.jwt") def test_identity_uuid_from_scope_missing( mock_jwt: MagicMock, bad_identity_uuid: Any, error_message: str, description: str, app: FastAPI, ) -> None: """Verify 401 when identity_uuid is missing from jwt.""" mock_jwt.get_identity_uuid.return_value = bad_identity_uuid with TestClient(app) as test_client: response = test_client.get("/test/", headers={"authorization": "bearer token"}) assert response.status_code == 401, description assert response.json() == {"detail": error_message}, description def test_identity_uuid_from_scope_no_token() -> None: """Verify 401 when no token in request scope.""" request = MagicMock(spec=Request) request.scope = {} with pytest.raises(HTTPException) as exc_info: identity_uuid_from_scope(request) assert exc_info.value.status_code == 401 assert exc_info.value.detail == "JWT not decoded by JWTAuthenticationMiddleware" def test_impersonated_by_identity_uuid_from_scope(app: FastAPI) -> None: """Verify endpoint get identity_uuid from jwt.""" with TestClient(app) as test_client: response = test_client.get( "/test/impersonated-by/", headers={"authorization": "bearer token"} ) assert response.text == identity_uuid_two @patch("pdp.fastapi.auth.jwt") def test_impersonated_by_identity_uuid_from_scope_missing( mock_jwt: MagicMock, app: FastAPI ) -> None: """Verify 200 when impersonated by identity uuid is missing from jwt.""" mock_jwt.get_impersonated_by_identity_uuid.return_value = None with TestClient(app) as test_client: response = test_client.get( "/test/impersonated-by/", headers={"authorization": "bearer token"} ) assert response.status_code == 200 @patch("pdp.fastapi.auth.jwt") def test_impersonated_by_identity_uuid_from_scope_bad_uuid( mock_jwt: MagicMock, app: FastAPI ) -> None: """Verify 401 when impersonated by identity uuid is bad uuid.""" mock_jwt.get_impersonated_by_identity_uuid.return_value = "not-a-uuid-4" with TestClient(app) as test_client: response = test_client.get( "/test/impersonated-by/", headers={"authorization": "bearer token"} ) assert response.status_code == 401 assert response.json() == { "detail": "Impersonated by Identity UUID is not valid" } def test_impersonated_by_identity_uuid_from_scope_no_token() -> None: """Verify 401 when no token in request scope.""" request = MagicMock(spec=Request) request.scope = {} with pytest.raises(HTTPException) as exc_info: impersonated_by_identity_uuid_from_scope(request) assert exc_info.value.status_code == 401 assert exc_info.value.detail == "JWT not decoded by JWTAuthenticationMiddleware" def test_user_type_from_scope(app: FastAPI) -> None: """Verify get user_type_from_scope from jwt.""" with patch("pdp.fastapi.auth.jwt") as mock_jwt: mock_jwt.get_user_type.return_value = "human" with TestClient(app) as test_client: response = test_client.get( "/test/user-type/", headers={"authorization": "bearer token"} ) assert response.text == "human" mock_jwt.get_user_type.assert_called_once_with( { config.JWT_USER_METADATA: { CLAIM_ORCHARD_IDENTITY_ID: identity_uuid_one }, config.JWT_IMPERSONATED_BY: identity_uuid_two, } ) def test_user_type_from_scope_no_token() -> None: """Verify get user_type_from_scope errors when no token in request scope.""" request = MagicMock(spec=Request) request.scope = {} with pytest.raises(HTTPException) as exc_info: user_type_from_scope(request) assert exc_info.value.status_code == 401 assert exc_info.value.detail == "JWT not decoded by JWTAuthenticationMiddleware" def test_user_type_from_scope_machine(app: FastAPI) -> None: """Verify get user_type_from_scope from jwt.""" with patch("pdp.fastapi.auth.jwt") as mock_jwt: mock_jwt.get_user_type.return_value = "machine" with TestClient(app) as test_client: response = test_client.get( "/test/user-type/", headers={"authorization": "bearer token"} ) assert response.text == "machine" mock_jwt.get_user_type.assert_called_once_with( { config.JWT_USER_METADATA: { CLAIM_ORCHARD_IDENTITY_ID: identity_uuid_one }, config.JWT_IMPERSONATED_BY: identity_uuid_two, } ) @pytest.mark.parametrize( "api_route, expect_action, cerbos_effect, expect_exception", [ ( "attach_detach_roles_by_identity_tenant", "attach_and_detach_role", "allow", False, ), ( "attach_detach_roles_by_identity_tenant", "attach_and_detach_role", "deny", True, ), ("get_roles_by_identity", "list_tenants", "allow", False), ("get_roles_by_identity", "list_tenants", "deny", True), ("check_identity_resources", "check_resources", "allow", False), ("check_identity_resources", "check_resources", "deny", True), ( "deactivate_one", "deactivate", "allow", False, ), ( "deactivate_one", "deactivate", "deny", True, ), ( "deactivate_all", "deactivate_all", "allow", False, ), ( "deactivate_all", "deactivate_all", "deny", True, ), ], ) @patch("pdp.logic.cerbos.RedisConnector") @patch("pdp.logic.cerbos.OwsParticipantClient") @patch("pdp.logic.cerbos.OwsAccountClient") @patch("pdp.fastapi.auth.cerbos.check_resources") @patch("pdp.fastapi.auth._build_identity_resources") async def test_check_authorization( mock_build_identity_resources: AsyncMock, mock_cerbos_check_resources: AsyncMock, mock_ows_account_client: MagicMock, mock_ows_participant_client: MagicMock, mock_redis_connector: MagicMock, api_route: str, expect_action: str, cerbos_effect: str, expect_exception: bool, identity_uuid: str, mock_pdp_tenant_roles: Dict[uuid.UUID, identity.TenantRoles], mock_my_adminable_tenant_roles: Dict[uuid.UUID, identity.TenantRoles], mock_async_cerbos_client: AsyncCerbosClient, mock_splitio_client: SplitioClient, ) -> None: """Test check_authorization for role administration operations.""" mock_splitio_client.get_treatment.return_value = FEATURE_ON mock_request = MagicMock() mock_request.get.return_value = APIRoute( path="some/path", name=api_route, endpoint=MagicMock() ) mock_build_identity_resources.return_value = [ identity.Resource( resource_id="some uuid", resource_type="identity", ), identity.Resource( resource_id="another uuid", resource_type="identity", ), ] mock_cerbos_check_resources.return_value = identity.CheckResourcesResponse( request_id="check_auth", resources=[ identity.CheckResourceActionResult( resource=mock_build_identity_resources.return_value[0], effect=cerbos_effect, action=expect_action, ), identity.CheckResourceActionResult( resource=mock_build_identity_resources.return_value[1], effect=cerbos_effect, action=expect_action, ), ], ) some_uuid = uuid.uuid4() impersonated_by_identity_uuid = uuid.uuid4() user_type = USER_TYPE_HUMAN expected_principal = Principal( identity_uuid=uuid.UUID(identity_uuid), user_type=user_type, pdp_tenant_roles=mock_pdp_tenant_roles, ows_permissions_tenant_roles=mock_my_adminable_tenant_roles, impersonated_by_identity_uuid=impersonated_by_identity_uuid, ) if expect_exception: with pytest.raises(Exception): await check_authorization( mock_request, some_uuid, uuid.UUID(identity_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, splitio_client=mock_splitio_client, user_type=user_type, impersonated_by_identity_uuid=impersonated_by_identity_uuid, ) else: await check_authorization( mock_request, some_uuid, uuid.UUID(identity_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, splitio_client=mock_splitio_client, user_type=user_type, impersonated_by_identity_uuid=impersonated_by_identity_uuid, ) mock_cerbos_check_resources.assert_awaited_once_with( identity_uuid=identity_uuid, check_resources_request=check_resources.CheckResourcesRequest( resources=[ identity.CheckResourceAction( resource=identity.Resource( resource_id="some uuid", resource_type="identity", ), action=expect_action, ), identity.CheckResourceAction( resource=identity.Resource( resource_id="another uuid", resource_type="identity", ), action=expect_action, ), ] ), 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, authenticated_identity_uuid=uuid.UUID(identity_uuid), principal=expected_principal, splitio_client=mock_splitio_client, ) mock_splitio_client.get_treatment.assert_called_once_with( attributes={"identity_id": str(impersonated_by_identity_uuid)}, feature_flag_name="pp_send_impersonated_by_identity_uuid", key="identity_id", ) @patch("pdp.logic.cerbos.RedisConnector") @patch("pdp.logic.cerbos.OwsParticipantClient") @patch("pdp.logic.cerbos.OwsAccountClient") @patch("pdp.fastapi.auth.cerbos.check_resources") @patch("pdp.fastapi.auth._build_identity_resources") async def test_check_authorization__when_send_impersonated_by_identity_uuid_is_disabled( # noqa: E501 mock_build_identity_resources: AsyncMock, mock_cerbos_check_resources: AsyncMock, mock_ows_account_client: MagicMock, mock_ows_participant_client: MagicMock, mock_redis_connector: MagicMock, identity_uuid: str, mock_pdp_tenant_roles: Dict[uuid.UUID, identity.TenantRoles], mock_my_adminable_tenant_roles: Dict[uuid.UUID, identity.TenantRoles], mock_async_cerbos_client: AsyncCerbosClient, mock_splitio_client: SplitioClient, ) -> None: """Test check_authorization for role administration operations when the FF is disabled.""" # noqa: E501 mock_splitio_client.get_treatment.return_value = FEATURE_OFF api_route = "deactivate_one" expect_action = "deactivate" cerbos_effect = "allow" mock_request = MagicMock() mock_request.get.return_value = APIRoute( path="some/path", name=api_route, endpoint=MagicMock() ) mock_build_identity_resources.return_value = [ identity.Resource( resource_id="some uuid", resource_type="identity", ), identity.Resource( resource_id="another uuid", resource_type="identity", ), ] mock_cerbos_check_resources.return_value = identity.CheckResourcesResponse( request_id="check_auth", resources=[ identity.CheckResourceActionResult( resource=mock_build_identity_resources.return_value[0], effect=cerbos_effect, action=expect_action, ), identity.CheckResourceActionResult( resource=mock_build_identity_resources.return_value[1], effect=cerbos_effect, action=expect_action, ), ], ) some_uuid = uuid.uuid4() impersonated_by_identity_uuid = uuid.uuid4() user_type = USER_TYPE_HUMAN await check_authorization( mock_request, some_uuid, uuid.UUID(identity_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, splitio_client=mock_splitio_client, user_type=user_type, impersonated_by_identity_uuid=impersonated_by_identity_uuid, ) # does not include a `Principal` arg. mock_cerbos_check_resources.assert_awaited_once_with( identity_uuid=identity_uuid, check_resources_request=check_resources.CheckResourcesRequest( resources=[ identity.CheckResourceAction( resource=identity.Resource( resource_id="some uuid", resource_type="identity", ), action=expect_action, ), identity.CheckResourceAction( resource=identity.Resource( resource_id="another uuid", resource_type="identity", ), action=expect_action, ), ] ), 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, authenticated_identity_uuid=uuid.UUID(identity_uuid), splitio_client=mock_splitio_client, ) mock_splitio_client.get_treatment.assert_called_once_with( attributes={"identity_id": str(impersonated_by_identity_uuid)}, feature_flag_name="pp_send_impersonated_by_identity_uuid", key="identity_id", ) @pytest.mark.parametrize( "api_route, action, cerbos_effect, expect_exception", [ ("cache_list", "list_cache", "allow", False), ("cache_list", "list_cache", "deny", True), ], ) @patch("pdp.logic.cerbos.RedisConnector") @patch("pdp.logic.cerbos.OwsParticipantClient") @patch("pdp.logic.cerbos.OwsAccountClient") @patch("pdp.fastapi.auth.cerbos.check_resources") @patch("pdp.fastapi.auth._build_infra_resources") async def test_check_authorization_infra( mock_build_infra_resources: AsyncMock, mock_cerbos_check_resources: AsyncMock, mock_ows_account_client: MagicMock, mock_ows_participant_client: MagicMock, mock_redis_connector: MagicMock, api_route: str, action: str, cerbos_effect: str, expect_exception: bool, mock_pdp_tenant_roles: Dict[uuid.UUID, identity.TenantRoles], mock_my_adminable_tenant_roles: Dict[uuid.UUID, identity.TenantRoles], mock_async_cerbos_client: AsyncCerbosClient, mock_splitio_client: SplitioClient, ) -> None: """Test check_authorization_infra for infra endpoints.""" mock_splitio_client.get_treatment.return_value = FEATURE_ON principal_uuid = uuid.uuid4() mock_request = MagicMock() mock_request.get.return_value = APIRoute( path="some/path", name=api_route, endpoint=MagicMock() ) mock_build_infra_resources.return_value = [ identity.Resource( resource_id="some uuid", resource_type=constants.INFRA, ) ] mock_cerbos_check_resources.return_value = identity.CheckResourcesResponse( request_id="check_auth", resources=[ identity.CheckResourceActionResult( resource=mock_build_infra_resources.return_value[0], effect=cerbos_effect, action=action, ), ], ) impersonated_by_identity_uuid = uuid.uuid4() user_type = USER_TYPE_HUMAN expected_principal = Principal( identity_uuid=principal_uuid, user_type=user_type, pdp_tenant_roles={}, impersonated_by_identity_uuid=impersonated_by_identity_uuid, ) if expect_exception: with pytest.raises(Exception): await check_authorization_infra( mock_request, principal_uuid, 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, user_type=user_type, impersonated_by_identity_uuid=impersonated_by_identity_uuid, ) else: await check_authorization_infra( mock_request, principal_uuid, 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, user_type=user_type, impersonated_by_identity_uuid=impersonated_by_identity_uuid, ) mock_cerbos_check_resources.assert_awaited_once_with( identity_uuid=str(principal_uuid), check_resources_request=check_resources.CheckResourcesRequest( resources=[ identity.CheckResourceAction( resource=identity.Resource( resource_id="some uuid", resource_type=constants.INFRA, ), action=action, ) ] ), pdp_tenant_roles={}, ows_permissions_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=principal_uuid, principal=expected_principal, splitio_client=mock_splitio_client, ) mock_splitio_client.get_treatment.assert_called_once_with( attributes={"identity_id": str(impersonated_by_identity_uuid)}, feature_flag_name="pp_send_impersonated_by_identity_uuid", key="identity_id", ) @patch("pdp.logic.cerbos.RedisConnector") @patch("pdp.logic.cerbos.OwsParticipantClient") @patch("pdp.logic.cerbos.OwsAccountClient") @patch("pdp.fastapi.auth.cerbos.check_resources") @patch("pdp.fastapi.auth._build_infra_resources") async def test_check_authorization_infra__when_send_impersonated_by_identity_uuid_is_disabled( # noqa: E501 mock_build_infra_resources: AsyncMock, mock_cerbos_check_resources: AsyncMock, mock_ows_account_client: MagicMock, mock_ows_participant_client: MagicMock, mock_redis_connector: MagicMock, mock_pdp_tenant_roles: Dict[uuid.UUID, identity.TenantRoles], mock_my_adminable_tenant_roles: Dict[uuid.UUID, identity.TenantRoles], mock_async_cerbos_client: AsyncCerbosClient, mock_splitio_client: SplitioClient, ) -> None: """Test check_authorization_infra for infra endpoints when the FF is disabled.""" mock_splitio_client.get_treatment.return_value = FEATURE_OFF api_route = "id_to_uuid_exchange" action = "id_to_uuid_exchange" cerbos_effect = "allow" principal_uuid = uuid.uuid4() mock_request = MagicMock() mock_request.get.return_value = APIRoute( path="some/path", name=api_route, endpoint=MagicMock() ) mock_build_infra_resources.return_value = [ identity.Resource( resource_id="some uuid", resource_type=constants.INFRA, ) ] mock_cerbos_check_resources.return_value = identity.CheckResourcesResponse( request_id="check_auth", resources=[ identity.CheckResourceActionResult( resource=mock_build_infra_resources.return_value[0], effect=cerbos_effect, action=action, ), ], ) impersonated_by_identity_uuid = uuid.uuid4() user_type = USER_TYPE_HUMAN await check_authorization_infra( mock_request, principal_uuid, 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, user_type=user_type, impersonated_by_identity_uuid=impersonated_by_identity_uuid, ) mock_cerbos_check_resources.assert_awaited_once_with( identity_uuid=str(principal_uuid), check_resources_request=check_resources.CheckResourcesRequest( resources=[ identity.CheckResourceAction( resource=identity.Resource( resource_id="some uuid", resource_type=constants.INFRA, ), action=action, ) ] ), pdp_tenant_roles={}, ows_permissions_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=principal_uuid, splitio_client=mock_splitio_client, ) mock_splitio_client.get_treatment.assert_called_once_with( attributes={"identity_id": str(impersonated_by_identity_uuid)}, feature_flag_name="pp_send_impersonated_by_identity_uuid", key="identity_id", ) @patch("pdp.fastapi.auth.cerbos") async def test_check_authorization_infra_bad_route( mock_cerbos: MagicMock, ) -> None: """Test check_authorization_infra throws error for unexpected routes.""" mock_request = MagicMock() mock_request.get.return_value = APIRoute( # check_identity_resources is not an infra route path="some/path", name="check_identity_resources", endpoint=MagicMock(), ) some_uuid = uuid.uuid4() with pytest.raises(Exception): await check_authorization_infra(mock_request, some_uuid, mock_cerbos) @patch("pdp.fastapi.auth.cerbos") async def test_check_authorization_bad_route( mock_cerbos: MagicMock, identity_uuid: str, ) -> None: """Test check_authorization for role administration operations.""" mock_request = MagicMock() mock_request.get.return_value = APIRoute( path="some/path", name="unknown", endpoint=MagicMock() ) mock_request.get.return_value.name = "unknown" mock_cerbos.check_resources = AsyncMock() mock_cerbos.check_resources.return_value = {"resources": [{"effect": "allow"}]} some_uuid = uuid.uuid4() with pytest.raises(Exception): await check_authorization(mock_request, some_uuid, uuid.UUID(identity_uuid)) @pytest.mark.parametrize( "items, total_records, expected, description", [ ([], 0, {}, "Empty list is an empty dict of tenants"), ( [ OwsPermissionsResource( type="Vendor", id="1", uuid="edec3b6e-1070-4363-b6b3-6cbed675f3bc" ), ], 1, { uuid.UUID( "edec3b6e-1070-4363-b6b3-6cbed675f3bc" ): identity.TenantRoles.model_validate( { "tenant_type": "account", "tenant_uuid": "edec3b6e-1070-4363-b6b3-6cbed675f3bc", "roles": [{"role": "ows_permissions_rap_admin"}], } ), }, "Vendor resource is converted to TenantRoles", ), ( [ OwsPermissionsResource( type="LabelParticipant", id="0", uuid="fe55b133-28ae-45eb-8080-3e26a164110d", ), OwsPermissionsResource( type="VeNDor", id="1", uuid="edec3b6e-1070-4363-b6b3-6cbed675f3bc" ), OwsPermissionsResource( type="Collaborator", id="2", uuid="91d9aea9-c6a7-4d33-9fec-0c61a5b3c04f", ), ], 3, { uuid.UUID( "edec3b6e-1070-4363-b6b3-6cbed675f3bc" ): identity.TenantRoles.model_validate( { "tenant_type": "account", "tenant_uuid": "edec3b6e-1070-4363-b6b3-6cbed675f3bc", "roles": [{"role": "ows_permissions_rap_admin"}], } ), }, "Unsupported resource types are ignored for TenantRoles", ), ], ) async def test_get_principal_ows_permissions_from_scope( items: List[OwsPermissionsResource], total_records: int, expected: Dict[uuid.UUID, identity.TenantRoles], description: str, mock_ows_permissions_client: MagicMock, ) -> None: """Test get_principal_ows_permissions_from_scope.""" mock_ows_permissions_client.collect_get_my_adminable_resources = AsyncMock( return_value=items ) mock_request = MagicMock() mock_request.scope = {} actual = await get_principal_ows_permissions_from_scope( mock_request, mock_ows_permissions_client ) assert actual == expected, description mock_ows_permissions_client.collect_get_my_adminable_resources.assert_called_once_with() # noqa: E501 assert mock_request.scope["principal_ows_permissions"] == expected, ( "Request scope should be populated with adminable resources" ) @pytest.mark.parametrize( "status_code", [status_code for status_code in range(400, 500)], ) async def test_get_principal_ows_permissions_from_scope_warns( status_code: int, mock_ows_permissions_client: MagicMock, caplog: Any, ) -> None: """Test get_principal_ows_permissions_from_scope warns on 40x.""" mock_response = MagicMock() mock_response.status_code = status_code mock_ows_permissions_client.collect_get_my_adminable_resources = AsyncMock( side_effect=HTTPStatusError( message=f"mock error {status_code}", request=MagicMock(), response=mock_response, ) ) mock_request = MagicMock() mock_request.scope = {} await get_principal_ows_permissions_from_scope( mock_request, mock_ows_permissions_client ) with caplog.at_level(logging.ERROR): assert len(caplog.records) == 1, ( f"Status: {status_code} should be considered a bad request" ) assert caplog.records[0].levelname == "WARNING", ( f"Status {status_code} should only be a warning" ) assert ( f"Bad request to ows-permissions mock error {status_code}" == caplog.records[0].msg ) @pytest.mark.parametrize( "status_code", [status_code for status_code in range(500, 600)], ) async def test_get_principal_ows_permissions_from_scope_errors( status_code: int, mock_ows_permissions_client: MagicMock, caplog: Any, ) -> None: """Test get_principal_ows_permissions_from_scope logs errors on 50x.""" mock_response = MagicMock() mock_response.status_code = status_code mock_ows_permissions_client.collect_get_my_adminable_resources = AsyncMock( side_effect=HTTPStatusError( message=f"mock error {status_code}", request=MagicMock(), response=mock_response, ) ) mock_request = MagicMock() mock_request.scope = {} await get_principal_ows_permissions_from_scope( mock_request, mock_ows_permissions_client ) with caplog.at_level(logging.ERROR): assert len(caplog.records) == 1, ( f"Status: {status_code} should be considered a server error" ) assert caplog.records[0].levelname == "ERROR", ( f"Status {status_code} should be an error" ) assert ( f"Server error from ows-permissions mock error {status_code}" == caplog.records[0].msg ) @pytest.mark.parametrize( "path_params, expected, json_call_count, description", [ ({}, [], 0, "attributes should be empty"), ( {"tenant_uuid": "67e7303c-0e2f-41b9-b247-420f1ccc8a4f"}, [ identity.Resource( resource_id="ab123456-1234-4c2b-9c23-123ab4000a1b", resource_type="identity", attributes={ "tenant": { "tenant_uuid": "67e7303c-0e2f-41b9-b247-420f1ccc8a4f", "tenant_type": "account", }, "identity_uuid": "ab123456-1234-4c2b-9c23-123ab4000a1b", }, ) ], 1, "tenant uuid should be in attributes", ), ], ) @pytest.mark.parametrize( "mock_request_json", [ pytest.param( { "tenant_type": "account", "tenant_uuid": "67e7303c-0e2f-41b9-b247-420f1ccc8a4f", "roles_to_attach": [], "roles_to_detach": [], }, id="Representative of a AttachDetachRolesRequest", ), pytest.param( { "tenant_type": "account", "tenant_uuid": "67e7303c-0e2f-41b9-b247-420f1ccc8a4f", }, id="Representative of a DeactivateOneRequest", ), ], ) @pytest.mark.parametrize( "route_name", [ ("attach_detach_roles_by_identity_tenant"), ("deactivate_one"), ], ) async def test_build_identity_resources_for_single_tenant( route_name: str, mock_request_json: Dict[str, Any], path_params: Dict[str, Any], expected: List[identity.Resource], json_call_count: int, description: str, identity_uuid_as_uuid: uuid.UUID, ) -> None: """Test _build_identity_resource for single tenant operation.""" mock_request = MagicMock(spec=Request) mock_request.path_params = path_params mock_request.json.return_value = mock_request_json mock_api_route = MagicMock() mock_api_route.name = route_name actual = await _build_identity_resources( mock_request, mock_api_route, identity_uuid_as_uuid, ) assert actual == expected, description assert mock_request.json.call_count == json_call_count @pytest.mark.parametrize( "route_name", [ pytest.param("get_roles_by_identity"), pytest.param("check_identity_resources"), pytest.param("deactivate_all"), ], ) async def test_build_identity_resources_for_bulk_operation( route_name: str, identity_uuid: str, ) -> None: """Test _build_identity_resource for authorizing a bulk operation.""" mock_request = MagicMock(spec=Request) mock_api_route = MagicMock() mock_api_route.name = route_name actual = await _build_identity_resources( mock_request, mock_api_route, uuid.UUID(identity_uuid) ) assert actual == [ identity.Resource( resource_id=identity_uuid, resource_type="identity", attributes={"identity_uuid": identity_uuid}, ) ], "List of one resource with the identity should be returned" assert mock_request.json.call_count == 0 @pytest.mark.parametrize( "input_tenants, input_identities, cerbos_decision, output_tenants", [ pytest.param( {}, [], identity.CheckResourcesResponse(request_id="1", resources=[]), {}, id="empty dict is an empty dict still", ), pytest.param( { uuid.UUID("3cc6e04c-4ee8-44be-ae83-7185a6e63278"): identity.TenantRoles( tenant_type="account", tenant_uuid=uuid.UUID("3cc6e04c-4ee8-44be-ae83-7185a6e63278"), roles=[], ) }, [ identity.Resource( resource_id="identity#3cc6e04c-4ee8-44be-ae83-7185a6e63278", resource_type="identity", attributes={ "tenant": {"tenant_uuid": "", "tenant_type": "account"} }, ) ], identity.CheckResourcesResponse( request_id="2", resources=[ identity.CheckResourceActionResult( resource=identity.Resource( resource_id="identity#3cc6e04c-4ee8-44be-ae83-7185a6e63278", resource_type="identity", ), action="view", effect="allow", ) ], ), { uuid.UUID("3cc6e04c-4ee8-44be-ae83-7185a6e63278"): identity.TenantRoles( tenant_type="account", tenant_uuid=uuid.UUID("3cc6e04c-4ee8-44be-ae83-7185a6e63278"), roles=[], ) }, id="tenant is kept because decision was allow", ), pytest.param( { uuid.UUID("3cc6e04c-4ee8-44be-ae83-7185a6e63278"): identity.TenantRoles( tenant_type="account", tenant_uuid=uuid.UUID("3cc6e04c-4ee8-44be-ae83-7185a6e63278"), roles=[], ) }, [ identity.Resource( resource_id="identity#3cc6e04c-4ee8-44be-ae83-7185a6e63278", resource_type="identity", attributes={ "tenant": {"tenant_uuid": "", "tenant_type": "account"} }, ) ], identity.CheckResourcesResponse( request_id="3", resources=[ identity.CheckResourceActionResult( resource=identity.Resource( resource_id="identity#3cc6e04c-4ee8-44be-ae83-7185a6e63278", resource_type="identity", ), action="view", effect="deny", ) ], ), {}, id="tenant is removed because decision is denied", ), ], ) @patch("pdp.logic.cerbos.RedisConnector") @patch("pdp.logic.cerbos.OwsParticipantClient") @patch("pdp.logic.cerbos.OwsAccountClient") @patch("pdp.fastapi.auth.cerbos") @patch("pdp.fastapi.auth.build_identity_resource_from_tenant_role") async def test_filter_for_identity_tenants( mock_build_identity_resource_from_tenant_role: MagicMock, mock_cerbos: MagicMock, mock_ows_account_client: MagicMock, mock_ows_participant_client: MagicMock, mock_redis_connector: MagicMock, input_tenants: Dict[uuid.UUID, identity.TenantRoles], input_identities: List[identity.Resource], cerbos_decision: identity.CheckResourcesResponse, output_tenants: Dict[uuid.UUID, identity.TenantRoles], identity_uuid: str, mock_my_adminable_tenant_roles: Dict[uuid.UUID, identity.TenantRoles], mock_async_cerbos_client: AsyncCerbosClient, caplog: Any, mock_splitio_client: SplitioClient, ) -> None: """Test filter_for_identity_tenants.""" mock_splitio_client.get_treatment.return_value = FEATURE_ON requested_identity_uuid = uuid.uuid4() mock_cerbos.check_resources = AsyncMock(return_value=cerbos_decision) mock_build_identity_resource_from_tenant_role.side_effect = input_identities impersonated_by_identity_uuid = uuid.uuid4() user_type = USER_TYPE_HUMAN expected_principal = Principal( identity_uuid=uuid.UUID(identity_uuid), user_type=user_type, pdp_tenant_roles={}, ows_permissions_tenant_roles=mock_my_adminable_tenant_roles, impersonated_by_identity_uuid=impersonated_by_identity_uuid, ) actual = await filter_for_identity_tenants( requested_identity_uuid, tenants=input_tenants, action="view", impersonated_by_identity_uuid=impersonated_by_identity_uuid, user_type=user_type, principal_identity_uuid=uuid.UUID(identity_uuid), 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, splitio_client=mock_splitio_client, ) if len(input_tenants) == 0: assert mock_build_identity_resource_from_tenant_role.call_count == 0 assert mock_cerbos.check_resources.call_count == 0 else: assert mock_build_identity_resource_from_tenant_role.call_count == len( input_tenants ) mock_cerbos.check_resources.assert_called_once_with( identity_uuid, check_resources.CheckResourcesRequest( resources=[ identity.CheckResourceAction( action="view", resource=input_identity, ) for input_identity in input_identities ] ), 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, authenticated_identity_uuid=uuid.UUID(identity_uuid), principal=expected_principal, splitio_client=mock_splitio_client, ) mock_splitio_client.get_treatment.assert_called_once_with( attributes={"identity_id": str(impersonated_by_identity_uuid)}, feature_flag_name="pp_send_impersonated_by_identity_uuid", key="identity_id", ) assert actual == output_tenants if input_tenants and not output_tenants: with caplog.at_level(logging.INFO): assert ( "List of tenants that the principal was not allowed to view:" in caplog.text ) @patch("pdp.logic.cerbos.RedisConnector") @patch("pdp.logic.cerbos.OwsParticipantClient") @patch("pdp.logic.cerbos.OwsAccountClient") @patch("pdp.fastapi.auth.cerbos") @patch("pdp.fastapi.auth.build_identity_resource_from_tenant_role") async def test_filter_for_identity_tenants__ff_disabled( mock_build_identity_resource_from_tenant_role: MagicMock, mock_cerbos: MagicMock, mock_ows_account_client: MagicMock, mock_ows_participant_client: MagicMock, mock_redis_connector: MagicMock, identity_uuid: str, mock_my_adminable_tenant_roles: Dict[uuid.UUID, identity.TenantRoles], mock_async_cerbos_client: AsyncCerbosClient, caplog: Any, mock_splitio_client: SplitioClient, ) -> None: """Test filter_for_identity_tenants with the FF disabled.""" mock_splitio_client.get_treatment.return_value = FEATURE_OFF input_tenants = { uuid.UUID("3cc6e04c-4ee8-44be-ae83-7185a6e63278"): identity.TenantRoles( tenant_type="account", tenant_uuid=uuid.UUID("3cc6e04c-4ee8-44be-ae83-7185a6e63278"), roles=[], ) } input_identities = [ identity.Resource( resource_id="identity#3cc6e04c-4ee8-44be-ae83-7185a6e63278", resource_type="identity", attributes={"tenant": {"tenant_uuid": "", "tenant_type": "account"}}, ) ] cerbos_decision = identity.CheckResourcesResponse( request_id="3", resources=[ identity.CheckResourceActionResult( resource=identity.Resource( resource_id="identity#3cc6e04c-4ee8-44be-ae83-7185a6e63278", resource_type="identity", ), action="view", effect="deny", ) ], ) output_tenants: Dict[uuid.UUID, identity.TenantRoles] = {} requested_identity_uuid = uuid.uuid4() mock_cerbos.check_resources = AsyncMock(return_value=cerbos_decision) mock_build_identity_resource_from_tenant_role.side_effect = input_identities impersonated_by_identity_uuid = uuid.uuid4() user_type = USER_TYPE_HUMAN actual = await filter_for_identity_tenants( requested_identity_uuid, tenants=input_tenants, action="view", impersonated_by_identity_uuid=impersonated_by_identity_uuid, user_type=user_type, principal_identity_uuid=uuid.UUID(identity_uuid), 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, splitio_client=mock_splitio_client, ) assert mock_build_identity_resource_from_tenant_role.call_count == len( input_tenants ) mock_cerbos.check_resources.assert_called_once_with( identity_uuid, check_resources.CheckResourcesRequest( resources=[ identity.CheckResourceAction( action="view", resource=input_identity, ) for input_identity in input_identities ] ), 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, authenticated_identity_uuid=uuid.UUID(identity_uuid), splitio_client=mock_splitio_client, ) assert actual == output_tenants if input_tenants and not output_tenants: with caplog.at_level(logging.INFO): assert ( "List of tenants that the principal was not allowed to view:" in caplog.text ) @patch("pdp.fastapi.auth.save_cache_object") @patch("pdp.fastapi.auth.get_object_from_cache") @patch("pdp.fastapi.auth.get_roles_by_identity") async def test_get_identity_pdp_tenant_roles_from_scope( mock_get_roles_by_identity: MagicMock, mock_get_object_from_cache: AsyncMock, mock_save_cache_object: AsyncMock, mock_identity_ddb_connector: MagicMock, mock_redis_connector: AsyncMock, ) -> None: """Test get_identity_pdp_tenant_roles_from_scope calls DynamoDB once.""" mock_get_object_from_cache.return_value = None mock_get_roles_by_identity.return_value = identity.RolesResponse.model_validate( { "tenants": { "04b48f72-5b47-425f-8b49-21f1ebc3f0cd": { "tenant_type": "account", "tenant_uuid": "04b48f72-5b47-425f-8b49-21f1ebc3f0cd", "roles": [{"role": "dog_whisperer"}], } }, "cursor": {"cursor": None}, } ) mock_save_cache_object.return_value = True mock_request = MagicMock() mock_request.scope = {} identity_uuid = uuid.uuid4() result = await get_identity_pdp_tenant_roles_from_scope( mock_request, identity_uuid, identity_ddb_connector=mock_identity_ddb_connector, redis_connector=mock_redis_connector, ) mock_get_object_from_cache.assert_called_once() mock_get_roles_by_identity.assert_called_once_with( str(identity_uuid), identity_ddb_connector=mock_identity_ddb_connector, ) mock_save_cache_object.assert_called_once() assert result == identity.TenantRolesMapValidator.validate_python( { "04b48f72-5b47-425f-8b49-21f1ebc3f0cd": { "tenant_type": "account", "tenant_uuid": "04b48f72-5b47-425f-8b49-21f1ebc3f0cd", "roles": [{"role": "dog_whisperer"}], } } ) assert mock_request.scope[f"principal_pdp@identity_uuid#{identity_uuid}"] == result @patch("pdp.fastapi.auth.get_roles_by_identity") async def test_get_identity_pdp_tenant_roles_from_scope_uses_it( mock_get_roles_by_identity: MagicMock, mock_identity_ddb_connector: MagicMock, ) -> None: """Test get_identity_pdp_tenant_roles_from_scope uses scope.""" expected = identity.TenantRolesMapValidator.validate_python( { "04b48f72-5b47-425f-8b49-21f1ebc3f0cd": { "tenant_type": "account", "tenant_uuid": "04b48f72-5b47-425f-8b49-21f1ebc3f0cd", "roles": [{"role": "dog_whisperer"}], }, "7e468f76-3848-4ba3-aaa9-dbc223eb11ca": { "tenant_type": "subaccount", "tenant_uuid": "7e468f76-3848-4ba3-aaa9-dbc223eb11ca", "roles": [{"role": "cat_scratcher"}], }, } ) mock_request = MagicMock() identity_uuid = uuid.uuid4() mock_request.scope = {f"principal_pdp@identity_uuid#{identity_uuid}": expected} result = await get_identity_pdp_tenant_roles_from_scope( mock_request, identity_uuid, identity_ddb_connector=mock_identity_ddb_connector, ) mock_get_roles_by_identity.assert_not_called() assert result == expected @pytest.mark.parametrize( "mock_tenant_roles_from_cache, mock_get_roles_by_identity_return_response", [ pytest.param( identity.TenantRolesMapValidator.validate_python( { "04b48f72-5b47-425f-8b49-21f1ebc3f0cd": { "tenant_type": "account", "tenant_uuid": "04b48f72-5b47-425f-8b49-21f1ebc3f0cd", "roles": [{"role": "dog_whisperer"}], } } ), None, id="Cached tenant roles are used", ), pytest.param( {}, None, id="Cached tenant roles are used, even if it is an empty dict" ), pytest.param( None, identity.RolesResponse.model_validate( { "tenants": { "8fdcb9bc-418f-43d4-ac9d-fbd0250ef7de": { "tenant_type": "account", "tenant_uuid": "8fdcb9bc-418f-43d4-ac9d-fbd0250ef7de", "roles": [{"role": "dog_whisperer"}], } }, "cursor": {"cursor": None}, } ), id="get_roles_by_identity is called when cache is empty", ), ], ) @patch("pdp.fastapi.auth.PydanticSchemaSerializer") @patch("pdp.fastapi.auth.get_roles_by_identity") @patch("pdp.fastapi.auth.get_object_from_cache") @patch("pdp.fastapi.auth.save_cache_object") async def test_get_identity_pdp_tenant_roles_from_scope_cached_data( mock_save_cache_object: AsyncMock, mock_get_object_from_cache: AsyncMock, mock_get_roles_by_identity: MagicMock, mock_serializer: MagicMock, mock_identity_ddb_connector: MagicMock, mock_redis_connector: MagicMock, mock_tenant_roles_from_cache: Dict[uuid.UUID, identity.TenantRoles] | None, mock_get_roles_by_identity_return_response: identity.RolesResponse | None, ) -> None: """Test get_identity_pdp_tenant_roles_from_scope uses cached data.""" mock_get_roles_by_identity.return_value = mock_get_roles_by_identity_return_response mock_get_object_from_cache.return_value = mock_tenant_roles_from_cache mock_request = MagicMock() identity_uuid = uuid.uuid4() mock_request.scope = {} result = await get_identity_pdp_tenant_roles_from_scope( mock_request, identity_uuid, identity_ddb_connector=mock_identity_ddb_connector, redis_connector=mock_redis_connector, ) mock_get_object_from_cache.assert_called_once_with( key=f"principal_pdp@identity_uuid#{identity_uuid}", redis_connector=mock_redis_connector, serializer=mock_serializer(PrincipalPdpCacheObject), cache_attribute_name="tenant_roles", ) if mock_tenant_roles_from_cache is not None: mock_get_roles_by_identity.assert_not_called() mock_save_cache_object.assert_not_called() assert result == mock_tenant_roles_from_cache else: assert mock_get_roles_by_identity_return_response mock_get_roles_by_identity.assert_called_once_with( str(identity_uuid), identity_ddb_connector=mock_identity_ddb_connector, ) assert result == mock_get_roles_by_identity_return_response.tenants mock_save_cache_object.assert_called_once_with( cache_object=PrincipalPdpCacheObject( identity_uuid=str(identity_uuid), tenant_roles=mock_get_roles_by_identity_return_response.tenants, ), cache_model_type=PrincipalPdpCacheObject, redis_connector=mock_redis_connector, ) @patch("pdp.fastapi.auth.save_cache_object") @patch("pdp.fastapi.auth.get_roles_by_identity") @patch("pdp.fastapi.auth.get_object_from_cache") async def test_get_identity_pdp_tenant_roles_from_scope_aggregates( mock_get_object_from_cache: AsyncMock, mock_get_roles_by_identity: MagicMock, mock_save_cache_object: AsyncMock, mock_identity_ddb_connector: MagicMock, mock_redis_connector: AsyncMock, ) -> None: """Test get_identity_pdp_tenant_roles_from_scope paginates DynamoDB.""" mock_get_object_from_cache.return_value = None mock_save_cache_object.return_value = True mock_get_roles_by_identity.side_effect = [ identity.RolesResponse.model_validate( { "tenants": { "04b48f72-5b47-425f-8b49-21f1ebc3f0cd": { "tenant_type": "account", "tenant_uuid": "04b48f72-5b47-425f-8b49-21f1ebc3f0cd", "roles": [{"role": "dog_whisperer"}], } }, "cursor": {"cursor": "curses"}, } ), identity.RolesResponse.model_validate( { "tenants": { "7e468f76-3848-4ba3-aaa9-dbc223eb11ca": { "tenant_type": "subaccount", "tenant_uuid": "7e468f76-3848-4ba3-aaa9-dbc223eb11ca", "roles": [{"role": "cat_scratcher"}], } }, "cursor": {"cursor": None}, } ), ] mock_request = MagicMock() mock_request.scope = {} identity_uuid = uuid.uuid4() result = await get_identity_pdp_tenant_roles_from_scope( mock_request, identity_uuid, identity_ddb_connector=mock_identity_ddb_connector, redis_connector=mock_redis_connector, ) mock_get_object_from_cache.assert_called_once() mock_save_cache_object.assert_called_once() mock_get_roles_by_identity.assert_has_calls( [ call( str(identity_uuid), identity_ddb_connector=mock_identity_ddb_connector, ), call( str(identity_uuid), identity_ddb_connector=mock_identity_ddb_connector, cursor="curses", ), ] ) assert result == identity.TenantRolesMapValidator.validate_python( { "04b48f72-5b47-425f-8b49-21f1ebc3f0cd": { "tenant_type": "account", "tenant_uuid": "04b48f72-5b47-425f-8b49-21f1ebc3f0cd", "roles": [{"role": "dog_whisperer"}], }, "7e468f76-3848-4ba3-aaa9-dbc223eb11ca": { "tenant_type": "subaccount", "tenant_uuid": "7e468f76-3848-4ba3-aaa9-dbc223eb11ca", "roles": [{"role": "cat_scratcher"}], }, } ) assert mock_request.scope[f"principal_pdp@identity_uuid#{identity_uuid}"] == result @patch("pdp.fastapi.auth.get_identity_pdp_tenant_roles_from_scope") async def test_get_principal_pdp_tenant_roles_from_scope( mock_get_identity_pdp_tenant_roles_from_scope: AsyncMock, mock_identity_ddb_connector: MagicMock, mock_redis_connector: AsyncMock, identity_uuid: str, ) -> None: """Test get_principal_pdp_tenant_roles_from_scope uses generic fn.""" expected = identity.TenantRolesMapValidator.validate_python( { "04b48f72-5b47-425f-8b49-21f1ebc3f0cd": { "tenant_type": "account", "tenant_uuid": "04b48f72-5b47-425f-8b49-21f1ebc3f0cd", "roles": [{"role": "dog_whisperer"}], }, "7e468f76-3848-4ba3-aaa9-dbc223eb11ca": { "tenant_type": "subaccount", "tenant_uuid": "7e468f76-3848-4ba3-aaa9-dbc223eb11ca", "roles": [{"role": "cat_scratcher"}], }, } ) mock_get_identity_pdp_tenant_roles_from_scope.return_value = expected mock_request = MagicMock() result = await get_principal_pdp_tenant_roles_from_scope( mock_request, uuid.UUID(identity_uuid), identity_ddb_connector=mock_identity_ddb_connector, redis_connector=mock_redis_connector, ) assert result == expected mock_get_identity_pdp_tenant_roles_from_scope.assert_awaited_once_with( mock_request, uuid.UUID(identity_uuid), identity_ddb_connector=mock_identity_ddb_connector, redis_connector=mock_redis_connector, )