from typing import Any import pytest from pdp.constants.constants import CacheEntryType from pdp.fastapi.schemas.cache import PrincipalPdpCacheObject @pytest.mark.parametrize( "payload, expect_exception, expected", [ pytest.param({}, True, None, id="An empty payload raises an error"), pytest.param( {"identity_uuid": "INVALID", "tenant_roles": {}}, True, None, id="Invalid identity_uuid", ), pytest.param( { "identity_uuid": "89b79b52-de18-4644-bd2b-bd28fdbeebaa", "tenant_roles": {"INVALID": {}}, }, True, None, id="Invalid tenant_roles", ), pytest.param( { "identity_uuid": "89b79b52-de18-4644-bd2b-bd28fdbeebaa", "tenant_roles": { "74e4aefa-d4a4-44f1-878f-1627612f0bb8": { "roles": [], "tenant_type": "account", "tenant_uuid": "74e4aefa-d4a4-44f1-878f-1627612f0bb8", } }, }, False, { "identity_uuid": "89b79b52-de18-4644-bd2b-bd28fdbeebaa", "tenant_roles": { "74e4aefa-d4a4-44f1-878f-1627612f0bb8": { "roles": [], "tenant_type": "account", "tenant_uuid": "74e4aefa-d4a4-44f1-878f-1627612f0bb8", } }, }, id="Valid payload", ), ], ) def test_principal_pdp_cache_object_validate( payload: Any, expect_exception: bool, expected: Any ) -> None: """Test PrincipalPdpCacheObject validation.""" if expect_exception: with pytest.raises(ValueError): PrincipalPdpCacheObject.model_validate(payload) else: assert PrincipalPdpCacheObject.model_validate( payload ) == PrincipalPdpCacheObject(**expected) @pytest.fixture() def cache_object() -> PrincipalPdpCacheObject: """Reusable PrincipalPdpCacheObject fixture.""" return PrincipalPdpCacheObject( identity_uuid="89b79b52-de18-4644-bd2b-bd28fdbeebaa", tenant_roles={}, ) def test_principal_pdp_cache_object_to_key_vals_tuple( cache_object: PrincipalPdpCacheObject, ) -> None: """Test PrincipalPdpCacheObject.to_key_vals_tuple().""" assert cache_object.to_key_vals_tuple() == [ ("identity_uuid", "89b79b52-de18-4644-bd2b-bd28fdbeebaa") ] def test_principal_pdp_cache_object_get_cache_entry_type( cache_object: PrincipalPdpCacheObject, ) -> None: """Test PrincipalPdpCacheObject.get_cache_entry_type().""" assert ( cache_object.get_cache_entry_type() == CacheEntryType.CACHE_ENTRY_PRINCIPAL_PDP ) def test_principal_pdp_cache_object_to_cache_key( cache_object: PrincipalPdpCacheObject, ) -> None: """Test PrincipalPdpCacheObject.to_cache_key().""" assert ( cache_object.to_cache_key() == "principal_pdp@identity_uuid#89b79b52-de18-4644-bd2b-bd28fdbeebaa" ) def test_principal_pdp_cache_object_get_ttl( cache_object: PrincipalPdpCacheObject, ) -> None: """Test PrincipalPdpCacheObject.get_ttl().""" assert cache_object.get_ttl() == 86400 def test_principal_pdp_cache_object_is_valid_to_save( cache_object: PrincipalPdpCacheObject, ) -> None: """Test PrincipalPdpCacheObject.is_valid_to_save().""" assert cache_object.is_valid_to_save() def test_principal_pdp_cache_object_is_not_valid_to_save( cache_object: PrincipalPdpCacheObject, ) -> None: """Test PrincipalPdpCacheObject.is_valid_to_save() fails.""" cache_object.tenant_roles = None assert not cache_object.is_valid_to_save()