"""Test Identity Logic.""" import logging import uuid from typing import Any, Dict, List, Optional from unittest.mock import AsyncMock, MagicMock, patch import freezegun import pytest from fastapi import HTTPException 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 USER_TYPE_MACHINE, AuthEffect, TenantType from pdp.fastapi.schemas.check_resource_type_actions import ( CheckResourceTypeAction, CheckResourceTypeActionResult, CheckResourceTypeActionsRequest, CheckResourceTypeActionsResponse, CreateCheckResourceActionsError, ) from pdp.fastapi.schemas.check_resources import CheckResourcesRequest from pdp.fastapi.schemas.deactivation import DeactivationSummary from pdp.fastapi.schemas.get_allowed_tenants import ( AllowedTenant, GetAllowedTenantsRequest, GetAllowedTenantsResponse, ) from pdp.fastapi.schemas.identity import ( CheckResourceAction, CheckResourceActionResult, CheckResourcesResponse, IdentityTenant, Resource, Role, RolesResponse, TenantRoles, ) from pdp.fastapi.schemas.principal import Principal from pdp.logic.identity import ( _extend_with_child_roles_to_detach, _keep_supported_roles, attach_and_detach_roles, check_resource_type_actions, deactivate_many, get_allowed_tenants, get_roles_by_identity, is_tenant_assigned, ) @patch("pdp.models.identity.Identity.get_roles") def test_get_roles_by_identity_success( mock_get_roles: MagicMock, mock_identity_ddb_connector: MagicMock ) -> None: """Test logic.get_roles_by_identity success.""" mock_get_roles.return_value = { "items": [ { "tenant_type": {"S": "account"}, "tenant_uuid": {"S": "6b70c3fd-6048-4635-b84b-b60ddcde9e9a"}, "roles": {"L": []}, } ], "cursor": {"cursor": "next page pls"}, } identity_uuid = uuid.uuid4() result = get_roles_by_identity( str(identity_uuid), identity_ddb_connector=mock_identity_ddb_connector ) assert result == RolesResponse.model_validate( { "tenants": { "6b70c3fd-6048-4635-b84b-b60ddcde9e9a": { "tenant_type": "account", "tenant_uuid": "6b70c3fd-6048-4635-b84b-b60ddcde9e9a", "roles": [], } }, "cursor": {"cursor": "next page pls"}, } ) @patch( "pdp.models.identity.Identity.get_roles", side_effect=Exception("Unable to access DynamoDB"), ) def test_get_roles_by_identity_failure( mock_get_roles: MagicMock, mock_identity_ddb_connector: MagicMock ) -> None: """Test get_roles_by_identity failure.""" identity_uuid = uuid.uuid4() with pytest.raises(Exception) as exception_info: get_roles_by_identity( str(identity_uuid), identity_ddb_connector=mock_identity_ddb_connector ) assert str(exception_info.value) == "Unable to access DynamoDB" @pytest.mark.parametrize( "permissions_by_tenant_response, expected", [ pytest.param(None, False, id="None means tenant is not assigned"), pytest.param( IdentityTenant( identity_uuid="d04cc050-f3aa-4bcb-a84f-7c8e6da76608", tenant_type=TenantType.TENANT_TYPE_ACCOUNT, tenant_uuid="a9dd9b42-e53d-11ee-be6d-4a2888760682", roles=[{"role": "dog"}], ), True, id="A TenantRoles object with roles means the tenant is assigned", ), pytest.param( IdentityTenant( identity_uuid="d04cc050-f3aa-4bcb-a84f-7c8e6da76608", tenant_type=TenantType.TENANT_TYPE_ACCOUNT, tenant_uuid="a9dd9b42-e53d-11ee-be6d-4a2888760682", roles=[], ), True, id="A TenantRoles object without roles means the tenant is assigned", ), ], ) @patch("pdp.models.identity.Identity.get_permissions_by_tenant") def test_is_tenant_assigned( mock_get_permissions_by_tenant: MagicMock, permissions_by_tenant_response: Optional[IdentityTenant], expected: bool, tenant_1_uuid: uuid.UUID, mock_identity_ddb_connector: MagicMock, ) -> None: """Test is_tenant_assigned.""" mock_get_permissions_by_tenant.return_value = permissions_by_tenant_response identity_uuid = "d04cc050-f3aa-4bcb-a84f-7c8e6da76608" actual = is_tenant_assigned( identity_uuid=identity_uuid, tenant_uuid=tenant_1_uuid, identity_ddb_connector=mock_identity_ddb_connector, ) mock_get_permissions_by_tenant.assert_called_once_with(tenant_1_uuid) assert actual == expected @freezegun.freeze_time("2022-01-01") @pytest.mark.parametrize( "principal, expected_principal_identity_id, expected_impersonated_by", [ pytest.param( Principal( identity_uuid="5fecc5ca-f7b3-44d1-9121-03588e809454", pdp_tenant_roles={}, ), "5fecc5ca-f7b3-44d1-9121-03588e809454", None, id="principal is not impersonated", ), pytest.param( Principal( identity_uuid="5fecc5ca-f7b3-44d1-9121-03588e809454", pdp_tenant_roles={}, impersonated_by_identity_uuid="d8313f52-9c60-4782-8dd6-7c5c01ccb979", ), "5fecc5ca-f7b3-44d1-9121-03588e809454", "d8313f52-9c60-4782-8dd6-7c5c01ccb979", id="principal is impersonated", ), ], ) @patch("pdp.models.identity.Identity.deactivate_many") @patch("pdp.models.identity.Identity.get_permissions_by_tenant") @patch("pdp.models.identity.Identity.get_empty_tenant_permissions_state") @patch("pdp.models.identity.Identity.update_tenant_permissions") async def test_attach_and_detach_roles_create( mock_update_tenant_perms: MagicMock, mock_get_empty_tenant_perms: MagicMock, mock_get_tenant_perms: MagicMock, mock_deactivate_many: MagicMock, principal: Principal, expected_principal_identity_id: str, expected_impersonated_by: str | None, tenant_1_uuid: uuid.UUID, mock_identity_ddb_connector: MagicMock, caplog: Any, ) -> None: """Test attach_and_detach_roles.""" identity_uuid = uuid.uuid4() mock_get_tenant_perms.return_value = None mock_get_empty_tenant_perms.return_value = IdentityTenant( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[], updated_at="2022-01-01T00:00:00+00:00", updated_by=expected_principal_identity_id, updated_impersonated_by=expected_impersonated_by, created_at="2022-01-01T00:00:00+00:00", created_by=expected_principal_identity_id, created_impersonated_by=expected_impersonated_by, ) result = await attach_and_detach_roles( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type=TenantType.TENANT_TYPE_ACCOUNT, roles_to_attach=[], roles_to_detach=[], authenticated_identity_uuid="authenticated uuid", identity_ddb_connector=mock_identity_ddb_connector, principal=principal, ) assert ( result.model_dump() == IdentityTenant( identity_uuid=str(identity_uuid), tenant_type="account", tenant_uuid=tenant_1_uuid, roles=[], version="0", updated_at="2022-01-01T00:00:00+00:00", updated_by=expected_principal_identity_id, updated_impersonated_by=expected_impersonated_by, created_at="2022-01-01T00:00:00+00:00", created_by=expected_principal_identity_id, created_impersonated_by=expected_impersonated_by, ).model_dump() ) mock_get_tenant_perms.assert_called_once_with(tenant_1_uuid) mock_get_empty_tenant_perms.assert_called_once_with( tenant_uuid=tenant_1_uuid, tenant_type=TenantType.TENANT_TYPE_ACCOUNT, created_at="2022-01-01T00:00:00+00:00", created_by=expected_principal_identity_id, created_impersonated_by=expected_impersonated_by, ) mock_update_tenant_perms.assert_not_called() mock_deactivate_many.assert_not_called() assert ( caplog.records[0].msg == f"Not creating identity/tenant pair identity: {identity_uuid} and tenant: {tenant_1_uuid} as there are no roles being added." # noqa: E501 ) @freezegun.freeze_time("2022-01-01") @patch("pdp.models.identity.Identity.deactivate_many") @patch("pdp.models.identity.Identity.get_permissions_by_tenant") @patch("pdp.models.identity.Identity.get_empty_tenant_permissions_state") @patch("pdp.models.identity.Identity.update_tenant_permissions") async def test_attach_and_detach_roles_create_wo_principal( mock_update_tenant_perms: MagicMock, mock_get_empty_tenant_perms: MagicMock, mock_get_tenant_perms: MagicMock, mock_deactivate_many: MagicMock, tenant_1_uuid: uuid.UUID, mock_identity_ddb_connector: MagicMock, caplog: Any, ) -> None: """Test attach_and_detach_roles without principal arg.""" identity_uuid = uuid.uuid4() mock_get_tenant_perms.return_value = None mock_get_empty_tenant_perms.return_value = IdentityTenant( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[], updated_at="2022-01-01T00:00:00+00:00", updated_by="authenticated uuid", created_at="2022-01-01T00:00:00+00:00", created_by="authenticated uuid", ) result = await attach_and_detach_roles( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type=TenantType.TENANT_TYPE_ACCOUNT, roles_to_attach=[], roles_to_detach=[], authenticated_identity_uuid="authenticated uuid", identity_ddb_connector=mock_identity_ddb_connector, ) assert ( result.model_dump() == IdentityTenant( identity_uuid=str(identity_uuid), tenant_type="account", tenant_uuid=tenant_1_uuid, roles=[], version="0", updated_at="2022-01-01T00:00:00+00:00", updated_by="authenticated uuid", created_at="2022-01-01T00:00:00+00:00", created_by="authenticated uuid", ).model_dump() ) mock_get_tenant_perms.assert_called_once_with(tenant_1_uuid) mock_get_empty_tenant_perms.assert_called_once_with( tenant_uuid=tenant_1_uuid, tenant_type=TenantType.TENANT_TYPE_ACCOUNT, created_at="2022-01-01T00:00:00+00:00", created_by="authenticated uuid", ) mock_update_tenant_perms.assert_not_called() mock_deactivate_many.assert_not_called() assert ( caplog.records[0].msg == f"Not creating identity/tenant pair identity: {identity_uuid} and tenant: {tenant_1_uuid} as there are no roles being added." # noqa: E501 ) @freezegun.freeze_time("2022-01-01") @pytest.mark.parametrize( "principal, expected_principal_identity_id, expected_impersonated_by", [ pytest.param( Principal( identity_uuid="5fecc5ca-f7b3-44d1-9121-03588e809454", pdp_tenant_roles={}, ), "5fecc5ca-f7b3-44d1-9121-03588e809454", None, id="principal is not impersonated", ), pytest.param( Principal( identity_uuid="5fecc5ca-f7b3-44d1-9121-03588e809454", pdp_tenant_roles={}, impersonated_by_identity_uuid="d8313f52-9c60-4782-8dd6-7c5c01ccb979", ), "5fecc5ca-f7b3-44d1-9121-03588e809454", "d8313f52-9c60-4782-8dd6-7c5c01ccb979", id="principal is impersonated", ), ], ) @patch("pdp.logic.identity.deactivate_many") @patch("pdp.models.identity.Identity.get_permissions_by_tenant") @patch("pdp.models.identity.Identity.get_empty_tenant_permissions_state") @patch("pdp.models.identity.Identity.update_tenant_permissions") async def test_attach_and_detach_roles_update( mock_update_tenant_perms: MagicMock, mock_get_empty_tenant_perms: MagicMock, mock_get_tenant_perms: MagicMock, mock_deactivate_many: AsyncMock, principal: Principal, expected_principal_identity_id: str, expected_impersonated_by: str | None, tenant_1_uuid: uuid.UUID, mock_identity_ddb_connector: MagicMock, ) -> None: """Test attach_and_detach_roles updates an existing tenant permission.""" identity_uuid = uuid.uuid4() mock_get_tenant_perms.return_value = IdentityTenant( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[{"role": "songwhip_read"}, {"role": "audience_development_analyst"}], version="23", created_at="before", created_by="previous user", ) mock_update_tenant_perms.return_value = IdentityTenant( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[{"role": "songwhip_read"}, {"role": "audience_development_client"}], version="24", created_at="before", created_by="previous user", updated_by=expected_principal_identity_id, updated_impersonated_by=expected_impersonated_by, ) result = await attach_and_detach_roles( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type=TenantType.TENANT_TYPE_ACCOUNT, roles_to_attach=[ Role(role="audience_development_client"), Role(role="audience_development_analyst"), ], roles_to_detach=[Role(role="audience_development_analyst")], authenticated_identity_uuid="authenticated uuid", identity_ddb_connector=mock_identity_ddb_connector, principal=principal, ) assert result == IdentityTenant( identity_uuid=str(identity_uuid), tenant_type="account", tenant_uuid=tenant_1_uuid, roles=[{"role": "songwhip_read"}, {"role": "audience_development_client"}], version="24", created_at="before", created_by="previous user", updated_by=expected_principal_identity_id, updated_impersonated_by=expected_impersonated_by, ) mock_get_tenant_perms.assert_called_with(tenant_1_uuid) mock_update_tenant_perms.assert_called_with( tenant_1_uuid, IdentityTenant( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[{"role": "songwhip_read"}, {"role": "audience_development_client"}], version="24", updated_at="2022-01-01T00:00:00+00:00", updated_by=expected_principal_identity_id, updated_impersonated_by=expected_impersonated_by, created_at="before", created_by="previous user", ), ) mock_get_empty_tenant_perms.assert_not_called() mock_deactivate_many.assert_not_called() @freezegun.freeze_time("2022-01-01") @patch("pdp.logic.identity.deactivate_many") @patch("pdp.models.identity.Identity.get_permissions_by_tenant") @patch("pdp.models.identity.Identity.get_empty_tenant_permissions_state") @patch("pdp.models.identity.Identity.update_tenant_permissions") async def test_attach_and_detach_roles_update_wo_principal( mock_update_tenant_perms: MagicMock, mock_get_empty_tenant_perms: MagicMock, mock_get_tenant_perms: MagicMock, mock_deactivate_many: AsyncMock, tenant_1_uuid: uuid.UUID, mock_identity_ddb_connector: MagicMock, ) -> None: """Test attach_and_detach_roles updates an existing tenant permission wo principal.""" # noqa: E501 identity_uuid = uuid.uuid4() mock_get_tenant_perms.return_value = IdentityTenant( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[{"role": "songwhip_read"}, {"role": "audience_development_analyst"}], version="23", created_at="before", created_by="previous user", ) mock_update_tenant_perms.return_value = IdentityTenant( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[{"role": "songwhip_read"}, {"role": "audience_development_client"}], version="24", created_at="before", created_by="previous user", ) result = await attach_and_detach_roles( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type=TenantType.TENANT_TYPE_ACCOUNT, roles_to_attach=[ Role(role="audience_development_client"), Role(role="audience_development_analyst"), ], roles_to_detach=[Role(role="audience_development_analyst")], authenticated_identity_uuid="authenticated uuid", identity_ddb_connector=mock_identity_ddb_connector, ) assert result == IdentityTenant( identity_uuid=str(identity_uuid), tenant_type="account", tenant_uuid=tenant_1_uuid, roles=[{"role": "songwhip_read"}, {"role": "audience_development_client"}], version="24", created_at="before", created_by="previous user", ) mock_get_tenant_perms.assert_called_with(tenant_1_uuid) mock_update_tenant_perms.assert_called_with( tenant_1_uuid, IdentityTenant( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[{"role": "songwhip_read"}, {"role": "audience_development_client"}], version="24", updated_at="2022-01-01T00:00:00+00:00", updated_by="authenticated uuid", created_at="before", created_by="previous user", ), ) mock_get_empty_tenant_perms.assert_not_called() mock_deactivate_many.assert_not_called() @freezegun.freeze_time("2022-01-01") @pytest.mark.parametrize( "principal, expected_principal_identity_id, expected_impersonated_by", [ pytest.param( Principal( identity_uuid="5fecc5ca-f7b3-44d1-9121-03588e809454", pdp_tenant_roles={}, ), "5fecc5ca-f7b3-44d1-9121-03588e809454", None, id="principal is not impersonated", ), pytest.param( Principal( identity_uuid="5fecc5ca-f7b3-44d1-9121-03588e809454", pdp_tenant_roles={}, impersonated_by_identity_uuid="d8313f52-9c60-4782-8dd6-7c5c01ccb979", ), "5fecc5ca-f7b3-44d1-9121-03588e809454", "d8313f52-9c60-4782-8dd6-7c5c01ccb979", id="principal is impersonated", ), ], ) @patch("pdp.logic.identity.deactivate_many") @patch("pdp.models.identity.Identity.get_permissions_by_tenant") @patch("pdp.models.identity.Identity.get_empty_tenant_permissions_state") @patch("pdp.models.identity.Identity.update_tenant_permissions") async def test_attach_and_detach_roles_update_deactivates( mock_update_tenant_perms: MagicMock, mock_get_empty_tenant_perms: MagicMock, mock_get_tenant_perms: MagicMock, mock_deactivate_many: AsyncMock, principal: Principal, expected_principal_identity_id: str, expected_impersonated_by: str | None, tenant_1_uuid: uuid.UUID, mock_identity_ddb_connector: MagicMock, ) -> None: """Test attach_and_detach_roles deactivates when there are no roles left.""" identity_uuid = uuid.uuid4() mock_get_tenant_perms.return_value = IdentityTenant( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[{"role": "songwhip_read"}, {"role": "audience_development_analyst"}], version="23", ) mock_deactivate_many.return_value = DeactivationSummary(deleted=0, remaining=1) mock_get_empty_tenant_perms.return_value = IdentityTenant( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[], version="0", ) result = await attach_and_detach_roles( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type=TenantType.TENANT_TYPE_ACCOUNT, roles_to_attach=[], roles_to_detach=[ Role(role="songwhip_read"), Role(role="audience_development_analyst"), ], authenticated_identity_uuid="authenticated uuid", identity_ddb_connector=mock_identity_ddb_connector, principal=principal, ) assert ( result.model_dump() == IdentityTenant( identity_uuid=str(identity_uuid), tenant_type="account", tenant_uuid=tenant_1_uuid, roles=[], version="0", ).model_dump() ) mock_get_tenant_perms.assert_called_with(tenant_1_uuid) mock_deactivate_many.assert_called_once_with( identity_uuid=str(identity_uuid), tenants={ tenant_1_uuid: TenantRoles( roles=[], tenant_type="account", tenant_uuid=tenant_1_uuid ) }, identity_ddb_connector=mock_identity_ddb_connector, authenticated_identity_uuid="authenticated uuid", principal=principal, ) mock_get_empty_tenant_perms.assert_called_once_with( tenant_uuid=tenant_1_uuid, tenant_type=TenantType.TENANT_TYPE_ACCOUNT, created_at="2022-01-01T00:00:00+00:00", created_by=expected_principal_identity_id, created_impersonated_by=expected_impersonated_by, ) mock_update_tenant_perms.assert_not_called() @freezegun.freeze_time("2022-01-01") @patch("pdp.logic.identity.deactivate_many") @patch("pdp.models.identity.Identity.get_permissions_by_tenant") @patch("pdp.models.identity.Identity.get_empty_tenant_permissions_state") @patch("pdp.models.identity.Identity.update_tenant_permissions") async def test_attach_and_detach_roles_update_deactivates_wo_principal( mock_update_tenant_perms: MagicMock, mock_get_empty_tenant_perms: MagicMock, mock_get_tenant_perms: MagicMock, mock_deactivate_many: AsyncMock, tenant_1_uuid: uuid.UUID, mock_identity_ddb_connector: MagicMock, ) -> None: """Test attach_and_detach_roles deactivates when there are no roles left (wo principal arg).""" # noqa: E501 identity_uuid = uuid.uuid4() mock_get_tenant_perms.return_value = IdentityTenant( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[{"role": "songwhip_read"}, {"role": "audience_development_analyst"}], version="23", ) mock_deactivate_many.return_value = DeactivationSummary(deleted=0, remaining=1) mock_get_empty_tenant_perms.return_value = IdentityTenant( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[], version="0", ) result = await attach_and_detach_roles( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type=TenantType.TENANT_TYPE_ACCOUNT, roles_to_attach=[], roles_to_detach=[ Role(role="songwhip_read"), Role(role="audience_development_analyst"), ], authenticated_identity_uuid="authenticated uuid", identity_ddb_connector=mock_identity_ddb_connector, ) assert ( result.model_dump() == IdentityTenant( identity_uuid=str(identity_uuid), tenant_type="account", tenant_uuid=tenant_1_uuid, roles=[], version="0", ).model_dump() ) mock_get_tenant_perms.assert_called_with(tenant_1_uuid) mock_deactivate_many.assert_called_once_with( identity_uuid=str(identity_uuid), tenants={ tenant_1_uuid: TenantRoles( roles=[], tenant_type="account", tenant_uuid=tenant_1_uuid ) }, identity_ddb_connector=mock_identity_ddb_connector, authenticated_identity_uuid="authenticated uuid", principal=None, ) mock_get_empty_tenant_perms.assert_called_once_with( tenant_uuid=tenant_1_uuid, tenant_type=TenantType.TENANT_TYPE_ACCOUNT, created_at="2022-01-01T00:00:00+00:00", created_by="authenticated uuid", ) mock_update_tenant_perms.assert_not_called() @freezegun.freeze_time("2022-01-01") @pytest.mark.parametrize( "principal, expected_principal_identity_id, expected_impersonated_by", [ pytest.param( Principal( identity_uuid="5fecc5ca-f7b3-44d1-9121-03588e809454", pdp_tenant_roles={}, ), "5fecc5ca-f7b3-44d1-9121-03588e809454", None, id="principal is not impersonated", ), pytest.param( Principal( identity_uuid="5fecc5ca-f7b3-44d1-9121-03588e809454", pdp_tenant_roles={}, impersonated_by_identity_uuid="d8313f52-9c60-4782-8dd6-7c5c01ccb979", ), "5fecc5ca-f7b3-44d1-9121-03588e809454", "d8313f52-9c60-4782-8dd6-7c5c01ccb979", id="principal is impersonated", ), ], ) @patch("pdp.models.identity.Identity.get_permissions_by_tenant") @patch("pdp.models.identity.Identity.get_empty_tenant_permissions_state") @patch("pdp.models.identity.Identity.update_tenant_permissions") @patch("pdp.logic.identity.deactivate_many") async def test_attach_and_detach_roles_uses_extended_child_roles( mock_deactivate_many: AsyncMock, mock_update_tenant_perms: MagicMock, mock_get_empty_tenant_perms: MagicMock, mock_get_tenant_perms: MagicMock, principal: Principal, expected_principal_identity_id: str, expected_impersonated_by: str | None, tenant_1_uuid: uuid.UUID, mock_identity_ddb_connector: MagicMock, ) -> None: """Test attach_and_detach_roles uses extended child roles.""" identity_uuid = uuid.uuid4() mock_get_tenant_perms.return_value = IdentityTenant( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[ {"role": "fansifter_can_view_fan_data"}, {"role": "fansifter_can_create_ad_reports"}, ], version="23", ) mock_update_tenant_perms.return_value = IdentityTenant( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[], version="24", ) mock_get_empty_tenant_perms.return_value = IdentityTenant( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[], version="0", ) result = await attach_and_detach_roles( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type=TenantType.TENANT_TYPE_ACCOUNT, roles_to_attach=[], roles_to_detach=[Role(role="fansifter_can_view_fan_data")], authenticated_identity_uuid="authenticated uuid", identity_ddb_connector=mock_identity_ddb_connector, principal=principal, ) assert result == IdentityTenant( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[], version="0", ) mock_get_tenant_perms.assert_called_once_with(tenant_1_uuid) mock_get_empty_tenant_perms.assert_called_once_with( tenant_uuid=tenant_1_uuid, tenant_type=TenantType.TENANT_TYPE_ACCOUNT, created_at="2022-01-01T00:00:00+00:00", created_by=expected_principal_identity_id, created_impersonated_by=expected_impersonated_by, ) mock_deactivate_many.assert_called_once_with( identity_uuid=str(identity_uuid), tenants={ tenant_1_uuid: TenantRoles( tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[], ) }, identity_ddb_connector=mock_identity_ddb_connector, authenticated_identity_uuid="authenticated uuid", principal=principal, ) @freezegun.freeze_time("2022-01-01") @patch("pdp.models.identity.Identity.get_permissions_by_tenant") @patch("pdp.models.identity.Identity.get_empty_tenant_permissions_state") @patch("pdp.models.identity.Identity.update_tenant_permissions") @patch("pdp.logic.identity.deactivate_many") async def test_attach_and_detach_roles_uses_extended_child_roles_wo_principal( mock_deactivate_many: AsyncMock, mock_update_tenant_perms: MagicMock, mock_get_empty_tenant_perms: MagicMock, mock_get_tenant_perms: MagicMock, tenant_1_uuid: uuid.UUID, mock_identity_ddb_connector: MagicMock, ) -> None: """Test attach_and_detach_roles uses extended child roles (wo principal arg).""" identity_uuid = uuid.uuid4() mock_get_tenant_perms.return_value = IdentityTenant( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[ {"role": "fansifter_can_view_fan_data"}, {"role": "fansifter_can_create_ad_reports"}, ], version="23", ) mock_update_tenant_perms.return_value = IdentityTenant( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[], version="24", ) mock_get_empty_tenant_perms.return_value = IdentityTenant( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[], version="0", ) result = await attach_and_detach_roles( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type=TenantType.TENANT_TYPE_ACCOUNT, roles_to_attach=[], roles_to_detach=[Role(role="fansifter_can_view_fan_data")], authenticated_identity_uuid="authenticated uuid", identity_ddb_connector=mock_identity_ddb_connector, ) assert ( result.model_dump() == IdentityTenant( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[], version="0", ).model_dump() ) mock_get_tenant_perms.assert_called_once_with(tenant_1_uuid) mock_get_empty_tenant_perms.assert_called_once_with( tenant_uuid=tenant_1_uuid, tenant_type=TenantType.TENANT_TYPE_ACCOUNT, created_at="2022-01-01T00:00:00+00:00", created_by="authenticated uuid", ) mock_deactivate_many.assert_called_once_with( identity_uuid=str(identity_uuid), tenants={ tenant_1_uuid: TenantRoles( tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[], version="0", ) }, identity_ddb_connector=mock_identity_ddb_connector, authenticated_identity_uuid="authenticated uuid", principal=None, ) @freezegun.freeze_time("2022-01-01") @pytest.mark.parametrize( "principal, expected_principal_identity_id, expected_impersonated_by", [ pytest.param( Principal( identity_uuid="5fecc5ca-f7b3-44d1-9121-03588e809454", pdp_tenant_roles={}, ), "5fecc5ca-f7b3-44d1-9121-03588e809454", None, id="principal is not impersonated", ), pytest.param( Principal( identity_uuid="5fecc5ca-f7b3-44d1-9121-03588e809454", pdp_tenant_roles={}, impersonated_by_identity_uuid="d8313f52-9c60-4782-8dd6-7c5c01ccb979", ), "5fecc5ca-f7b3-44d1-9121-03588e809454", "d8313f52-9c60-4782-8dd6-7c5c01ccb979", id="principal is impersonated", ), ], ) @patch("pdp.models.identity.Identity.get_permissions_by_tenant") @patch("pdp.models.identity.Identity.get_empty_tenant_permissions_state") @patch("pdp.models.identity.Identity.update_tenant_permissions") async def test_attach_and_detach_roles_keeps_for_supported_roles( mock_update_tenant_perms: MagicMock, mock_get_empty_tenant_perms: MagicMock, mock_get_tenant_perms: MagicMock, principal: Principal, expected_principal_identity_id: str, expected_impersonated_by: str | None, tenant_1_uuid: uuid.UUID, mock_identity_ddb_connector: MagicMock, caplog: Any, ) -> None: """Test attach_and_detach_roles attaches only supported roles.""" identity_uuid = uuid.uuid4() mock_get_tenant_perms.return_value = IdentityTenant( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[ {"role": "audience_development_analyst"}, ], version="23", ) mock_update_tenant_perms.return_value = IdentityTenant( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[{"role": "audience_development_analyst"}], version="24", ) result = await attach_and_detach_roles( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type=TenantType.TENANT_TYPE_ACCOUNT, roles_to_attach=[ Role(role="bad"), Role(role="non existent"), Role(role="songwhip_read"), ], roles_to_detach=[], authenticated_identity_uuid="authenticated uuid", identity_ddb_connector=mock_identity_ddb_connector, principal=principal, ) assert result == IdentityTenant( identity_uuid=str(identity_uuid), tenant_type="account", tenant_uuid=tenant_1_uuid, roles=[{"role": "audience_development_analyst"}], version="24", ) mock_get_tenant_perms.assert_called_with(tenant_1_uuid) mock_get_empty_tenant_perms.assert_not_called() mock_update_tenant_perms.assert_called_once_with( tenant_1_uuid, IdentityTenant( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[ Role(role="audience_development_analyst"), Role(role="songwhip_read"), ], version="24", updated_at="2022-01-01T00:00:00+00:00", updated_by=expected_principal_identity_id, updated_impersonated_by=expected_impersonated_by, ), ) with caplog.at_level(logging.WARNING): assert ( caplog.records[0].msg == "Unsupported roles found in the list, discarding them: ['bad', 'non existent']" # noqa: E501 ) @freezegun.freeze_time("2022-01-01") @patch("pdp.models.identity.Identity.get_permissions_by_tenant") @patch("pdp.models.identity.Identity.get_empty_tenant_permissions_state") @patch("pdp.models.identity.Identity.update_tenant_permissions") async def test_attach_and_detach_roles_keeps_for_supported_roles_wo_principal( mock_update_tenant_perms: MagicMock, mock_get_empty_tenant_perms: MagicMock, mock_get_tenant_perms: MagicMock, tenant_1_uuid: uuid.UUID, mock_identity_ddb_connector: MagicMock, caplog: Any, ) -> None: """Test attach_and_detach_roles attaches only supported roles (wo principal arg).""" identity_uuid = uuid.uuid4() mock_get_tenant_perms.return_value = IdentityTenant( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[ {"role": "audience_development_analyst"}, ], version="23", ) mock_update_tenant_perms.return_value = IdentityTenant( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[{"role": "audience_development_analyst"}], version="24", ) result = await attach_and_detach_roles( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type=TenantType.TENANT_TYPE_ACCOUNT, roles_to_attach=[ Role(role="bad"), Role(role="non existent"), Role(role="songwhip_read"), ], roles_to_detach=[], authenticated_identity_uuid="authenticated uuid", identity_ddb_connector=mock_identity_ddb_connector, ) assert result == IdentityTenant( identity_uuid=str(identity_uuid), tenant_type="account", tenant_uuid=tenant_1_uuid, roles=[{"role": "audience_development_analyst"}], version="24", ) mock_get_tenant_perms.assert_called_with(tenant_1_uuid) mock_get_empty_tenant_perms.assert_not_called() mock_update_tenant_perms.assert_called_once_with( tenant_1_uuid, IdentityTenant( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[ Role(role="audience_development_analyst"), Role(role="songwhip_read"), ], version="24", updated_at="2022-01-01T00:00:00+00:00", updated_by="authenticated uuid", ), ) with caplog.at_level(logging.WARNING): assert ( caplog.records[0].msg == "Unsupported roles found in the list, discarding them: ['bad', 'non existent']" # noqa: E501 ) @freezegun.freeze_time("2022-01-01") @pytest.mark.parametrize( "principal,", [ pytest.param( Principal( identity_uuid="5fecc5ca-f7b3-44d1-9121-03588e809454", pdp_tenant_roles={}, ), id="principal is not impersonated", ), pytest.param( Principal( identity_uuid="5fecc5ca-f7b3-44d1-9121-03588e809454", pdp_tenant_roles={}, impersonated_by_identity_uuid="d8313f52-9c60-4782-8dd6-7c5c01ccb979", ), id="principal is impersonated", ), ], ) @patch("pdp.models.identity.Identity.get_permissions_by_tenant") @patch("pdp.models.identity.Identity.get_empty_tenant_permissions_state") @patch("pdp.models.identity.Identity.update_tenant_permissions") async def test_attach_and_detach_roles_no_changes( mock_update_tenant_perms: MagicMock, mock_get_empty_tenant_perms: MagicMock, mock_get_tenant_perms: MagicMock, principal: Principal, tenant_1_uuid: uuid.UUID, mock_identity_ddb_connector: MagicMock, ) -> None: """Test attach_and_detach_roles does not commit when no changes to permission.""" identity_uuid = uuid.uuid4() mock_get_tenant_perms.return_value = IdentityTenant( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[{"role": "songwhip_read"}, {"role": "audience_development_analyst"}], version="23", ) mock_update_tenant_perms.return_value = IdentityTenant( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[{"role": "songwhip_read"}, {"role": "audience_development_analyst"}], version="23", ) await attach_and_detach_roles( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type=TenantType.TENANT_TYPE_ACCOUNT, roles_to_attach=[Role(role="audience_development_analyst")], roles_to_detach=[], authenticated_identity_uuid="authenticated uuid", identity_ddb_connector=mock_identity_ddb_connector, principal=principal, ) mock_get_tenant_perms.assert_called_with(tenant_1_uuid) mock_get_empty_tenant_perms.assert_not_called() mock_update_tenant_perms.assert_not_called() @freezegun.freeze_time("2022-01-01") @patch("pdp.models.identity.Identity.get_permissions_by_tenant") @patch("pdp.models.identity.Identity.get_empty_tenant_permissions_state") @patch("pdp.models.identity.Identity.update_tenant_permissions") async def test_attach_and_detach_roles_no_changes_wo_principal( mock_update_tenant_perms: MagicMock, mock_get_empty_tenant_perms: MagicMock, mock_get_tenant_perms: MagicMock, tenant_1_uuid: uuid.UUID, mock_identity_ddb_connector: MagicMock, ) -> None: """Test attach_and_detach_roles does not commit when no changes to permission (wo principal arg).""" # noqa: E501 identity_uuid = uuid.uuid4() mock_get_tenant_perms.return_value = IdentityTenant( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[{"role": "songwhip_read"}, {"role": "audience_development_analyst"}], version="23", ) mock_update_tenant_perms.return_value = IdentityTenant( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[{"role": "songwhip_read"}, {"role": "audience_development_analyst"}], version="23", ) await attach_and_detach_roles( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type=TenantType.TENANT_TYPE_ACCOUNT, roles_to_attach=[Role(role="audience_development_analyst")], roles_to_detach=[], authenticated_identity_uuid="authenticated uuid", identity_ddb_connector=mock_identity_ddb_connector, ) mock_get_tenant_perms.assert_called_with(tenant_1_uuid) mock_get_empty_tenant_perms.assert_not_called() mock_update_tenant_perms.assert_not_called() @freezegun.freeze_time("2022-01-01") @pytest.mark.parametrize( "principal", [ pytest.param( Principal( identity_uuid="5fecc5ca-f7b3-44d1-9121-03588e809454", pdp_tenant_roles={}, ), id="principal is not impersonated", ), pytest.param( Principal( identity_uuid="5fecc5ca-f7b3-44d1-9121-03588e809454", pdp_tenant_roles={}, impersonated_by_identity_uuid="d8313f52-9c60-4782-8dd6-7c5c01ccb979", ), id="principal is impersonated", ), ], ) @patch("pdp.models.identity.Identity.get_permissions_by_tenant") @patch("pdp.models.identity.Identity.get_empty_tenant_permissions_state") @patch("pdp.models.identity.Identity.update_tenant_permissions") async def test_attach_and_detach_roles_update_errors_on_tenant_type( mock_update_tenant_perms: MagicMock, mock_get_empty_tenant_perms: MagicMock, mock_get_tenant_perms: MagicMock, principal: Principal, tenant_1_uuid: uuid.UUID, mock_identity_ddb_connector: MagicMock, ) -> None: """Test attach_and_detach_roles errors when the tenant_type does not match.""" identity_uuid = uuid.uuid4() mock_get_tenant_perms.return_value = IdentityTenant( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[], version="23", ) with pytest.raises(HTTPException): await attach_and_detach_roles( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type=TenantType("company_brand"), roles_to_attach=[], roles_to_detach=[], authenticated_identity_uuid="authenticated uuid", identity_ddb_connector=mock_identity_ddb_connector, principal=principal, ) mock_get_tenant_perms.assert_called_with(tenant_1_uuid) mock_get_empty_tenant_perms.assert_not_called() mock_update_tenant_perms.assert_not_called() @freezegun.freeze_time("2022-01-01") @patch("pdp.models.identity.Identity.get_permissions_by_tenant") @patch("pdp.models.identity.Identity.get_empty_tenant_permissions_state") @patch("pdp.models.identity.Identity.update_tenant_permissions") async def test_attach_and_detach_roles_update_errors_on_tenant_type_wo_principal( mock_update_tenant_perms: MagicMock, mock_get_empty_tenant_perms: MagicMock, mock_get_tenant_perms: MagicMock, tenant_1_uuid: uuid.UUID, mock_identity_ddb_connector: MagicMock, ) -> None: """Test attach_and_detach_roles errors when the tenant_type does not match (wo principal arg).""" # noqa: E501 identity_uuid = uuid.uuid4() mock_get_tenant_perms.return_value = IdentityTenant( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[], version="23", ) with pytest.raises(HTTPException): await attach_and_detach_roles( identity_uuid=str(identity_uuid), tenant_uuid=tenant_1_uuid, tenant_type=TenantType("company_brand"), roles_to_attach=[], roles_to_detach=[], authenticated_identity_uuid="authenticated uuid", identity_ddb_connector=mock_identity_ddb_connector, ) mock_get_tenant_perms.assert_called_with(tenant_1_uuid) mock_get_empty_tenant_perms.assert_not_called() mock_update_tenant_perms.assert_not_called() @pytest.mark.parametrize( "description, request_body, check_response, expected, expected_cerbos_check_resource_request", # noqa: E501 [ ( "Request with an empty resource type actions list should return a response with an empty list.", # noqa E501 CheckResourceTypeActionsRequest( resource_type_actions=[], ), CheckResourcesResponse( request_id="123", resources=[], ), CheckResourceTypeActionsResponse(resource_type_actions=[]), None, ), ( "A resource type action with a deny effect from cerbos should return a response with a deny effect.", # noqa E501 CheckResourceTypeActionsRequest( resource_type_actions=[{"resource_type": "audience", "action": "view"}], ), CheckResourcesResponse( request_id="123", resources=[ CheckResourceActionResult( resource=Resource( resource_id="0", resource_type="audience", attributes={} ), action="view", effect=AuthEffect.AUTH_EFFECT_DENY, errors={"validation_errors": None}, ) ], ), CheckResourceTypeActionsResponse( resource_type_actions=[ CheckResourceTypeActionResult( resource_type="audience", action="view", effect=AuthEffect.AUTH_EFFECT_DENY, errors={}, ) ] ), CheckResourcesRequest( resources=[ CheckResourceAction( resource=Resource( resource_type="audience", resource_id="0", attributes={ "tenant": { "tenant_type": "subaccount", "tenant_uuid": "04b48f72-5b47-425f-8b49-21f1ebc3f0cd", # noqa: E501 }, "identity_uuid": "ed925587-8aca-4de7-afa4-e83fd69225cb", }, ), action="view", ), ] ), ), ( "The corresponding allow/deny effect from cerbos for a resource type action should be returned in the response.", # noqa E501 CheckResourceTypeActionsRequest( resource_type_actions=[ {"resource_type": "audience", "action": "view"}, {"resource_type": "fan_data_list", "action": "edit"}, ], ), CheckResourcesResponse( request_id="123", resources=[ CheckResourceActionResult( resource=Resource( resource_id="0", resource_type="audience", attributes={} ), action="view", effect=AuthEffect.AUTH_EFFECT_ALLOW, errors={"validation_errors": None}, ), CheckResourceActionResult( resource=Resource( resource_id="0", resource_type="fan_data_list", attributes={}, ), action="edit", effect=AuthEffect.AUTH_EFFECT_DENY, errors={"validation_errors": None}, ), ], ), CheckResourceTypeActionsResponse( resource_type_actions=[ CheckResourceTypeActionResult( resource_type="audience", action="view", effect=AuthEffect.AUTH_EFFECT_ALLOW, errors={}, ), CheckResourceTypeActionResult( resource_type="fan_data_list", action="edit", effect=AuthEffect.AUTH_EFFECT_DENY, errors={}, ), ] ), CheckResourcesRequest( resources=[ CheckResourceAction( resource=Resource( resource_type="audience", resource_id="0", attributes={ "tenant": { "tenant_type": "subaccount", "tenant_uuid": "04b48f72-5b47-425f-8b49-21f1ebc3f0cd", # noqa: E501 }, "identity_uuid": "ed925587-8aca-4de7-afa4-e83fd69225cb", }, ), action="view", ), CheckResourceAction( resource=Resource( resource_type="fan_data_list", resource_id="0", attributes={ "tenant": { "tenant_type": "subaccount", "tenant_uuid": "04b48f72-5b47-425f-8b49-21f1ebc3f0cd", # noqa: E501 }, "identity_uuid": "ed925587-8aca-4de7-afa4-e83fd69225cb", }, ), action="edit", ), ] ), ), ], ) @patch("pdp.logic.identity.cerbos.check_resources") async def test_check_resource_type_actions_no_principal( mock_check_resources: AsyncMock, mock_pdp_tenant_roles: MagicMock, mock_splitio_client: MagicMock, mock_async_cerbos_client: MagicMock, mock_ows_account_client: OwsAccountClient, mock_ows_participant_client: OwsParticipantClient, mock_redis_connector: RedisConnector, description: str, request_body: CheckResourceTypeActionsRequest, check_response: CheckResourcesResponse, expected: CheckResourceTypeActionsResponse, expected_cerbos_check_resource_request: CheckResourcesRequest, ) -> None: """Test check_resource_type_actions does not pass Principal when not provided. Delete this test when consolidating to always pass Principal obj, likely as part of tearing down the pp_send_impersonated_by_identity_uuid flag. """ identity_uuid = "ed925587-8aca-4de7-afa4-e83fd69225cb" mock_check_resources.return_value = check_response response = await check_resource_type_actions( request_body, str(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, ) assert response == expected, description if not expected_cerbos_check_resource_request: mock_check_resources.assert_not_called() else: mock_check_resources.assert_called_once_with( str(identity_uuid), check_resources_request=expected_cerbos_check_resource_request, 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=None, principal=None, splitio_client=mock_splitio_client, ) @pytest.mark.parametrize( "description, request_body, check_response, expected, expected_cerbos_check_resource_request", # noqa: E501 [ ( "Request with an empty resource type actions list should return a response with an empty list.", # noqa E501 CheckResourceTypeActionsRequest( resource_type_actions=[], ), CheckResourcesResponse( request_id="123", resources=[], ), CheckResourceTypeActionsResponse(resource_type_actions=[]), None, ), ( "A resource type action with a deny effect from cerbos should return a response with a deny effect.", # noqa E501 CheckResourceTypeActionsRequest( resource_type_actions=[{"resource_type": "audience", "action": "view"}], ), CheckResourcesResponse( request_id="123", resources=[ CheckResourceActionResult( resource=Resource( resource_id="0", resource_type="audience", attributes={} ), action="view", effect=AuthEffect.AUTH_EFFECT_DENY, errors={"validation_errors": None}, ) ], ), CheckResourceTypeActionsResponse( resource_type_actions=[ CheckResourceTypeActionResult( resource_type="audience", action="view", effect=AuthEffect.AUTH_EFFECT_DENY, errors={}, ) ] ), CheckResourcesRequest( resources=[ CheckResourceAction( resource=Resource( resource_type="audience", resource_id="0", attributes={ "tenant": { "tenant_type": "subaccount", "tenant_uuid": "04b48f72-5b47-425f-8b49-21f1ebc3f0cd", # noqa: E501 }, "identity_uuid": "ed925587-8aca-4de7-afa4-e83fd69225cb", }, ), action="view", ), ] ), ), ( "The corresponding allow/deny effect from cerbos for a resource type action should be returned in the response.", # noqa E501 CheckResourceTypeActionsRequest( resource_type_actions=[ {"resource_type": "audience", "action": "view"}, {"resource_type": "fan_data_list", "action": "edit"}, ], ), CheckResourcesResponse( request_id="123", resources=[ CheckResourceActionResult( resource=Resource( resource_id="0", resource_type="audience", attributes={} ), action="view", effect=AuthEffect.AUTH_EFFECT_ALLOW, errors={"validation_errors": None}, ), CheckResourceActionResult( resource=Resource( resource_id="0", resource_type="fan_data_list", attributes={}, ), action="edit", effect=AuthEffect.AUTH_EFFECT_DENY, errors={"validation_errors": None}, ), ], ), CheckResourceTypeActionsResponse( resource_type_actions=[ CheckResourceTypeActionResult( resource_type="audience", action="view", effect=AuthEffect.AUTH_EFFECT_ALLOW, errors={}, ), CheckResourceTypeActionResult( resource_type="fan_data_list", action="edit", effect=AuthEffect.AUTH_EFFECT_DENY, errors={}, ), ] ), CheckResourcesRequest( resources=[ CheckResourceAction( resource=Resource( resource_type="audience", resource_id="0", attributes={ "tenant": { "tenant_type": "subaccount", "tenant_uuid": "04b48f72-5b47-425f-8b49-21f1ebc3f0cd", # noqa: E501 }, "identity_uuid": "ed925587-8aca-4de7-afa4-e83fd69225cb", }, ), action="view", ), CheckResourceAction( resource=Resource( resource_type="fan_data_list", resource_id="0", attributes={ "tenant": { "tenant_type": "subaccount", "tenant_uuid": "04b48f72-5b47-425f-8b49-21f1ebc3f0cd", # noqa: E501 }, "identity_uuid": "ed925587-8aca-4de7-afa4-e83fd69225cb", }, ), action="edit", ), ] ), ), ], ) @patch("pdp.logic.identity.cerbos.check_resources") async def test_check_resource_type_actions( mock_check_resources: AsyncMock, mock_pdp_tenant_roles: MagicMock, mock_splitio_client: MagicMock, mock_async_cerbos_client: MagicMock, mock_ows_account_client: OwsAccountClient, mock_ows_participant_client: OwsParticipantClient, mock_redis_connector: RedisConnector, description: str, request_body: CheckResourceTypeActionsRequest, check_response: CheckResourcesResponse, expected: CheckResourceTypeActionsResponse, expected_cerbos_check_resource_request: CheckResourcesRequest, ) -> None: """Test check_resource_type_actions.""" identity_uuid = "ed925587-8aca-4de7-afa4-e83fd69225cb" mock_check_resources.return_value = check_response mock_principal = MagicMock(spec=Principal) response = await check_resource_type_actions( request_body, str(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, principal=mock_principal, ) assert response == expected, description if not expected_cerbos_check_resource_request: mock_check_resources.assert_not_called() else: mock_check_resources.assert_called_once_with( str(identity_uuid), check_resources_request=expected_cerbos_check_resource_request, 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=None, principal=mock_principal, splitio_client=mock_splitio_client, ) @patch("pdp.logic.identity.cerbos.check_resources") async def test_check_resource_type_actions_empty_pdp_tenant_roles( mock_check_resources: MagicMock, mock_async_cerbos_client: MagicMock, mock_ows_account_client: OwsAccountClient, mock_ows_participant_client: OwsParticipantClient, mock_redis_connector: RedisConnector, mock_splitio_client: MagicMock, ) -> None: """Test check_resource_type_actions returns DENY when identity has no roles.""" request_id = "test-cerbos-request-id" mock_principal = MagicMock(spec=Principal) mock_check_resources.return_value = CheckResourcesResponse( request_id=request_id, resources=[ CheckResourceActionResult( resource=Resource(resource_id=0, resource_type="audience"), action="view", effect=AuthEffect.AUTH_EFFECT_DENY, ), CheckResourceActionResult( resource=Resource(resource_id=0, resource_type="fan_data_list"), action="edit", effect=AuthEffect.AUTH_EFFECT_ALLOW, ), ], ) request_body = CheckResourceTypeActionsRequest( resource_type_actions=[ {"resource_type": "audience", "action": "view"}, {"resource_type": "fan_data_list", "action": "edit"}, ], ) identity_uuid = str(uuid.uuid4()) resp = await check_resource_type_actions( request_body, identity_uuid, 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, principal=mock_principal, ) assert resp == CheckResourceTypeActionsResponse( resource_type_actions=[ CheckResourceTypeActionResult( resource_type="audience", action="view", effect=AuthEffect.AUTH_EFFECT_DENY, errors={}, ), CheckResourceTypeActionResult( resource_type="fan_data_list", action="edit", effect=AuthEffect.AUTH_EFFECT_ALLOW, errors={}, ), ] ) mock_check_resources.assert_called_once_with( identity_uuid, check_resources_request=CheckResourcesRequest( resources=[ CheckResourceAction( resource=Resource( resource_id="0", resource_type="audience", attributes={ "identity_uuid": identity_uuid, }, ), action="view", ), CheckResourceAction( resource=Resource( resource_id="0", resource_type="fan_data_list", attributes={ "identity_uuid": identity_uuid, }, ), action="edit", ), ], include_resource_attributes_in_response=False, ), 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=None, principal=mock_principal, splitio_client=mock_splitio_client, ) @patch.object( CheckResourceTypeAction, "create_check_resource_actions", ) async def test_check_resource_type_actions_create_resource_actions_error( mock_create_check_resource_actions: MagicMock, mock_async_cerbos_client: MagicMock, mock_ows_account_client: OwsAccountClient, mock_ows_participant_client: OwsParticipantClient, mock_redis_connector: RedisConnector, mock_splitio_client: MagicMock, caplog: pytest.LogCaptureFixture, ) -> None: """Test check_resource_type_actions handles CreateCheckResourceActionsError.""" identity_uuid = str(uuid.uuid4()) request_body = CheckResourceTypeActionsRequest( resource_type_actions=[ CheckResourceTypeAction(**{"resource_type": "audience", "action": "view"}), CheckResourceTypeAction( **{"resource_type": "fan_data_list", "action": "edit"} ), ], ) mock_create_check_resource_actions.side_effect = [ CreateCheckResourceActionsError("test_error") ] resp = await check_resource_type_actions( request_body, identity_uuid, 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, ) assert resp == CheckResourceTypeActionsResponse( resource_type_actions=[ CheckResourceTypeActionResult( resource_type="audience", action="view", effect=AuthEffect.AUTH_EFFECT_DENY, errors={}, ), CheckResourceTypeActionResult( resource_type="fan_data_list", action="edit", effect=AuthEffect.AUTH_EFFECT_DENY, errors={}, ), ] ) with caplog.at_level(logging.ERROR): assert ( caplog.records[0].message == "create_check_resource_actions error: 'test_error'" ) assert hasattr(caplog.records[0], "pdp") assert caplog.records[0].pdp == { "authenticated_identity_uuid": "None", "identity_uuid": identity_uuid, "resource_type_actions": b'[{"resource_type":"audience","action":"view"},{"resource_type":"fan_data_list","action":"edit"}]', # noqa: E501 } @pytest.mark.parametrize( "principal", [ pytest.param(None, id="No Principal is passed along"), pytest.param( Principal( identity_uuid=uuid.UUID("5e135c46-eacf-42fb-a590-692120e5b324"), user_type=USER_TYPE_MACHINE, pdp_tenant_roles={}, impersonated_by_identity_uuid=uuid.UUID( "a8ee6ee9-093c-4fdb-977c-b096d19a7515" ), ), id="Valid Principal is passed along", ), ], ) @patch("pdp.logic.cerbos.check_resources") @patch.object( CheckResourcesResponse, "filter_for_allowed_tenants", ) @patch.object( GetAllowedTenantsRequest, "create_check_resource_actions", ) async def test_get_allowed_tenants( mock_create_check_resource_actions: MagicMock, mock_filter_for_allowed_tenants: MagicMock, mock_check_resources: AsyncMock, principal: Principal, mock_pdp_tenant_roles: MagicMock, mock_async_cerbos_client: MagicMock, mock_ows_account_client: OwsAccountClient, mock_ows_participant_client: OwsParticipantClient, mock_redis_connector: RedisConnector, mock_splitio_client: SplitioClient, ) -> None: """Test get_allowed_tenants.""" identity_uuid = uuid.uuid4() request_body = GetAllowedTenantsRequest( action="view", resource_type="audience", ) mock_check_resources_list = [ CheckResourceAction( resource=Resource( resource_id=0, resource_type="audience", attributes={ "tenant": { "tenant_uuid": "04b48f72-5b47-425f-8b49-21f1ebc3f0cd", "tenant_type": "subaccount", } }, ), action="view", ) ] mock_create_check_resource_actions.return_value = mock_check_resources_list mock_check_resources_response = CheckResourcesResponse( request_id="123", resources=[ CheckResourceActionResult( resource=Resource( resource_id="0", resource_type="audience", attributes={ "tenant": { "tenant_type": "subaccount", "tenant_uuid": "04b48f72-5b47-425f-8b49-21f1ebc3f0cd", }, }, ), action="view", effect=AuthEffect.AUTH_EFFECT_ALLOW, errors={"validation_errors": None}, ) ], ) mock_check_resources.return_value = mock_check_resources_response mock_filter_for_allowed_tenants.return_value = [ AllowedTenant( tenant_type="subaccount", tenant_uuid="04b48f72-5b47-425f-8b49-21f1ebc3f0cd", ) ] if principal: # Params include `principal` response = await get_allowed_tenants( request_body, str(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, splitio_client=mock_splitio_client, principal=principal, ) else: # No `principal` param. Should use the `None` default response = await get_allowed_tenants( request_body, str(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, splitio_client=mock_splitio_client, ) assert response == GetAllowedTenantsResponse( action="view", resource_type="audience", tenants=[ { "tenant_type": "subaccount", "tenant_uuid": "04b48f72-5b47-425f-8b49-21f1ebc3f0cd", } ], ) mock_create_check_resource_actions.assert_called_once_with( mock_pdp_tenant_roles, ) mock_check_resources.assert_called_once_with( str(identity_uuid), check_resources_request=CheckResourcesRequest( resources=mock_check_resources_list, ), 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, include_resource_attributes=True, principal=principal, splitio_client=mock_splitio_client, ) mock_filter_for_allowed_tenants.assert_called_once() @patch("pdp.logic.cerbos.check_resources") @patch.object( CheckResourcesResponse, "filter_for_allowed_tenants", ) @patch.object( GetAllowedTenantsRequest, "create_check_resource_actions", ) async def test_get_allowed_tenants_empty_tenant_roles( mock_create_check_resource_actions: MagicMock, mock_filter_for_allowed_tenants: MagicMock, mock_check_resources: AsyncMock, mock_async_cerbos_client: MagicMock, mock_ows_account_client: OwsAccountClient, mock_ows_participant_client: OwsParticipantClient, mock_redis_connector: RedisConnector, mock_splitio_client: SplitioClient, caplog: pytest.LogCaptureFixture, ) -> None: """Test get_allowed_tenants when pdp_tenant_roles are empty.""" identity_uuid = uuid.uuid4() mock_create_check_resource_actions.side_effect = CreateCheckResourceActionsError( "test_error" ) request_body = GetAllowedTenantsRequest( action="view", resource_type="audience", ) response = await get_allowed_tenants( request_body, str(identity_uuid), 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, splitio_client=mock_splitio_client, ) assert response == GetAllowedTenantsResponse( action="view", resource_type="audience", tenants=[] ) mock_check_resources.assert_not_called() mock_filter_for_allowed_tenants.assert_not_called() mock_create_check_resource_actions.assert_called_with( {}, ) with caplog.at_level(logging.ERROR): assert ( caplog.records[0].message == "create_check_resource_actions error: 'test_error'" ) assert hasattr(caplog.records[0], "pdp") assert caplog.records[0].pdp == { "authenticated_identity_uuid": str(identity_uuid), "identity_uuid": str(identity_uuid), "resource_type_actions": { "resource_type": "audience", "action": "view", }, "function": "get_allowed_tenants", } @pytest.mark.parametrize( "deactivate_many_model_return, expected", [ pytest.param( [], DeactivationSummary(deleted=2, remaining=0), id="both tenants were deleted", ), pytest.param( [ { "identity_uuid": "123", "tenant_uuid": "4378cd91-7852-49eb-b39f-653dec222db8", } ], DeactivationSummary(deleted=1, remaining=1), id="one unprocessed tenant remaining", ), pytest.param( [ { "identity_uuid": "123", "tenant_uuid": "a9dd9b42-e53d-11ee-be6d-4a2888760682", }, { "identity_uuid": "123", "tenant_uuid": "4378cd91-7852-49eb-b39f-653dec222db8", }, ], DeactivationSummary(deleted=0, remaining=2), id="both tenants were unprocessed.", ), ], ) @pytest.mark.parametrize( "principal", [ pytest.param( Principal( identity_uuid="5fecc5ca-f7b3-44d1-9121-03588e809454", pdp_tenant_roles={}, ), id="principal is None", ), pytest.param( Principal( identity_uuid="5fecc5ca-f7b3-44d1-9121-03588e809454", pdp_tenant_roles={}, ), id="principal is not impersonated", ), pytest.param( Principal( identity_uuid="5fecc5ca-f7b3-44d1-9121-03588e809454", pdp_tenant_roles={}, impersonated_by_identity_uuid="d8313f52-9c60-4782-8dd6-7c5c01ccb979", ), id="principal is impersonated", ), ], ) @patch( "pdp.models.identity.Identity.write_tombstone_records_for_deactivations", new_callable=AsyncMock, ) @patch("pdp.models.identity.Identity.deactivate_many") async def test_deactivate_many( mock_deactivate_many: MagicMock, mock_write_tombstone_records_for_deactivations: AsyncMock, mock_identity_ddb_connector: MagicMock, principal: Principal | None, tenant_1_uuid: uuid.UUID, tenant_2_uuid: uuid.UUID, deactivate_many_model_return: List[Dict[str, str]], expected: DeactivationSummary, ) -> None: """Test deactivate_many.""" identity_uuid = "123" authenticated_identity_uuid = str(uuid.uuid4()) tenants = { tenant_1_uuid: TenantRoles( tenant_uuid=tenant_1_uuid, tenant_type="account", roles=[], ), tenant_2_uuid: TenantRoles( tenant_uuid=tenant_2_uuid, tenant_type="subaccount", roles=[], ), } mock_deactivate_many.return_value = deactivate_many_model_return mock_write_tombstone_records_for_deactivations.return_value = [] result = await deactivate_many( identity_uuid=identity_uuid, tenants=tenants, identity_ddb_connector=mock_identity_ddb_connector, authenticated_identity_uuid=authenticated_identity_uuid, principal=principal, ) assert result == expected mock_deactivate_many.assert_called_once_with(tenants) mock_write_tombstone_records_for_deactivations.assert_called_once_with( authenticated_identity_uuid=authenticated_identity_uuid, request_tenants=tenants, unprocessed_items=deactivate_many_model_return, principal=principal, ) @pytest.mark.parametrize( "roles_to_detach, expected", [ pytest.param( [], [], id="An empty list should stay empty", ), pytest.param( [Role(role="fansifter_can_create_ad_reports")], [Role(role="fansifter_can_create_ad_reports")], id="""A child role should be detached without adding additional roles to detach""", ), pytest.param( [Role(role="songwhip_read")], [Role(role="songwhip_read")], id="""A role without parent/child relationship should be detached without adding additional roles to detach""", ), pytest.param( [ Role(role="songwhip_read"), Role(role="fansifter_can_view_fan_data"), ], [ Role(role="songwhip_read"), Role(role="fansifter_can_view_fan_data"), Role(role="fansifter_can_create_ad_reports"), Role(role="fansifter_can_share_ad_campaign_audiences"), Role(role="fansifter_can_connect_ad_accounts"), Role(role="fansifter_can_view_email_campaigns"), Role(role="fansifter_can_create_email_campaigns"), Role(role="fansifter_can_view_sms_campaigns"), Role(role="fansifter_can_create_sms_campaigns"), ], id="A parent role should result in child roles being detached", ), pytest.param( [ Role(role="songwhip_read"), Role(role="fansifter_can_view_fan_data"), Role(role="fansifter_can_create_ad_reports"), ], [ Role(role="songwhip_read"), Role(role="fansifter_can_view_fan_data"), Role(role="fansifter_can_create_ad_reports"), Role(role="fansifter_can_create_ad_reports"), Role(role="fansifter_can_share_ad_campaign_audiences"), Role(role="fansifter_can_connect_ad_accounts"), Role(role="fansifter_can_view_email_campaigns"), Role(role="fansifter_can_create_email_campaigns"), Role(role="fansifter_can_view_sms_campaigns"), Role(role="fansifter_can_create_sms_campaigns"), ], id="Duplicates can occur in extending with child roles to detach", ), ], ) def test__extend_with_child_roles_to_detach( roles_to_detach: List[Role], expected: List[Role], ) -> None: """Test _extend_with_child_roles_to_detach.""" actual = _extend_with_child_roles_to_detach(roles_to_detach) assert actual == expected @pytest.mark.parametrize( "roles, expected", [ pytest.param([], [], id="An empty list is still an empty list"), pytest.param([Role(role="dog")], [], id="all roles may be removed"), pytest.param( [ Role(role="dog"), Role(role="audience_development_admin"), Role(role="fansifter_can_view_fan_data"), ], [ Role(role="audience_development_admin"), Role(role="fansifter_can_view_fan_data"), ], id="known supported roles are kept", ), pytest.param( [ Role(role="dog"), Role(role="audience_development_admin"), Role(role="fansifter_can_view_fan_data"), Role(role="audience_development_admin"), ], [ Role(role="audience_development_admin"), Role(role="fansifter_can_view_fan_data"), Role(role="audience_development_admin"), ], id="duplicate known supported roles are kept", ), ], ) def test__keep_supported_roles(roles: List[Role], expected: List[Role]) -> None: """Test _keep_supported_roles.""" actual = _keep_supported_roles(roles) assert actual == expected