"""Tests for tenant validator helper functions.""" import uuid from typing import Any, Dict, List from unittest.mock import AsyncMock, MagicMock, patch from uuid import UUID import pytest 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 TenantType from pdp.fastapi.schemas.identity import ( AttachDetachRolesRequest, Role, TenantRoles, ) from pdp.fastapi.schemas.tenant import Tenant, TenantHierarchy from pdp.proxies.tenant_validators import ( DEFAULT_TENANT_TYPES_FOR_VALIDATION_CHECKS, InvalidAttachDetachRolesRequest, InvalidTenantsError, assert_valid_attach_detach_request, assert_valid_tenants, ) TENANT_ACCOUNT = Tenant( tenant_type=TenantType.TENANT_TYPE_ACCOUNT, tenant_uuid=UUID("f6e82400-917a-4fc3-adce-d84fd1494f8a"), ) TENANT_SUBACCOUNT = Tenant( tenant_type=TenantType.TENANT_TYPE_SUBACCOUNT, tenant_uuid=UUID("8d90b605-a6a4-45a3-bf8a-803bb8ca24a0"), ) TENANT_LP = Tenant( tenant_type=TenantType.TENANT_TYPE_LABEL_PARTICIPANT, tenant_uuid=UUID("6607258a-7c51-46a8-b5b0-a70ac9a71a41"), ) TENANT_COMPANY_BRAND = Tenant( tenant_type=TenantType.TENANT_TYPE_COMPANY_BRAND, tenant_uuid=UUID("9df8b5f9-f17b-45bf-a254-40b4deb52017"), ) TENANT_PARENT_COMPANY = Tenant( tenant_type=TenantType.TENANT_TYPE_PARENT_COMPANY, tenant_uuid=UUID("955a1bbd-b623-4ea1-ab5f-8d6620c442fb"), ) TENANT_ROLE_ACCOUNT = TenantRoles( tenant_type=TenantType.TENANT_TYPE_ACCOUNT, tenant_uuid=UUID("f6e82400-917a-4fc3-adce-d84fd1494f8a"), roles=[Role(role="test_role")], ) TENANT_ROLE_SUBACCOUNT = TenantRoles( tenant_type=TenantType.TENANT_TYPE_SUBACCOUNT, tenant_uuid=UUID("8d90b605-a6a4-45a3-bf8a-803bb8ca24a0"), roles=[Role(role="test_role")], ) TENANT_ROLE_LP = TenantRoles( tenant_type=TenantType.TENANT_TYPE_LABEL_PARTICIPANT, tenant_uuid=UUID("6607258a-7c51-46a8-b5b0-a70ac9a71a41"), roles=[Role(role="test_role")], ) @pytest.mark.parametrize( "tenants, mock_gather_result, expected_invalid_tenants, expected_proxy_tenants, tenant_types_to_validate", # noqa: E501 [ pytest.param( [TENANT_ACCOUNT, TENANT_SUBACCOUNT, TENANT_LP, TENANT_COMPANY_BRAND], { TENANT_ACCOUNT.tenant_uuid: TenantHierarchy( company_brand=Tenant( tenant_uuid=UUID("9df8b5f9-f17b-45bf-a254-40b4deb52017"), tenant_type=TenantType.TENANT_TYPE_COMPANY_BRAND, ) ), TENANT_COMPANY_BRAND.tenant_uuid: {}, TENANT_LP.tenant_uuid: TenantHierarchy( company_brand=Tenant( tenant_uuid=UUID("9df8b5f9-f17b-45bf-a254-40b4deb52017"), tenant_type=TenantType.TENANT_TYPE_COMPANY_BRAND, ) ), }, [TENANT_SUBACCOUNT], [TENANT_ACCOUNT, TENANT_SUBACCOUNT, TENANT_LP, TENANT_COMPANY_BRAND], DEFAULT_TENANT_TYPES_FOR_VALIDATION_CHECKS, id="Should raise an error when the proxy does not fetch a hierarchy for the subaccount tenant", # noqa: E501 ), pytest.param( [TENANT_ACCOUNT, TENANT_COMPANY_BRAND, TENANT_PARENT_COMPANY], { TENANT_ACCOUNT.tenant_uuid: TenantHierarchy( company_brand=Tenant( tenant_uuid=UUID("9df8b5f9-f17b-45bf-a254-40b4deb52017"), tenant_type=TenantType.TENANT_TYPE_COMPANY_BRAND, ) ), TENANT_COMPANY_BRAND.tenant_uuid: {}, }, [TENANT_PARENT_COMPANY], [TENANT_ACCOUNT, TENANT_COMPANY_BRAND, TENANT_PARENT_COMPANY], DEFAULT_TENANT_TYPES_FOR_VALIDATION_CHECKS, id="Should raise an error when the proxy does not fetch a hierarchy for the parent_company tenant", # noqa: E501 ), pytest.param( [TENANT_ACCOUNT, TENANT_SUBACCOUNT, TENANT_LP, TENANT_COMPANY_BRAND], {}, [TENANT_ACCOUNT], [TENANT_ACCOUNT], [TenantType.TENANT_TYPE_ACCOUNT], id="Should only validate account tenant type.", ), pytest.param( [TENANT_ACCOUNT, TENANT_SUBACCOUNT, TENANT_LP, TENANT_COMPANY_BRAND], {}, [TENANT_SUBACCOUNT], [TENANT_SUBACCOUNT], [TenantType.TENANT_TYPE_SUBACCOUNT], id="Should only validate subaccount tenant type.", ), pytest.param( [TENANT_ACCOUNT, TENANT_SUBACCOUNT, TENANT_LP, TENANT_COMPANY_BRAND], {}, [TENANT_COMPANY_BRAND], [TENANT_COMPANY_BRAND], [TenantType.TENANT_TYPE_COMPANY_BRAND], id="Should only validate company brand tenant type.", ), ], ) @patch("pdp.proxies.tenant_validators.MultiTenantProxy") async def test_has_valid_tenants__raises_error( mock_multi_tenant_proxy_cls: MagicMock, tenants: List[Tenant], mock_gather_result: Dict[UUID, TenantHierarchy], expected_invalid_tenants: List[Tenant], expected_proxy_tenants: List[Tenant], tenant_types_to_validate: List[TenantType], mock_ows_participant_client: OwsParticipantClient, mock_ows_account_client: OwsAccountClient, mock_redis_connector: RedisConnector, ) -> None: """Test assert_valid_tenants raises an error when invalid tenants are found.""" mock_proxy = AsyncMock() mock_multi_tenant_proxy_cls.return_value = mock_proxy mock_proxy.gather_tenant_hierarchies = AsyncMock( return_value=mock_gather_result, ) with pytest.raises(InvalidTenantsError) as exc_info: await assert_valid_tenants( tenants=tenants, redis_connector=mock_redis_connector, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, tenant_types_to_validate=tenant_types_to_validate, ) assert exc_info.value.invalid_tenants == expected_invalid_tenants assert exc_info.value.status_code == 400 mock_multi_tenant_proxy_cls.assert_called_with( tenants=expected_proxy_tenants, redis_client=mock_redis_connector, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, ) @pytest.mark.parametrize( "tenants, mock_gather_result, expected_proxy_tenants, tenant_types_to_validate", [ pytest.param( [TENANT_ACCOUNT, TENANT_SUBACCOUNT, TENANT_LP, TENANT_COMPANY_BRAND], { TENANT_ACCOUNT.tenant_uuid: TenantHierarchy( company_brand=Tenant( tenant_uuid=UUID("9df8b5f9-f17b-45bf-a254-40b4deb52017"), tenant_type=TenantType.TENANT_TYPE_COMPANY_BRAND, ) ), TENANT_SUBACCOUNT.tenant_uuid: TenantHierarchy( account=TENANT_ACCOUNT, company_brand=Tenant( tenant_uuid=UUID("9df8b5f9-f17b-45bf-a254-40b4deb52017"), tenant_type=TenantType.TENANT_TYPE_COMPANY_BRAND, ), ), TENANT_LP.tenant_uuid: TenantHierarchy( account=TENANT_ACCOUNT, company_brand=Tenant( tenant_uuid=UUID("9df8b5f9-f17b-45bf-a254-40b4deb52017"), tenant_type=TenantType.TENANT_TYPE_COMPANY_BRAND, ), ), TENANT_COMPANY_BRAND.tenant_uuid: {}, }, [TENANT_ACCOUNT, TENANT_SUBACCOUNT, TENANT_LP, TENANT_COMPANY_BRAND], DEFAULT_TENANT_TYPES_FOR_VALIDATION_CHECKS, id="Should return True if accounts and subaccounts are valid.", ), pytest.param( [TENANT_ACCOUNT, TENANT_SUBACCOUNT, TENANT_LP, TENANT_COMPANY_BRAND], { TENANT_ACCOUNT.tenant_uuid: TenantHierarchy( company_brand=Tenant( tenant_uuid=UUID("9df8b5f9-f17b-45bf-a254-40b4deb52017"), tenant_type=TenantType.TENANT_TYPE_COMPANY_BRAND, ) ), }, [TENANT_ACCOUNT], [TenantType.TENANT_TYPE_ACCOUNT], id="Should only validate account tenant type.", ), pytest.param( [TENANT_ACCOUNT, TENANT_SUBACCOUNT, TENANT_LP, TENANT_COMPANY_BRAND], { TENANT_SUBACCOUNT.tenant_uuid: TenantHierarchy( account=TENANT_ACCOUNT, company_brand=Tenant( tenant_uuid=UUID("9df8b5f9-f17b-45bf-a254-40b4deb52017"), tenant_type=TenantType.TENANT_TYPE_COMPANY_BRAND, ), ) }, [TENANT_SUBACCOUNT], [TenantType.TENANT_TYPE_SUBACCOUNT], id="Should only validate subaccount tenant type.", ), pytest.param( [TENANT_ACCOUNT, TENANT_SUBACCOUNT, TENANT_LP, TENANT_COMPANY_BRAND], {TENANT_COMPANY_BRAND.tenant_uuid: {}}, [TENANT_COMPANY_BRAND], [TenantType.TENANT_TYPE_COMPANY_BRAND], id="Should only validate company brand tenant type.", ), pytest.param( [ TENANT_ACCOUNT, TENANT_SUBACCOUNT, TENANT_LP, TENANT_COMPANY_BRAND, TENANT_PARENT_COMPANY, ], {TENANT_PARENT_COMPANY.tenant_uuid: {}}, [TENANT_PARENT_COMPANY], [TenantType.TENANT_TYPE_PARENT_COMPANY], id="Should only validate parent company tenant type.", ), ], ) @patch("pdp.proxies.tenant_validators.MultiTenantProxy") async def test_has_valid_tenants( mock_multi_tenant_proxy_cls: MagicMock, tenants: List[Tenant], mock_gather_result: Dict[UUID, TenantHierarchy], expected_proxy_tenants: List[Tenant], tenant_types_to_validate: List[TenantType], mock_ows_participant_client: OwsParticipantClient, mock_ows_account_client: OwsAccountClient, mock_redis_connector: RedisConnector, ) -> None: """Test assert_valid_tenants returns True when all tenants are valid.""" mock_proxy = AsyncMock() mock_multi_tenant_proxy_cls.return_value = mock_proxy mock_proxy.gather_tenant_hierarchies = AsyncMock( return_value=mock_gather_result, ) result = await assert_valid_tenants( tenants=tenants, redis_connector=mock_redis_connector, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, tenant_types_to_validate=tenant_types_to_validate, ) assert result mock_multi_tenant_proxy_cls.assert_called_with( tenants=expected_proxy_tenants, redis_client=mock_redis_connector, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, ) @pytest.mark.parametrize( "tenants, tenant_types_to_validate", [ pytest.param( [TENANT_ACCOUNT, TENANT_SUBACCOUNT], [TenantType.TENANT_TYPE_LABEL_PARTICIPANT], id="tenant_types_to_validate has only LP, the function should not validate any tenants because the input does not include an LP.", # noqa: E501 ), pytest.param( [], DEFAULT_TENANT_TYPES_FOR_VALIDATION_CHECKS, id="returns True when tenants is empty.", ), ], ) @patch("pdp.proxies.tenant_validators.MultiTenantProxy") async def test_has_valid_tenants__no_tenants( mock_multi_tenant_proxy_cls: MagicMock, tenants: List[Tenant], tenant_types_to_validate: List[TenantType], mock_ows_participant_client: OwsParticipantClient, mock_ows_account_client: OwsAccountClient, mock_redis_connector: RedisConnector, ) -> None: """Test assert_valid_tenants returns True when tenants is empty.""" mock_proxy = AsyncMock() mock_multi_tenant_proxy_cls.return_value = mock_proxy mock_proxy.gather_tenant_hierarchies = AsyncMock( return_value={}, ) result = await assert_valid_tenants( tenants=tenants, redis_connector=mock_redis_connector, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, tenant_types_to_validate=tenant_types_to_validate, ) assert result mock_multi_tenant_proxy_cls.assert_not_called() @patch("pdp.proxies.tenant_validators.MultiTenantProxy") async def test_has_valid_tenantroles( mock_multi_tenant_proxy_cls: MagicMock, mock_ows_participant_client: OwsParticipantClient, mock_ows_account_client: OwsAccountClient, mock_redis_connector: RedisConnector, ) -> None: """Test assert_valid_tenants returns True when all TenantRoles are valid.""" mock_proxy = AsyncMock() mock_multi_tenant_proxy_cls.return_value = mock_proxy mock_proxy.gather_tenant_hierarchies = AsyncMock( return_value={ TENANT_ROLE_ACCOUNT.tenant_uuid: TenantHierarchy( company_brand=Tenant( tenant_uuid=UUID("9df8b5f9-f17b-45bf-a254-40b4deb52017"), tenant_type=TenantType.TENANT_TYPE_COMPANY_BRAND, ) ), TENANT_ROLE_SUBACCOUNT.tenant_uuid: TenantHierarchy( account=TENANT_ACCOUNT, company_brand=Tenant( tenant_uuid=UUID("9df8b5f9-f17b-45bf-a254-40b4deb52017"), tenant_type=TenantType.TENANT_TYPE_COMPANY_BRAND, ), ), TENANT_ROLE_LP.tenant_uuid: TenantHierarchy( account=TENANT_ACCOUNT, company_brand=Tenant( tenant_uuid=UUID("9df8b5f9-f17b-45bf-a254-40b4deb52017"), tenant_type=TenantType.TENANT_TYPE_COMPANY_BRAND, ), ), }, ) result = await assert_valid_tenants( tenants=[TENANT_ROLE_ACCOUNT, TENANT_ROLE_SUBACCOUNT, TENANT_ROLE_LP], redis_connector=mock_redis_connector, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, tenant_types_to_validate=DEFAULT_TENANT_TYPES_FOR_VALIDATION_CHECKS, ) assert result mock_multi_tenant_proxy_cls.assert_called_with( tenants=[TENANT_ROLE_ACCOUNT, TENANT_ROLE_SUBACCOUNT, TENANT_ROLE_LP], redis_client=mock_redis_connector, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, ) @patch("pdp.proxies.tenant_validators.MultiTenantProxy") async def test_has_valid_tenants__tenantroles__raises_error( mock_multi_tenant_proxy_cls: MagicMock, mock_ows_participant_client: OwsParticipantClient, mock_ows_account_client: OwsAccountClient, mock_redis_connector: RedisConnector, ) -> None: """Test assert_valid_tenants raises an error for a TenantRoles type object.""" mock_proxy = AsyncMock() mock_multi_tenant_proxy_cls.return_value = mock_proxy mock_proxy.gather_tenant_hierarchies = AsyncMock( return_value={ TENANT_ROLE_ACCOUNT.tenant_uuid: TenantHierarchy( company_brand=Tenant( tenant_uuid=UUID("9df8b5f9-f17b-45bf-a254-40b4deb52017"), tenant_type=TenantType.TENANT_TYPE_COMPANY_BRAND, ) ) }, ) with pytest.raises(InvalidTenantsError) as exc_info: await assert_valid_tenants( tenants=[TENANT_ROLE_ACCOUNT, TENANT_ROLE_SUBACCOUNT, TENANT_ROLE_LP], redis_connector=mock_redis_connector, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, tenant_types_to_validate=DEFAULT_TENANT_TYPES_FOR_VALIDATION_CHECKS, ) assert exc_info.value.invalid_tenants == [TENANT_ROLE_SUBACCOUNT, TENANT_ROLE_LP] assert exc_info.value.status_code == 400 mock_multi_tenant_proxy_cls.assert_called_with( tenants=[TENANT_ROLE_ACCOUNT, TENANT_ROLE_SUBACCOUNT, TENANT_ROLE_LP], redis_client=mock_redis_connector, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, ) @pytest.mark.parametrize( "mock_request, should_trigger_validation", [ pytest.param( AttachDetachRolesRequest( tenant_type=TenantType.TENANT_TYPE_ACCOUNT, tenant_uuid=uuid.uuid4(), roles_to_attach=[Role(role="test_role")], roles_to_detach=[], ), True, id="An account tenant with roles to attach should be validated", ), pytest.param( AttachDetachRolesRequest( tenant_type=TenantType.TENANT_TYPE_ACCOUNT, tenant_uuid=uuid.uuid4(), roles_to_attach=[], roles_to_detach=[], ), False, id="An Account tenant with no roles should not be validated", ), pytest.param( AttachDetachRolesRequest( tenant_type=TenantType.TENANT_TYPE_ACCOUNT, tenant_uuid=uuid.uuid4(), roles_to_attach=[Role(role="test_role")], roles_to_detach=[Role(role="another_role")], ), True, id="An Account tenant with both roles to attach and detach should be validated", # noqa: E501 ), pytest.param( AttachDetachRolesRequest( tenant_type=TenantType.TENANT_TYPE_ACCOUNT, tenant_uuid=uuid.uuid4(), roles_to_attach=[], roles_to_detach=[Role(role="another_role")], ), True, id="An Account tenant with only roles to detach should be validated", ), pytest.param( AttachDetachRolesRequest( tenant_type=TenantType.TENANT_TYPE_ACCOUNT, tenant_uuid=uuid.uuid4(), roles_to_attach=[Role(role="test_role")], roles_to_detach=[], ), True, id="An Account tenant with only roles to attach should be validated", ), ], ) @patch("pdp.proxies.tenant_validators.assert_valid_tenants") async def test_assert_valid_attach_detach_request( mock_assert_valid_tenants: MagicMock, mock_request: AttachDetachRolesRequest, should_trigger_validation: bool, mock_ows_participant_client: OwsParticipantClient, mock_ows_account_client: OwsAccountClient, mock_redis_connector: RedisConnector, ) -> None: """Test assert_valid_attach_detach_request.""" mock_assert_valid_tenants.return_value = True mock_raw_request = AsyncMock() mock_raw_request.json.return_value = mock_request.model_dump() await assert_valid_attach_detach_request( request=mock_raw_request, redis_connector=mock_redis_connector, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, ) if should_trigger_validation: mock_assert_valid_tenants.assert_called_with( tenants=[ Tenant( tenant_type=mock_request.tenant_type, tenant_uuid=mock_request.tenant_uuid, ) ], redis_connector=mock_redis_connector, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, ) else: mock_assert_valid_tenants.assert_not_called() @pytest.mark.parametrize( "bad_request", [ pytest.param({}, id="empty request"), pytest.param( { "tenant_uuid": "not a uuid", "tenant_type": "account", "roles_to_attach": [], "roles_to_detach": [{"role": "mock"}], }, id="not a tenant uuid", ), pytest.param( { "tenant_uuid": "573d0372-7f2f-48a6-8deb-c9a6558f9549", "tenant_type": "account", "roles_to_attach": [], "roles_to_detach": ["mock"], }, id="bad Role", ), ], ) @patch("pdp.proxies.tenant_validators.assert_valid_tenants") async def test_assert_valid_attach_detach_request_bad_request( mock_assert_valid_tenants: AsyncMock, bad_request: dict[str, Any], mock_ows_participant_client: OwsParticipantClient, mock_ows_account_client: OwsAccountClient, mock_redis_connector: RedisConnector, ) -> None: """Test assert_valid_attach_detach_request.""" mock_raw_request = AsyncMock() mock_raw_request.json.return_value = bad_request with pytest.raises(InvalidAttachDetachRolesRequest): await assert_valid_attach_detach_request( request=mock_raw_request, redis_connector=mock_redis_connector, ows_account_client=mock_ows_account_client, ows_participant_client=mock_ows_participant_client, ) mock_assert_valid_tenants.assert_not_called() TEST_ACCOUNT_UUID_01 = "1727f13d-95b4-4172-95e7-e67f6d8587f0" TEST_ACCOUNT_UUID_02 = "397a8321-b84a-4e5a-acd9-7e572b226492"