"""Infra endpoint tests.""" import uuid from typing import Any, Dict, List from unittest.mock import AsyncMock, MagicMock, patch import pytest from fastapi import FastAPI from fastapi.testclient import TestClient from splitio.client.factory import Client as SplitioClient from pdp import config from pdp.connectors.features import FEATURE_OFF, FEATURE_ON from pdp.connectors.ows_account import OwsAccountClient from pdp.connectors.ows_participant import ( LookupParticipant, LookupParticipantsResponse, OwsParticipantClient, ) from pdp.connectors.ows_permissions import OwsPermissionsClient from pdp.connectors.redis_client import RedisConnector from pdp.constants.constants import ( USER_TYPE_MACHINE, CacheEntryType, TenantType, ) from pdp.fastapi.auth import ( impersonated_by_identity_uuid_from_scope, user_type_from_scope, ) from pdp.fastapi.schemas.check_resources import CheckResourcesRequest from pdp.fastapi.schemas.identity import TenantRoles from pdp.fastapi.schemas.infra import BludgeonCacheRequest from pdp.fastapi.schemas.principal import Principal from pdp.fastapi.schemas.tenant import ( IdExchangeTenantHierarchy, Tenant, UuidToIdExchangeTenant, ) def test_hello(test_client: TestClient) -> None: """Verify the /hello/ endpoint responds with 200 and OK status.""" response = test_client.get("/hello/") assert response.status_code == 200 assert response.json() == {"status": "ok"} def test_world(test_client: TestClient, identity_uuid: str) -> None: """Verify the root endpoint responds with the expected response and 200.""" response = test_client.get("/?name=awesome") assert response.status_code == 200 assert ( response.text == f"Hello awesome from fastapi! identity_uuid='{identity_uuid}'." ) @pytest.mark.parametrize( "feature_is_enabled, expected_response", [ (True, {"message": "bonjour"}), (False, {"message": "hello"}), ], ) @patch("pdp.fastapi.routers.infra.BooleanFeature") def test_bonjour( mock_boolean_feature: MagicMock, feature_is_enabled: bool, expected_response: Dict[str, Any], test_client: TestClient, identity_uuid: str, ) -> None: """Verify the feature flag demo endpoint.""" mock_boolean_feature_instance = mock_boolean_feature.return_value mock_boolean_feature_instance.is_on_for_identity.return_value = feature_is_enabled response = test_client.get(config.CONNECTIVITY_SPLITIO) assert response.status_code == 200 assert response.json() == expected_response mock_boolean_feature_instance.is_on_for_identity.assert_called_with(identity_uuid) @patch("pdp.fastapi.routers.infra.RedisConnector") def test_redis_ping( mock_redis: AsyncMock, test_client: TestClient, monkeypatch: Any ) -> None: """Verify the endpoint to check redis connectivity handler.""" monkeypatch.setattr(config, "REDIS_URL", "redis://no.such.server:6379") mock_redis_instance = AsyncMock(spec=RedisConnector) mock_redis.return_value = mock_redis_instance mock_redis_instance.ping.return_value = True response = test_client.get(config.CONNECTIVITY_REDIS) assert response.status_code == 200 assert response.json() == { "status": True, "redis_url": "redis://no.such.server:6379", } def test_ows_permissions_connectivity( mock_ows_permissions_client: OwsPermissionsClient, test_client: TestClient, ) -> None: """Verify handler to check ows-permissions connectivity.""" response = test_client.get(config.CONNECTIVITY_OWS_PERMISSIONS) assert response.status_code == 200 assert response.json() == { "items": [ { "id": 7123, "type": "Vendor", "uuid": "573d0372-7f2f-48a6-8deb-c9a6558f9549", } ], "pagination": {"total_records": 1, "type": "classico"}, } @pytest.mark.parametrize( "lookup_request, expected_response, expected_status_code, description", [ ( { "uuids": ["da1c65c4-4d35-4ea7-a4c0-17d3ac43466a"], "fetch_flags": ["TENANT_HIERARCHY"], }, { "vendors": [ { "vendor_id": 0, "uuid": "da1c65c4-4d35-4ea7-a4c0-17d3ac43466a", "company_brand_uuid": "d25a4cd1-e820-45f2-be5c-56edcfeb8298", "parent_company_uuid": None, } ] }, 200, "Lookup with TENANT_HIERARCHY should return a company brand", ), ( { "uuids": [ "da1c65c4-4d35-4ea7-a4c0-17d3ac43466a", "71f2132f-40f2-4f84-9fcc-3cf825c32d87", ] }, { "vendors": [ { "vendor_id": 0, "uuid": "da1c65c4-4d35-4ea7-a4c0-17d3ac43466a", "company_brand_uuid": None, "parent_company_uuid": None, }, { "vendor_id": 1, "uuid": "71f2132f-40f2-4f84-9fcc-3cf825c32d87", "company_brand_uuid": None, "parent_company_uuid": None, }, ] }, 200, "Lookup without TENANT_HIERARCHY should have a null company brand", ), ( {}, {}, 422, "Empty response", ), ], ) def test_ows_account_connectivity( lookup_request: Dict[str, Any], expected_response: Dict[str, Any], expected_status_code: int, description: str, mock_ows_account_client: OwsAccountClient, test_client: TestClient, ) -> None: """Verify handler to check ows-account connectivity.""" response = test_client.post( config.CONNECTIVITY_OWS_ACCOUNT, json=lookup_request, ) if expected_status_code == 200: assert response.status_code == expected_status_code, description assert response.json() == expected_response, description else: assert response.status_code == expected_status_code, description assert response.json()["detail"][0]["msg"] == "Field required", description @pytest.mark.parametrize( "body, returned_cache_items, expected_response, expected_status_code, description", [ ( {"keys": []}, None, { "detail": [ { "type": "too_short", "loc": ["body", "keys"], "msg": "List should have at least 1 item after validation, not 0", # noqa: E501 "input": [], "ctx": { "actual_length": 0, "field_type": "List", "min_length": 1, }, } ] }, 422, "Should return 422 if request has no keys", ), ( { "keys": [ "tenant_hierarchy_fff741c2-6def-4493-bfdf-c2bcb1128e02", "tenant_hierarchy_not-a-cached-tenant", ] }, [ { "vendor_id": 6971, "uuid": "fff741c2-6def-4493-bfdf-c2bcb1128e02", "company_brand_uuid": "d25a4cd1-e820-45f2-be5c-56edcfeb8298", "parent_company_uuid": "955a1bbd-b623-4ea1-ab5f-8d6620c442fb", }, {}, ], { "items": [ { "key": "tenant_hierarchy_fff741c2-6def-4493-bfdf-c2bcb1128e02", "value": { "vendor_id": 6971, "uuid": "fff741c2-6def-4493-bfdf-c2bcb1128e02", "company_brand_uuid": "d25a4cd1-e820-45f2-be5c-56edcfeb8298", # noqa: E501 "parent_company_uuid": "955a1bbd-b623-4ea1-ab5f-8d6620c442fb", # noqa: E501 }, }, {"key": "tenant_hierarchy_not-a-cached-tenant", "value": {}}, ] }, 200, "Should return 200 and the list of cache items for a valid response", ), ], ) def test_cache_list( body: Dict[str, Any], returned_cache_items: List[Any], expected_response: Dict[str, Any], expected_status_code: int, description: str, mock_redis_connector: MagicMock, test_client: TestClient, ) -> None: """Verify the cache list handler.""" mock_redis_connector.mget = AsyncMock(return_value=returned_cache_items) response = test_client.post( "/cache/list/", json=body, ) assert response.status_code == expected_status_code assert response.json() == expected_response, description @pytest.mark.parametrize( "body, returned_cache_items, expected_response, expected_status_code, description", # noqa: E501 [ ( {"keys": []}, None, { "detail": [ { "type": "too_short", "loc": ["body", "keys"], "msg": "List should have at least 1 item after validation, not 0", # noqa: E501 "input": [], "ctx": { "actual_length": 0, "field_type": "List", "min_length": 1, }, } ] }, 422, "Should return 422 if request has no keys", ), ( { "keys": [ "tenant_hierarchy_fff741c2-6def-4493-bfdf-c2bcb1128e02", "tenant_hierarchy_not-a-cached-tenant", ] }, [ { "vendor_id": 6971, "uuid": "fff741c2-6def-4493-bfdf-c2bcb1128e02", "company_brand_uuid": "d25a4cd1-e820-45f2-be5c-56edcfeb8298", "parent_company_uuid": "955a1bbd-b623-4ea1-ab5f-8d6620c442fb", }, None, ], { "items": [ { "key": "tenant_hierarchy_fff741c2-6def-4493-bfdf-c2bcb1128e02", "value": { "vendor_id": 6971, "uuid": "fff741c2-6def-4493-bfdf-c2bcb1128e02", "company_brand_uuid": "d25a4cd1-e820-45f2-be5c-56edcfeb8298", # noqa: E501 "parent_company_uuid": "955a1bbd-b623-4ea1-ab5f-8d6620c442fb", # noqa: E501 }, "deleted": True, }, { "key": "tenant_hierarchy_not-a-cached-tenant", "value": {}, "deleted": False, }, ] }, 200, "Should return 200 and the list of cache items for a valid response", ), ], ) def test_cache_bust( body: Dict[str, Any], returned_cache_items: List[Any], expected_response: Dict[str, Any], expected_status_code: int, description: str, mock_redis_connector: MagicMock, test_client: TestClient, ) -> None: """Verify the cache bust handler.""" mock_redis_connector.getdel = AsyncMock(return_value=returned_cache_items) response = test_client.post( "/cache/bust/", json=body, ) assert response.status_code == expected_status_code assert response.json() == expected_response, description @pytest.mark.parametrize( "body, expected_response, expected_status_code, description", [ ( { "cache_entry_type": "no.such.type", "delete": True, }, { "detail": [ { "type": "enum", "loc": ["body", "cache_entry_type"], "msg": "Input should be 'cerbos_policy_metadata', 'allowed_tenants', 'list_tenant_roles', 'principal_pdp' or 'tenant_hierarchy'", # noqa: E501 "input": "no.such.type", "ctx": { "expected": "'cerbos_policy_metadata', 'allowed_tenants', 'list_tenant_roles', 'principal_pdp' or 'tenant_hierarchy'" # noqa: E501 }, } ] }, 422, "Should return 422 if request has invalid cache_entry_type", ), ( { "cache_entry_type": CacheEntryType.CACHE_ENTRY_TENANT_HIERARCHY.value, "delete": "not.a.bool", }, { "detail": [ { "type": "bool_parsing", "loc": ["body", "delete"], "msg": "Input should be a valid boolean, unable to interpret input", # noqa: E501 "input": "not.a.bool", } ] }, 422, "Should return 422 if request has invalid cache_entry_type", ), ( { "cache_entry_type": CacheEntryType.CACHE_ENTRY_TENANT_HIERARCHY.value, "delete": True, }, { "total_rows_before_delete": 10, "total_rows_affected": 1, "performed_delete": True, }, 200, "Should return a 200 with a valid response", ), ( { "cache_entry_type": CacheEntryType.CACHE_ENTRY_TENANT_HIERARCHY.value, }, { "total_rows_before_delete": 10, "total_rows_affected": 1, "performed_delete": True, }, 200, "Should return a 200 with a request missing the delete attribute.", ), ], ) @patch("pdp.logic.infra.cache_bludgeon") def test_cache_bludgeon( mock_cache_bludgeon: AsyncMock, body: Dict[str, Any], expected_response: Dict[str, Any], expected_status_code: int, description: str, mock_redis_connector: RedisConnector, test_client: TestClient, ) -> None: """Test the cache bludgeon endpoint.""" mock_cache_bludgeon.return_value = expected_response response = test_client.post( "/cache/bludgeon/", json=body, ) assert response.status_code == expected_status_code assert response.json() == expected_response, description if expected_status_code == 200: mock_cache_bludgeon.assert_called_with( bludgeon_cache_request=BludgeonCacheRequest.model_validate(body), redis_connector=mock_redis_connector, ) else: mock_cache_bludgeon.assert_not_called() def test_ows_participant_connectivity( test_client: TestClient, mock_ows_participant_client: MagicMock ) -> None: """Test the ows_participant_test demo handler.""" mock_ows_participant_client.lookup_participants_by_uuids = AsyncMock( return_value=LookupParticipantsResponse( label_participants=[ LookupParticipant( uuid="54381846-a817-4ef4-9e58-7eae15bcf1dd", vendor_uuid="0757c0f1-4bcb-41ef-b8e0-1e980effe998", subaccount_uuid=None, company_brand_uuid="d25a4cd1-e820-45f2-be5c-56edcfeb8298", parent_company_uuid="955a1bbd-b623-4ea1-ab5f-8d6620c442fb", ) ] ) ) response = test_client.post( "/infra/connectivity/ows-participant/", json={"uuids": ["54381846-a817-4ef4-9e58-7eae15bcf1dd"]}, ) assert response.status_code == 200 assert response.json() == { "label_participants": [ { "uuid": "54381846-a817-4ef4-9e58-7eae15bcf1dd", "vendor_uuid": "0757c0f1-4bcb-41ef-b8e0-1e980effe998", "subaccount_uuid": None, "company_brand_uuid": "d25a4cd1-e820-45f2-be5c-56edcfeb8298", "parent_company_uuid": "955a1bbd-b623-4ea1-ab5f-8d6620c442fb", }, ] } @patch("pdp.fastapi.routers.infra.AsyncCerbosClient") @patch("pdp.fastapi.auth.cerbos.check_resources") def test_check_connectivity_cerbos( mock_cerbos_check_resources: AsyncMock, mock_async_cerbos_client: AsyncMock, app: FastAPI, test_client: TestClient, identity_uuid: str, mock_splitio_client: SplitioClient, mock_pdp_tenant_roles: Dict[uuid.UUID, TenantRoles], mock_redis_connector: RedisConnector, mock_ows_account_client: OwsAccountClient, mock_ows_participant_client: OwsParticipantClient, ) -> None: """Test the cerbos connectivity endpoint.""" mock_async_cerbos_client_instance = mock_async_cerbos_client.return_value mock_splitio_client.get_treatment.return_value = FEATURE_ON impersonated_by_identity_uuid = uuid.uuid4() app.dependency_overrides[impersonated_by_identity_uuid_from_scope] = ( lambda: impersonated_by_identity_uuid ) app.dependency_overrides[user_type_from_scope] = lambda: USER_TYPE_MACHINE mock_request_body = { "resources": [ { "resource": { "resource_id": "123", "resource_type": "fan_data_list", "attributes": {"tenant_type": "account", "tenant_uuid": "fail now"}, }, "action": "view", } ] } mock_cerbos_check_resources_response = { "request_id": "hello", "resources": [ { "resource": { "resource_id": "123", "resource_type": "Identity", "attributes": {"attribute": "1"}, }, "action": "view", "effect": "deny", "errors": {"validation_errors": []}, } ], } mock_cerbos_check_resources.return_value = mock_cerbos_check_resources_response user_type = USER_TYPE_MACHINE expected_principal = Principal( identity_uuid=uuid.UUID(identity_uuid), user_type=user_type, pdp_tenant_roles=mock_pdp_tenant_roles, impersonated_by_identity_uuid=impersonated_by_identity_uuid, ) response = test_client.post( config.CONNECTIVITY_CERBOS_FARGATE, json=mock_request_body ) assert response.json() == mock_cerbos_check_resources_response mock_cerbos_check_resources.assert_called_once_with( identity_uuid=str(identity_uuid), check_resources_request=CheckResourcesRequest.model_validate(mock_request_body), pdp_tenant_roles=mock_pdp_tenant_roles, cerbos_client=mock_async_cerbos_client_instance, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, redis_connector=mock_redis_connector, authenticated_identity_uuid=uuid.UUID(identity_uuid), principal=expected_principal, include_resource_attributes=False, splitio_client=mock_splitio_client, ) mock_splitio_client.get_treatment.assert_called_once_with( attributes={"identity_id": str(impersonated_by_identity_uuid)}, feature_flag_name="pp_send_impersonated_by_identity_uuid", key="identity_id", ) @patch("pdp.fastapi.routers.infra.AsyncCerbosClient") @patch("pdp.fastapi.auth.cerbos.check_resources") def test_check_connectivity_cerbos__ff_disabled( mock_cerbos_check_resources: AsyncMock, mock_async_cerbos_client: AsyncMock, app: FastAPI, test_client: TestClient, identity_uuid: str, mock_splitio_client: SplitioClient, mock_pdp_tenant_roles: Dict[uuid.UUID, TenantRoles], mock_redis_connector: RedisConnector, mock_ows_account_client: OwsAccountClient, mock_ows_participant_client: OwsParticipantClient, ) -> None: """Test the cerbos connectivity endpoint with FF disabled.""" mock_async_cerbos_client_instance = mock_async_cerbos_client.return_value mock_splitio_client.get_treatment.return_value = FEATURE_OFF mock_request_body = { "resources": [ { "resource": { "resource_id": "123", "resource_type": "fan_data_list", "attributes": {"tenant_type": "account", "tenant_uuid": "fail now"}, }, "action": "view", } ] } mock_cerbos_check_resources_response = { "request_id": "hello", "resources": [ { "resource": { "resource_id": "123", "resource_type": "Identity", "attributes": {"attribute": "1"}, }, "action": "view", "effect": "deny", "errors": {"validation_errors": []}, } ], } mock_cerbos_check_resources.return_value = mock_cerbos_check_resources_response response = test_client.post( config.CONNECTIVITY_CERBOS_FARGATE, json=mock_request_body ) assert response.json() == mock_cerbos_check_resources_response mock_cerbos_check_resources.assert_called_once_with( identity_uuid=str(identity_uuid), check_resources_request=CheckResourcesRequest.model_validate(mock_request_body), pdp_tenant_roles=mock_pdp_tenant_roles, cerbos_client=mock_async_cerbos_client_instance, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, redis_connector=mock_redis_connector, authenticated_identity_uuid=uuid.UUID(identity_uuid), include_resource_attributes=False, splitio_client=mock_splitio_client, ) mock_splitio_client.get_treatment.assert_called_once_with( attributes={"identity_id": str(identity_uuid)}, feature_flag_name="pp_send_impersonated_by_identity_uuid", key="identity_id", ) @patch("pdp.logic.infra.gather_id_to_uuid_exchange") def test_id_to_uuid_exchange( mock_gather_id_to_uuid_exchange: AsyncMock, test_client: TestClient, mock_redis_connector: MagicMock, ) -> None: """Test the id_to_uuid_exchange endpoint.""" mock_redis_connector.mget = AsyncMock(return_value={}) mock_request_body = [ {"tenant_id": 1, "tenant_type": "account"}, ] return_value = { TenantType.TENANT_TYPE_ACCOUNT: { "1": IdExchangeTenantHierarchy( tenant_id=1, tenant_uuid="87682992-bff2-40ff-aa75-18eb6214679e", tenant_type=TenantType.TENANT_TYPE_ACCOUNT, company_brand=Tenant( tenant_uuid="d25a4cd1-e820-45f2-be5c-56edcfeb8298", tenant_type=TenantType.TENANT_TYPE_COMPANY_BRAND, ), parent_company=Tenant( tenant_uuid="babd6f81-00a9-4001-a5a2-45e512a39f93", tenant_type=TenantType.TENANT_TYPE_PARENT_COMPANY, ), ), }, TenantType.TENANT_TYPE_SUBACCOUNT: {}, } mock_gather_id_to_uuid_exchange.return_value = return_value response = test_client.post(config.ID_TO_UUID_EXCHANGE, json=mock_request_body) assert response.status_code == 200 assert response.json() == { "account": { "1": { "subaccount": None, "account": None, "company_brand": { "tenant_type": "company_brand", "tenant_uuid": "d25a4cd1-e820-45f2-be5c-56edcfeb8298", }, "parent_company": { "tenant_type": "parent_company", "tenant_uuid": "babd6f81-00a9-4001-a5a2-45e512a39f93", }, "tenant_id": 1, "tenant_type": "account", "tenant_uuid": "87682992-bff2-40ff-aa75-18eb6214679e", } }, "subaccount": {}, } mock_gather_id_to_uuid_exchange.assert_called_once() @patch("pdp.fastapi.routers.infra.infra") def test_uuid_to_id_exchange( mock_infra_logic: MagicMock, mock_ows_account_client: MagicMock, mock_redis_connector: RedisConnector, test_client: TestClient, ) -> None: expected = { "573d0372-7f2f-48a6-8deb-c9a6558f9549": { "subaccount": None, "account": None, "company_brand": { "tenant_type": "company_brand", "tenant_uuid": "d25a4cd1-e820-45f2-be5c-56edcfeb8298", }, "parent_company": { "tenant_type": "parent_company", "tenant_uuid": "babd6f81-00a9-4001-a5a2-45e512a39f93", }, "tenant_id": 7123, "tenant_type": "account", "tenant_uuid": "573d0372-7f2f-48a6-8deb-c9a6558f9549", } } tenants = [ { "tenant_type": "account", "tenant_uuid": "573d0372-7f2f-48a6-8deb-c9a6558f9549", }, ] mock_infra_logic.gather_uuid_to_id_exchange = AsyncMock(return_value=expected) response = test_client.post( "/infra/tenant/uuid-to-id-exchange/", json=tenants, ) assert response.status_code == 200, response.text assert response.json() == expected, response.text mock_infra_logic.gather_uuid_to_id_exchange.assert_called_once_with( [UuidToIdExchangeTenant.model_validate(tenants[0])], ows_account_client=mock_ows_account_client, redis_connector=mock_redis_connector, )