"""Test Identity Model.""" import logging import uuid from typing import Any, Dict, List from unittest.mock import AsyncMock, MagicMock, patch from uuid import UUID, uuid4 import freezegun import pytest from pdp import config from pdp.constants.constants import TenantType from pdp.fastapi.schemas.identity import ( IdentityTenant, TenantRoles, TombstoneIdentityTenant, TombstoneIdentityTenantValidator, ) from pdp.fastapi.schemas.principal import Principal from pdp.fastapi.schemas.tenant import Tenant from pdp.models.identity import Identity from tests.unit.conftest import factory_dynamo_identity_role def test_get_roles(mock_identity_ddb_connector: MagicMock) -> None: """Test get roles.""" mock_identity_ddb_connector.query_by_hash_key.return_value = { "items": [{"role": "everything"}], "cursor": {"cursor": None}, } identity = Identity("my id", identity_ddb_connector=mock_identity_ddb_connector) result = identity.get_roles() mock_identity_ddb_connector.query_by_hash_key.assert_called_once_with( "my id", cursor=None ) assert result == {"items": [{"role": "everything"}], "cursor": {"cursor": None}} def test_get_roles_uses_cursor(mock_identity_ddb_connector: MagicMock) -> None: """Test get roles.""" mock_identity_ddb_connector.query_by_hash_key.return_value = { "items": [{"role": "everything"}], "cursor": {"cursor": None}, } identity = Identity("my id", identity_ddb_connector=mock_identity_ddb_connector) result = identity.get_roles(cursor="hello") mock_identity_ddb_connector.query_by_hash_key.assert_called_once_with( "my id", cursor="hello" ) assert result == {"items": [{"role": "everything"}], "cursor": {"cursor": None}} def test_update_roles( tenant_1_uuid: UUID, mock_identity_ddb_connector: MagicMock ) -> None: """Test update roles.""" mock_identity_ddb_connector.update_item.return_value = { "Attributes": {"roles": ["many roles"]} } my_id = str(uuid4()) identity = Identity(my_id, identity_ddb_connector=mock_identity_ddb_connector) result = identity.update_roles(tenant_1_uuid, {"roles": ["many roles"]}) mock_identity_ddb_connector.update_item.assert_called_with( hash_key=my_id, item={"roles": ["many roles"]}, range_key=str(tenant_1_uuid) ) assert result == {"Attributes": {"roles": ["many roles"]}} def test_get_empty_tenant_permissions_state( tenant_1_uuid: UUID, tenant_1_uuid_as_string: str, mock_identity_ddb_connector: MagicMock, ) -> None: """Test get_empty_tenant_permissions_state.""" identity = Identity("my id", identity_ddb_connector=mock_identity_ddb_connector) result = identity.get_empty_tenant_permissions_state( tenant_1_uuid, TenantType.TENANT_TYPE_ACCOUNT, "created at now", "created by me", ) assert result == IdentityTenant( **{ "created_at": "created at now", "created_by": "created by me", "identity_uuid": "my id", "roles": [], "tenant_type": "account", "tenant_uuid": tenant_1_uuid_as_string, "updated_at": "created at now", "updated_by": "created by me", "version": "0", } ) def test_get_empty_tenant_permissions_state_with_created_impersonated_by( tenant_1_uuid: UUID, tenant_1_uuid_as_string: str, mock_identity_ddb_connector: MagicMock, ) -> None: """Test get_empty_tenant_permissions_state with created_impersonated_by.""" identity = Identity("my id", identity_ddb_connector=mock_identity_ddb_connector) result = identity.get_empty_tenant_permissions_state( tenant_1_uuid, TenantType.TENANT_TYPE_ACCOUNT, "created at now", "created by me", "created impersonated by", ) assert result == IdentityTenant( **{ "created_at": "created at now", "created_impersonated_by": "created impersonated by", "created_by": "created by me", "identity_uuid": "my id", "roles": [], "tenant_type": "account", "tenant_uuid": tenant_1_uuid_as_string, "updated_at": "created at now", "updated_by": "created by me", "updated_impersonated_by": "created impersonated by", "version": "0", } ) def test_get_permissions_by_tenant( tenant_1_uuid: UUID, tenant_1_uuid_as_string: str, mock_identity_ddb_connector: MagicMock, ) -> None: """Test get_permissions_by_tenant.""" mock_identity_ddb_connector.query_by_hash_key.return_value = { "items": [ factory_dynamo_identity_role(tenant_uuid=tenant_1_uuid_as_string), ], "count": 1, } identity = Identity( "c5879365-2c07-4c19-a3f0-cc030c01b6a5", identity_ddb_connector=mock_identity_ddb_connector, ) result = identity.get_permissions_by_tenant(tenant_1_uuid) mock_identity_ddb_connector.query_by_hash_key.assert_called_once_with( hash_key="c5879365-2c07-4c19-a3f0-cc030c01b6a5", range_key=tenant_1_uuid_as_string, ) assert result == IdentityTenant( identity_uuid="c5879365-2c07-4c19-a3f0-cc030c01b6a5", tenant_uuid=tenant_1_uuid_as_string, tenant_type="account", roles=[{"role": "audience_development_admin"}], ) def test_get_permissions_by_tenant_no_items( tenant_1_uuid: UUID, tenant_1_uuid_as_string: str, mock_identity_ddb_connector: MagicMock, ) -> None: """Test get_permissions_by_tenant.""" mock_identity_ddb_connector.query_by_hash_key.return_value = { "items": [], "count": 1, } identity = Identity( "c5879365-2c07-4c19-a3f0-cc030c01b6a5", identity_ddb_connector=mock_identity_ddb_connector, ) result = identity.get_permissions_by_tenant(tenant_1_uuid) mock_identity_ddb_connector.query_by_hash_key.assert_called_once_with( hash_key="c5879365-2c07-4c19-a3f0-cc030c01b6a5", range_key=tenant_1_uuid_as_string, ) assert result is None def test_update_tenant_permissions( tenant_1_uuid: UUID, mock_identity_ddb_connector: MagicMock ) -> None: """Test update_tenant_permissions.""" mock_identity_ddb_connector.update_item.return_value = { "updated_at": {"S": "updated at then"}, "roles": {"L": [{"M": {"role": {"S": "beekeeper"}}}]}, "updated_by": {"S": "updated by you"}, "created_at": {"S": "created at now"}, "tenant_type": {"S": "account"}, "created_by": {"S": "created by me"}, "version": {"S": "1345"}, } identity = Identity( "c5879365-2c07-4c19-a3f0-cc030c01b6a5", identity_ddb_connector=mock_identity_ddb_connector, ) tenant_state = IdentityTenant( identity_uuid="c5879365-2c07-4c19-a3f0-cc030c01b6a5", tenant_uuid=tenant_1_uuid, tenant_type="account", created_at="created at now", created_by="created by me", version="1345", updated_at="updated at then", updated_by="updated by you", roles=[{"role": "beekeeper"}], ) result = identity.update_tenant_permissions(tenant_1_uuid, tenant_state) mock_identity_ddb_connector.update_item.assert_called_with( hash_key="c5879365-2c07-4c19-a3f0-cc030c01b6a5", item={ "created_at": {"S": "created at now"}, "created_by": {"S": "created by me"}, "created_impersonated_by": { "NULL": True, }, "roles": {"L": [{"M": {"role": {"S": "beekeeper"}}}]}, "tenant_type": {"S": "account"}, "updated_at": {"S": "updated at then"}, "updated_by": {"S": "updated by you"}, "updated_impersonated_by": { "NULL": True, }, "version": {"S": "1345"}, }, range_key=str(tenant_1_uuid), ) assert result == tenant_state async def test_deactivate_many_empty_dict( mock_identity_ddb_connector: AsyncMock, ) -> None: """Test deactivate_many when dict is empty.""" identity_uuid = "my id" identity = Identity( identity_uuid, identity_ddb_connector=mock_identity_ddb_connector ) mock_identity_ddb_connector.delete_items = AsyncMock(return_value=[]) result = await identity.deactivate_many({}) assert result == [] mock_identity_ddb_connector.delete_items.assert_not_called() async def test_deactivate_many( mock_identity_ddb_connector: AsyncMock, tenant_1_uuid: uuid.UUID, tenant_1_uuid_as_string: str, tenant_2_uuid: uuid.UUID, ) -> None: """Test deactivate_many.""" identity_uuid = "my id" identity = Identity( identity_uuid, identity_ddb_connector=mock_identity_ddb_connector ) 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_identity_ddb_connector.delete_items = AsyncMock( return_value=[ { "mock_hash_key": identity_uuid, "mock_range_key": tenant_1_uuid_as_string, } ] ) result = await identity.deactivate_many(tenants) assert result == [ { "mock_hash_key": identity_uuid, "mock_range_key": tenant_1_uuid_as_string, } ] mock_identity_ddb_connector.delete_items.assert_called_once_with( [ { "identity_uuid": identity_uuid, "tenant_uuid": tenant_1_uuid, }, { "identity_uuid": identity_uuid, "tenant_uuid": tenant_2_uuid, }, ], ) @freezegun.freeze_time("2022-01-01") def test_get_tombstone_state(mock_identity_ddb_connector: MagicMock) -> None: """Test Identity.get_tombstone_state.""" identity_uuid = "e3aff896-5619-4d0a-9181-49b333096146" tenant_uuid = "2856ba90-0ea3-4b27-82d8-2a48f5b7bcb0" ttl = 5 # >>> datetime.datetime(2022, 1, 1, tzinfo=datetime.UTC).timestamp() frozen_epoch_ts = 1640995200 identity = Identity( identity_uuid, identity_ddb_connector=mock_identity_ddb_connector, ) tombstone_state = identity.get_tombstone_state( Tenant(tenant_type=TenantType.TENANT_TYPE_ACCOUNT, tenant_uuid=tenant_uuid), deleted_at="2022-01-01T00:00:00+00:00", deleted_by="authenticated uuid", ttl=ttl, ) assert tombstone_state == TombstoneIdentityTenant( **{ "created_at": "2022-01-01T00:00:00+00:00", "created_by": "authenticated uuid", "created_impersonated_by": None, "expires_at": frozen_epoch_ts + ttl, "identity_uuid": "TOMBSTONE:e3aff896-5619-4d0a-9181-49b333096146:1640995200", # noqa: E501 "tenant_uuid": "TOMBSTONE:2856ba90-0ea3-4b27-82d8-2a48f5b7bcb0", "is_tombstone": True, "roles": [], "tenant_type": "account", "updated_at": "2022-01-01T00:00:00+00:00", "updated_by": "authenticated uuid", "updated_impersonated_by": None, "version": "0", } ) @freezegun.freeze_time("2022-01-01") def test_get_tombstone_state_with_deleted_impersonated_by( mock_identity_ddb_connector: MagicMock, ) -> None: """Test Identity.get_tombstone_state with deleted_impersonated_by.""" identity_uuid = "e3aff896-5619-4d0a-9181-49b333096146" tenant_uuid = "2856ba90-0ea3-4b27-82d8-2a48f5b7bcb0" ttl = 5 # >>> datetime.datetime(2022, 1, 1, tzinfo=datetime.UTC).timestamp() frozen_epoch_ts = 1640995200 identity = Identity( identity_uuid, identity_ddb_connector=mock_identity_ddb_connector, ) tombstone_state = identity.get_tombstone_state( Tenant(tenant_type=TenantType.TENANT_TYPE_ACCOUNT, tenant_uuid=tenant_uuid), deleted_at="2022-01-01T00:00:00+00:00", deleted_by="authenticated uuid", deleted_impersonated_by="impersonator uuid", ttl=ttl, ) assert tombstone_state == TombstoneIdentityTenant( **{ "created_at": "2022-01-01T00:00:00+00:00", "created_by": "authenticated uuid", "created_impersonated_by": "impersonator uuid", "expires_at": frozen_epoch_ts + ttl, "identity_uuid": "TOMBSTONE:e3aff896-5619-4d0a-9181-49b333096146:1640995200", # noqa: E501 "tenant_uuid": "TOMBSTONE:2856ba90-0ea3-4b27-82d8-2a48f5b7bcb0", "is_tombstone": True, "roles": [], "tenant_type": "account", "updated_at": "2022-01-01T00:00:00+00:00", "updated_by": "authenticated uuid", "updated_impersonated_by": "impersonator uuid", "version": "0", } ) @pytest.mark.parametrize( "request_tenants, unprocessed_items, expected", [ pytest.param( { uuid.UUID("86491fd9-518d-4087-bb74-046b2b8b99e5"): TenantRoles( tenant_type=TenantType.TENANT_TYPE_ACCOUNT, tenant_uuid=uuid.UUID("86491fd9-518d-4087-bb74-046b2b8b99e5"), roles=[], ), uuid.UUID("31120501-2d3f-4e51-8149-f4365a381995"): TenantRoles( tenant_type=TenantType.TENANT_TYPE_ACCOUNT, tenant_uuid=uuid.UUID("31120501-2d3f-4e51-8149-f4365a381995"), roles=[], ), }, [ { config.IDENTITY_HASH_KEY: "57c25895-ce10-483f-b187-44c24308f8da", config.IDENTITY_RANGE_KEY: uuid.UUID( "31120501-2d3f-4e51-8149-f4365a381995" ), }, ], [ TombstoneIdentityTenant( **{ "created_at": "2022-01-01T00:00:00+00:00", "created_by": "0c009413-1ee6-444f-a875-571298eca67e", "expires_at": 1640995201, "identity_uuid": "TOMBSTONE:57c25895-ce10-483f-b187-44c24308f8da:1640995200", # noqa: E501 "tenant_uuid": "TOMBSTONE:86491fd9-518d-4087-bb74-046b2b8b99e5", "is_tombstone": True, "roles": [], "tenant_type": "account", "updated_at": "2022-01-01T00:00:00+00:00", "updated_by": "0c009413-1ee6-444f-a875-571298eca67e", "version": "0", } ), ], id="The result should not include a tenant in unprocessed_items", ), pytest.param( { uuid.UUID("86491fd9-518d-4087-bb74-046b2b8b99e5"): TenantRoles( tenant_type=TenantType.TENANT_TYPE_ACCOUNT, tenant_uuid=uuid.UUID("86491fd9-518d-4087-bb74-046b2b8b99e5"), roles=[], ), uuid.UUID("31120501-2d3f-4e51-8149-f4365a381995"): TenantRoles( tenant_type=TenantType.TENANT_TYPE_ACCOUNT, tenant_uuid=uuid.UUID("31120501-2d3f-4e51-8149-f4365a381995"), roles=[], ), }, [ { config.IDENTITY_HASH_KEY: "57c25895-ce10-483f-b187-44c24308f8da", config.IDENTITY_RANGE_KEY: uuid.UUID( "31120501-2d3f-4e51-8149-f4365a381995" ), }, { config.IDENTITY_HASH_KEY: "57c25895-ce10-483f-b187-44c24308f8da", config.IDENTITY_RANGE_KEY: uuid.UUID( "31120501-2d3f-4e51-8149-f4365a381995" ), }, { config.IDENTITY_HASH_KEY: "57c25895-ce10-483f-b187-44c24308f8da", config.IDENTITY_RANGE_KEY: uuid.UUID( "31120501-2d3f-4e51-8149-f4365a381995" ), }, ], [ TombstoneIdentityTenant( **{ "created_at": "2022-01-01T00:00:00+00:00", "created_by": "0c009413-1ee6-444f-a875-571298eca67e", "expires_at": 1640995201, "identity_uuid": "TOMBSTONE:57c25895-ce10-483f-b187-44c24308f8da:1640995200", # noqa: E501 "tenant_uuid": "TOMBSTONE:86491fd9-518d-4087-bb74-046b2b8b99e5", "is_tombstone": True, "roles": [], "tenant_type": "account", "updated_at": "2022-01-01T00:00:00+00:00", "updated_by": "0c009413-1ee6-444f-a875-571298eca67e", "version": "0", } ), ], id="Handles when unprocessed_items contains duplicates", ), pytest.param( { uuid.UUID("86491fd9-518d-4087-bb74-046b2b8b99e5"): TenantRoles( tenant_type=TenantType.TENANT_TYPE_ACCOUNT, tenant_uuid=uuid.UUID("86491fd9-518d-4087-bb74-046b2b8b99e5"), roles=[], ), uuid.UUID("31120501-2d3f-4e51-8149-f4365a381995"): TenantRoles( tenant_type=TenantType.TENANT_TYPE_ACCOUNT, tenant_uuid=uuid.UUID("31120501-2d3f-4e51-8149-f4365a381995"), roles=[], ), }, [], [ TombstoneIdentityTenant( **{ "created_at": "2022-01-01T00:00:00+00:00", "created_by": "0c009413-1ee6-444f-a875-571298eca67e", "expires_at": 1640995201, "identity_uuid": "TOMBSTONE:57c25895-ce10-483f-b187-44c24308f8da:1640995200", # noqa: E501 "tenant_uuid": "TOMBSTONE:86491fd9-518d-4087-bb74-046b2b8b99e5", "is_tombstone": True, "roles": [], "tenant_type": "account", "updated_at": "2022-01-01T00:00:00+00:00", "updated_by": "0c009413-1ee6-444f-a875-571298eca67e", "version": "0", } ), TombstoneIdentityTenant( **{ "created_at": "2022-01-01T00:00:00+00:00", "created_by": "0c009413-1ee6-444f-a875-571298eca67e", "expires_at": 1640995201, "identity_uuid": "TOMBSTONE:57c25895-ce10-483f-b187-44c24308f8da:1640995200", # noqa: E501 "tenant_uuid": "TOMBSTONE:31120501-2d3f-4e51-8149-f4365a381995", "is_tombstone": True, "roles": [], "tenant_type": "account", "updated_at": "2022-01-01T00:00:00+00:00", "updated_by": "0c009413-1ee6-444f-a875-571298eca67e", "version": "0", } ), ], id="The result includes all tenants when unprocessed_items is empty.", ), pytest.param( { uuid.UUID("86491fd9-518d-4087-bb74-046b2b8b99e5"): TenantRoles( tenant_type=TenantType.TENANT_TYPE_ACCOUNT, tenant_uuid=uuid.UUID("86491fd9-518d-4087-bb74-046b2b8b99e5"), roles=[], ), uuid.UUID("31120501-2d3f-4e51-8149-f4365a381995"): TenantRoles( tenant_type=TenantType.TENANT_TYPE_ACCOUNT, tenant_uuid=uuid.UUID("31120501-2d3f-4e51-8149-f4365a381995"), roles=[], ), }, [ { config.IDENTITY_HASH_KEY: "57c25895-ce10-483f-b187-44c24308f8da", config.IDENTITY_RANGE_KEY: uuid.UUID( "86491fd9-518d-4087-bb74-046b2b8b99e5" ), }, { config.IDENTITY_HASH_KEY: "57c25895-ce10-483f-b187-44c24308f8da", config.IDENTITY_RANGE_KEY: uuid.UUID( "31120501-2d3f-4e51-8149-f4365a381995" ), }, {"some": "attribute"}, ], [], id="The result should include 0 tombstones when every tenant is in unprocessed_items.", # noqa: E501 ), pytest.param( {}, [], [], id="Its OK if both inputs are empty.", ), pytest.param( {}, [ { config.IDENTITY_HASH_KEY: "57c25895-ce10-483f-b187-44c24308f8da", config.IDENTITY_RANGE_KEY: uuid.UUID( "86491fd9-518d-4087-bb74-046b2b8b99e5" ), }, { config.IDENTITY_HASH_KEY: "57c25895-ce10-483f-b187-44c24308f8da", config.IDENTITY_RANGE_KEY: uuid.UUID( "31120501-2d3f-4e51-8149-f4365a381995" ), }, {"some": "attribute"}, ], [], id="It shouldn't happen but an empty request list + unprocessed entries should return nothing.", # noqa: E501 ), ], ) @freezegun.freeze_time("2022-01-01") def test_generate_tombstone_records( request_tenants: Dict[uuid.UUID, TenantRoles], unprocessed_items: List[Dict[str, Any]], expected: List[TombstoneIdentityTenant], ) -> None: """Test _generate_tombstone_records.""" authenticated_identity_uuid = "0c009413-1ee6-444f-a875-571298eca67e" identity_uuid = "57c25895-ce10-483f-b187-44c24308f8da" identity = Identity(identity_uuid, MagicMock()) deactivated_tenants = identity._generate_tombstone_records( authenticated_identity_uuid=authenticated_identity_uuid, request_tenants=request_tenants, unprocessed_items=unprocessed_items, ) assert sorted(deactivated_tenants) == sorted(expected) @pytest.mark.parametrize( "request_tenants, unprocessed_items, expected", [ pytest.param( { uuid.UUID("86491fd9-518d-4087-bb74-046b2b8b99e5"): TenantRoles( tenant_type=TenantType.TENANT_TYPE_ACCOUNT, tenant_uuid=uuid.UUID("86491fd9-518d-4087-bb74-046b2b8b99e5"), roles=[], ), uuid.UUID("31120501-2d3f-4e51-8149-f4365a381995"): TenantRoles( tenant_type=TenantType.TENANT_TYPE_ACCOUNT, tenant_uuid=uuid.UUID("31120501-2d3f-4e51-8149-f4365a381995"), roles=[], ), }, [ { config.IDENTITY_HASH_KEY: "57c25895-ce10-483f-b187-44c24308f8da", config.IDENTITY_RANGE_KEY: uuid.UUID( "31120501-2d3f-4e51-8149-f4365a381995" ), }, ], [ TombstoneIdentityTenant( **{ "created_at": "2022-01-01T00:00:00+00:00", "created_by": "0c009413-1ee6-444f-a875-571298eca67e", "created_impersonated_by": "a2120f7d-df8b-49d0-abd6-82754554927a", # noqa: E501 "expires_at": 1640995201, "identity_uuid": "TOMBSTONE:57c25895-ce10-483f-b187-44c24308f8da:1640995200", # noqa: E501 "tenant_uuid": "TOMBSTONE:86491fd9-518d-4087-bb74-046b2b8b99e5", "is_tombstone": True, "roles": [], "tenant_type": "account", "updated_at": "2022-01-01T00:00:00+00:00", "updated_by": "0c009413-1ee6-444f-a875-571298eca67e", "updated_impersonated_by": "a2120f7d-df8b-49d0-abd6-82754554927a", # noqa: E501 "version": "0", } ), ], id="The result should not include a tenant in unprocessed_items", ), pytest.param( { uuid.UUID("86491fd9-518d-4087-bb74-046b2b8b99e5"): TenantRoles( tenant_type=TenantType.TENANT_TYPE_ACCOUNT, tenant_uuid=uuid.UUID("86491fd9-518d-4087-bb74-046b2b8b99e5"), roles=[], ), uuid.UUID("31120501-2d3f-4e51-8149-f4365a381995"): TenantRoles( tenant_type=TenantType.TENANT_TYPE_ACCOUNT, tenant_uuid=uuid.UUID("31120501-2d3f-4e51-8149-f4365a381995"), roles=[], ), }, [ { config.IDENTITY_HASH_KEY: "57c25895-ce10-483f-b187-44c24308f8da", config.IDENTITY_RANGE_KEY: uuid.UUID( "31120501-2d3f-4e51-8149-f4365a381995" ), }, { config.IDENTITY_HASH_KEY: "57c25895-ce10-483f-b187-44c24308f8da", config.IDENTITY_RANGE_KEY: uuid.UUID( "31120501-2d3f-4e51-8149-f4365a381995" ), }, { config.IDENTITY_HASH_KEY: "57c25895-ce10-483f-b187-44c24308f8da", config.IDENTITY_RANGE_KEY: uuid.UUID( "31120501-2d3f-4e51-8149-f4365a381995" ), }, ], [ TombstoneIdentityTenant( **{ "created_at": "2022-01-01T00:00:00+00:00", "created_by": "0c009413-1ee6-444f-a875-571298eca67e", "created_impersonated_by": "a2120f7d-df8b-49d0-abd6-82754554927a", # noqa: E501 "expires_at": 1640995201, "identity_uuid": "TOMBSTONE:57c25895-ce10-483f-b187-44c24308f8da:1640995200", # noqa: E501 "tenant_uuid": "TOMBSTONE:86491fd9-518d-4087-bb74-046b2b8b99e5", "is_tombstone": True, "roles": [], "tenant_type": "account", "updated_at": "2022-01-01T00:00:00+00:00", "updated_by": "0c009413-1ee6-444f-a875-571298eca67e", "updated_impersonated_by": "a2120f7d-df8b-49d0-abd6-82754554927a", # noqa: E501 "version": "0", } ), ], id="Handles when unprocessed_items contains duplicates", ), pytest.param( { uuid.UUID("86491fd9-518d-4087-bb74-046b2b8b99e5"): TenantRoles( tenant_type=TenantType.TENANT_TYPE_ACCOUNT, tenant_uuid=uuid.UUID("86491fd9-518d-4087-bb74-046b2b8b99e5"), roles=[], ), uuid.UUID("31120501-2d3f-4e51-8149-f4365a381995"): TenantRoles( tenant_type=TenantType.TENANT_TYPE_ACCOUNT, tenant_uuid=uuid.UUID("31120501-2d3f-4e51-8149-f4365a381995"), roles=[], ), }, [], [ TombstoneIdentityTenant( **{ "created_at": "2022-01-01T00:00:00+00:00", "created_by": "0c009413-1ee6-444f-a875-571298eca67e", "created_impersonated_by": "a2120f7d-df8b-49d0-abd6-82754554927a", # noqa: E501 "expires_at": 1640995201, "identity_uuid": "TOMBSTONE:57c25895-ce10-483f-b187-44c24308f8da:1640995200", # noqa: E501 "tenant_uuid": "TOMBSTONE:86491fd9-518d-4087-bb74-046b2b8b99e5", "is_tombstone": True, "roles": [], "tenant_type": "account", "updated_at": "2022-01-01T00:00:00+00:00", "updated_by": "0c009413-1ee6-444f-a875-571298eca67e", "updated_impersonated_by": "a2120f7d-df8b-49d0-abd6-82754554927a", # noqa: E501 "version": "0", } ), TombstoneIdentityTenant( **{ "created_at": "2022-01-01T00:00:00+00:00", "created_by": "0c009413-1ee6-444f-a875-571298eca67e", "created_impersonated_by": "a2120f7d-df8b-49d0-abd6-82754554927a", # noqa: E501 "expires_at": 1640995201, "identity_uuid": "TOMBSTONE:57c25895-ce10-483f-b187-44c24308f8da:1640995200", # noqa: E501 "tenant_uuid": "TOMBSTONE:31120501-2d3f-4e51-8149-f4365a381995", "is_tombstone": True, "roles": [], "tenant_type": "account", "updated_at": "2022-01-01T00:00:00+00:00", "updated_by": "0c009413-1ee6-444f-a875-571298eca67e", "updated_impersonated_by": "a2120f7d-df8b-49d0-abd6-82754554927a", # noqa: E501 "version": "0", } ), ], id="The result includes all tenants when unprocessed_items is empty.", ), pytest.param( { uuid.UUID("86491fd9-518d-4087-bb74-046b2b8b99e5"): TenantRoles( tenant_type=TenantType.TENANT_TYPE_ACCOUNT, tenant_uuid=uuid.UUID("86491fd9-518d-4087-bb74-046b2b8b99e5"), roles=[], ), uuid.UUID("31120501-2d3f-4e51-8149-f4365a381995"): TenantRoles( tenant_type=TenantType.TENANT_TYPE_ACCOUNT, tenant_uuid=uuid.UUID("31120501-2d3f-4e51-8149-f4365a381995"), roles=[], ), }, [ { config.IDENTITY_HASH_KEY: "57c25895-ce10-483f-b187-44c24308f8da", config.IDENTITY_RANGE_KEY: uuid.UUID( "86491fd9-518d-4087-bb74-046b2b8b99e5" ), }, { config.IDENTITY_HASH_KEY: "57c25895-ce10-483f-b187-44c24308f8da", config.IDENTITY_RANGE_KEY: uuid.UUID( "31120501-2d3f-4e51-8149-f4365a381995" ), }, {"some": "attribute"}, ], [], id="The result should include 0 tombstones when every tenant is in unprocessed_items.", # noqa: E501 ), pytest.param( {}, [], [], id="Its OK if both inputs are empty.", ), pytest.param( {}, [ { config.IDENTITY_HASH_KEY: "57c25895-ce10-483f-b187-44c24308f8da", config.IDENTITY_RANGE_KEY: uuid.UUID( "86491fd9-518d-4087-bb74-046b2b8b99e5" ), }, { config.IDENTITY_HASH_KEY: "57c25895-ce10-483f-b187-44c24308f8da", config.IDENTITY_RANGE_KEY: uuid.UUID( "31120501-2d3f-4e51-8149-f4365a381995" ), }, {"some": "attribute"}, ], [], id="It shouldn't happen but an empty request list + unprocessed entries should return nothing.", # noqa: E501 ), ], ) @freezegun.freeze_time("2022-01-01") def test_generate_tombstone_records_with_impersonated_by_identity_uuid( request_tenants: Dict[uuid.UUID, TenantRoles], unprocessed_items: List[Dict[str, Any]], expected: List[TombstoneIdentityTenant], ) -> None: """Test _generate_tombstone_records with impersonated_by_identity_uuid.""" authenticated_identity_uuid = "0c009413-1ee6-444f-a875-571298eca67e" impersonated_by_identity_uuid = "a2120f7d-df8b-49d0-abd6-82754554927a" identity_uuid = "57c25895-ce10-483f-b187-44c24308f8da" identity = Identity(identity_uuid, MagicMock()) deactivated_tenants = identity._generate_tombstone_records( authenticated_identity_uuid=authenticated_identity_uuid, request_tenants=request_tenants, unprocessed_items=unprocessed_items, impersonated_by_identity_uuid=impersonated_by_identity_uuid, ) assert sorted(deactivated_tenants) == sorted(expected) @pytest.mark.parametrize( "tombstone_records, unprocessed_items", [ pytest.param( [ { "created_at": "2022-01-01T00:00:00+00:00", "created_by": "0c009413-1ee6-444f-a875-571298eca67e", "created_impersonated_by": None, "expires_at": 1640995205, "identity_uuid": "TOMBSTONE:57c25895-ce10-483f-b187-44c24308f8da:1640995200", # noqa: E501 "tenant_uuid": "TOMBSTONE:86491fd9-518d-4087-bb74-046b2b8b99e5", "is_tombstone": True, "roles": [], "tenant_type": "account", "updated_at": "2022-01-01T00:00:00+00:00", "updated_by": "0c009413-1ee6-444f-a875-571298eca67e", "updated_impersonated_by": None, "version": "0", }, { "created_at": "2022-01-01T00:00:00+00:00", "created_by": "0c009413-1ee6-444f-a875-571298eca67e", "created_impersonated_by": None, "expires_at": 1640995205, "identity_uuid": "TOMBSTONE:57c25895-ce10-483f-b187-44c24308f8da:1640995200", # noqa: E501 "tenant_uuid": "TOMBSTONE:31120501-2d3f-4e51-8149-f4365a381995", "is_tombstone": True, "roles": [], "tenant_type": "account", "updated_at": "2022-01-01T00:00:00+00:00", "updated_by": "0c009413-1ee6-444f-a875-571298eca67e", "updated_impersonated_by": None, "version": "0", }, ], [ { "created_at": "2022-01-01T00:00:00+00:00", "created_by": "0c009413-1ee6-444f-a875-571298eca67e", "expires_at": 1640995205, "identity_uuid": "TOMBSTONE:57c25895-ce10-483f-b187-44c24308f8da:1640995200", # noqa: E501 "tenant_uuid": "TOMBSTONE:31120501-2d3f-4e51-8149-f4365a381995", "is_tombstone": True, "roles": [], "tenant_type": "account", "updated_at": "2022-01-01T00:00:00+00:00", "updated_by": "0c009413-1ee6-444f-a875-571298eca67e", "version": "0", } ], id="1 tombstone write succeeds and 1 fails", ), pytest.param( [ { "created_at": "2022-01-01T00:00:00+00:00", "created_by": "0c009413-1ee6-444f-a875-571298eca67e", "created_impersonated_by": None, "expires_at": 1640995205, "identity_uuid": "TOMBSTONE:57c25895-ce10-483f-b187-44c24308f8da:1640995200", # noqa: E501 "tenant_uuid": "TOMBSTONE:86491fd9-518d-4087-bb74-046b2b8b99e5", "is_tombstone": True, "roles": [], "tenant_type": "account", "updated_at": "2022-01-01T00:00:00+00:00", "updated_by": "0c009413-1ee6-444f-a875-571298eca67e", "updated_impersonated_by": None, "version": "0", }, { "created_at": "2022-01-01T00:00:00+00:00", "created_by": "0c009413-1ee6-444f-a875-571298eca67e", "created_impersonated_by": None, "expires_at": 1640995205, "identity_uuid": "TOMBSTONE:57c25895-ce10-483f-b187-44c24308f8da:1640995200", # noqa: E501 "tenant_uuid": "TOMBSTONE:31120501-2d3f-4e51-8149-f4365a381995", "is_tombstone": True, "roles": [], "tenant_type": "account", "updated_at": "2022-01-01T00:00:00+00:00", "updated_by": "0c009413-1ee6-444f-a875-571298eca67e", "updated_impersonated_by": None, "version": "0", }, ], [], id="2 tombstone writes succeeds and 0 fails", ), ], ) async def test_write_tombstone_records( tombstone_records: List[TombstoneIdentityTenant], unprocessed_items: List[Dict[str, Any]], mock_identity_ddb_connector: AsyncMock, caplog: pytest.LogCaptureFixture, ) -> None: """Test write_tombstone_records.""" identity_uuid = "31be3bb9-fbff-43d7-9519-48b16aabbc7f" identity = Identity( identity_uuid, identity_ddb_connector=mock_identity_ddb_connector ) mock_identity_ddb_connector.put_items = AsyncMock(return_value=unprocessed_items) result = await identity._write_tombstone_records( TombstoneIdentityTenantValidator.validate_python(tombstone_records) ) assert result == unprocessed_items mock_identity_ddb_connector.put_items.assert_called_once_with( tombstone_records, ) with caplog.at_level(logging.ERROR): assert len(caplog.records) == len(unprocessed_items) for record in caplog.records: assert "Failed to write tombstone record" in record.msg async def test_write_tombstone_records_no_input( mock_identity_ddb_connector: AsyncMock, caplog: pytest.LogCaptureFixture, ) -> None: """Test write_tombstone_records with an empty list.""" identity_uuid = "31be3bb9-fbff-43d7-9519-48b16aabbc7f" identity = Identity( identity_uuid, identity_ddb_connector=mock_identity_ddb_connector ) mock_identity_ddb_connector.put_items = AsyncMock(return_value=[]) result = await identity._write_tombstone_records([]) assert result == [] mock_identity_ddb_connector.put_items.assert_not_called() with caplog.at_level(logging.ERROR): assert len(caplog.records) == 0 @patch.object(Identity, "_generate_tombstone_records") @patch.object(Identity, "_write_tombstone_records", new_callable=AsyncMock) async def test_write_tombstone_records_for_deactivations( mock_write_tombstone_records: AsyncMock, mock_generate_tombstone_records: MagicMock, mock_identity_ddb_connector: AsyncMock, ) -> None: """Test write_tombstone_records_for_deactivations.""" authenticated_identity_uuid = "0c009413-1ee6-444f-a875-571298eca67e" identity_uuid = "57c25895-ce10-483f-b187-44c24308f8da" identity = Identity( identity_uuid, identity_ddb_connector=mock_identity_ddb_connector ) request_tenants = { uuid.UUID("86491fd9-518d-4087-bb74-046b2b8b99e5"): TenantRoles( tenant_type=TenantType.TENANT_TYPE_ACCOUNT, tenant_uuid=uuid.UUID("86491fd9-518d-4087-bb74-046b2b8b99e5"), roles=[], ), } tombstone_records = [ TombstoneIdentityTenant( **{ "created_at": "2022-01-01T00:00:00+00:00", "created_by": authenticated_identity_uuid, "expires_at": 1640995205, "identity_uuid": "TOMBSTONE:57c25895-ce10-483f-b187-44c24308f8da:1640995200", # noqa: E501 "tenant_uuid": "TOMBSTONE:86491fd9-518d-4087-bb74-046b2b8b99e5", "is_tombstone": True, "roles": [], "tenant_type": "account", "updated_at": "2022-01-01T00:00:00+00:00", "updated_by": authenticated_identity_uuid, "version": "0", } ) ] mock_generate_tombstone_records.return_value = tombstone_records mock_write_tombstone_records.return_value = [] result = await identity.write_tombstone_records_for_deactivations( authenticated_identity_uuid, request_tenants, unprocessed_items=[], ) assert result == [] mock_generate_tombstone_records.assert_called_once_with( authenticated_identity_uuid=authenticated_identity_uuid, request_tenants=request_tenants, unprocessed_items=[], ) mock_write_tombstone_records.assert_called_once_with(tombstone_records) @patch.object(Identity, "_generate_tombstone_records") @patch.object(Identity, "_write_tombstone_records", new_callable=AsyncMock) async def test_write_tombstone_records_for_deactivations_with_principal( mock_write_tombstone_records: AsyncMock, mock_generate_tombstone_records: MagicMock, mock_identity_ddb_connector: AsyncMock, ) -> None: """Test write_tombstone_records_for_deactivations with principal.""" authenticated_identity_uuid = "0c009413-1ee6-444f-a875-571298eca67e" identity_uuid = "57c25895-ce10-483f-b187-44c24308f8da" identity = Identity( identity_uuid, identity_ddb_connector=mock_identity_ddb_connector ) request_tenants = { uuid.UUID("86491fd9-518d-4087-bb74-046b2b8b99e5"): TenantRoles( tenant_type=TenantType.TENANT_TYPE_ACCOUNT, tenant_uuid=uuid.UUID("86491fd9-518d-4087-bb74-046b2b8b99e5"), roles=[], ), } tombstone_records = [ TombstoneIdentityTenant( **{ "created_at": "2022-01-01T00:00:00+00:00", "created_by": authenticated_identity_uuid, "expires_at": 1640995205, "identity_uuid": "TOMBSTONE:57c25895-ce10-483f-b187-44c24308f8da:1640995200", # noqa: E501 "tenant_uuid": "TOMBSTONE:86491fd9-518d-4087-bb74-046b2b8b99e5", "is_tombstone": True, "roles": [], "tenant_type": "account", "updated_at": "2022-01-01T00:00:00+00:00", "updated_by": authenticated_identity_uuid, "version": "0", } ) ] mock_generate_tombstone_records.return_value = tombstone_records mock_write_tombstone_records.return_value = [] result = await identity.write_tombstone_records_for_deactivations( authenticated_identity_uuid, request_tenants, unprocessed_items=[], principal=Principal( identity_uuid=authenticated_identity_uuid, pdp_tenant_roles={}, # irrelevant to this test case ), ) assert result == [] mock_generate_tombstone_records.assert_called_once_with( authenticated_identity_uuid=authenticated_identity_uuid, request_tenants=request_tenants, unprocessed_items=[], impersonated_by_identity_uuid=None, ) mock_write_tombstone_records.assert_called_once_with(tombstone_records) @patch.object(Identity, "_generate_tombstone_records") @patch.object(Identity, "_write_tombstone_records", new_callable=AsyncMock) async def test_write_tombstone_records_for_deactivations_with_impersonation( mock_write_tombstone_records: AsyncMock, mock_generate_tombstone_records: MagicMock, mock_identity_ddb_connector: AsyncMock, ) -> None: """Test write_tombstone_records_for_deactivations with impersonated principal.""" authenticated_identity_uuid = "0c009413-1ee6-444f-a875-571298eca67e" impersonated_by_identity_uuid = "a2120f7d-df8b-49d0-abd6-82754554927a" identity_uuid = "57c25895-ce10-483f-b187-44c24308f8da" identity = Identity( identity_uuid, identity_ddb_connector=mock_identity_ddb_connector ) request_tenants = { uuid.UUID("86491fd9-518d-4087-bb74-046b2b8b99e5"): TenantRoles( tenant_type=TenantType.TENANT_TYPE_ACCOUNT, tenant_uuid=uuid.UUID("86491fd9-518d-4087-bb74-046b2b8b99e5"), roles=[], ), } tombstone_records = [ TombstoneIdentityTenant( **{ "created_at": "2022-01-01T00:00:00+00:00", "created_by": authenticated_identity_uuid, "created_impersonated_by": impersonated_by_identity_uuid, "expires_at": 1640995205, "identity_uuid": "TOMBSTONE:57c25895-ce10-483f-b187-44c24308f8da:1640995200", # noqa: E501 "tenant_uuid": "TOMBSTONE:86491fd9-518d-4087-bb74-046b2b8b99e5", "is_tombstone": True, "roles": [], "tenant_type": "account", "updated_at": "2022-01-01T00:00:00+00:00", "updated_by": authenticated_identity_uuid, "updated_impersonated_by": impersonated_by_identity_uuid, "version": "0", } ) ] mock_generate_tombstone_records.return_value = tombstone_records mock_write_tombstone_records.return_value = [] result = await identity.write_tombstone_records_for_deactivations( authenticated_identity_uuid, request_tenants, unprocessed_items=[], principal=Principal( identity_uuid=authenticated_identity_uuid, pdp_tenant_roles={}, # irrelevant to this test case impersonated_by_identity_uuid=impersonated_by_identity_uuid, ), ) assert result == [] mock_generate_tombstone_records.assert_called_once_with( authenticated_identity_uuid=authenticated_identity_uuid, request_tenants=request_tenants, unprocessed_items=[], impersonated_by_identity_uuid=impersonated_by_identity_uuid, ) mock_write_tombstone_records.assert_called_once_with(tombstone_records) @pytest.mark.parametrize( "principal", [ pytest.param(None, id="principal is None"), pytest.param( Principal( identity_uuid=uuid4(), pdp_tenant_roles={}, ), id="principal is not impersonated", ), pytest.param( Principal( identity_uuid=uuid4(), pdp_tenant_roles={}, impersonated_by_identity_uuid=uuid4(), ), id="principal is impersonated", ), ], ) @patch.object(Identity, "_generate_tombstone_records") @patch.object(Identity, "_write_tombstone_records", new_callable=AsyncMock) async def test_write_tombstone_records_for_deactivations__empty_input( mock_write_tombstone_records: AsyncMock, mock_generate_tombstone_records: MagicMock, principal: Principal | None, mock_identity_ddb_connector: AsyncMock, ) -> None: """Test write_tombstone_records_for_deactivations when no unprocessed items.""" authenticated_identity_uuid = "0c009413-1ee6-444f-a875-571298eca67e" identity_uuid = "57c25895-ce10-483f-b187-44c24308f8da" identity = Identity( identity_uuid, identity_ddb_connector=mock_identity_ddb_connector ) mock_generate_tombstone_records.return_value = [] mock_write_tombstone_records.return_value = [] result = await identity.write_tombstone_records_for_deactivations( authenticated_identity_uuid, request_tenants={}, unprocessed_items=[], principal=principal, ) assert result == [] mock_generate_tombstone_records.assert_not_called() mock_write_tombstone_records.assert_not_called()