"""Integration tests for PUT /{identity_uuid}/tenant/{tenant_uuid}/attach-and-detach/roles/.""" # noqa import uuid from typing import Any, Dict import pytest import requests from mypy_boto3_dynamodb import DynamoDBClient from pdp.config import DYNAMODB_TABLE_IDENTITY from pdp.fastapi.schemas.identity import AttachDetachRolesRequest from pdp.models.identity import HASH_KEY as IDENTITY_HASH_KEY from pdp.models.identity import RANGE_KEY as IDENTITY_RANGE_KEY from tests.integration import config, utils from tests.integration.conftest import seed_test_pp_identity def test_attach_detach_roles_no_roles_left( default_boto_client: DynamoDBClient, bearer_token_pdptest_rap_admin_user: str, bearer_token_pdptest_user_identity_uuid: str, ) -> None: """Verify that when no roles are left to detach, the DynamoDB item is deleted.""" # Seed the pdp_test_user_uuid identity with a tenant and a role. tenant_uuid = "573d0372-7f2f-48a6-8deb-c9a6558f9549" seed_test_pp_identity( default_boto_client, bearer_token_pdptest_user_identity_uuid, str(tenant_uuid), "audience_development_client", tenant_type="account", ) # Fetch pdp_test_user_uuid's seeded tenants and verify the tenant was added. seeded_global_tenants = default_boto_client.query( TableName=DYNAMODB_TABLE_IDENTITY, ExpressionAttributeValues={":hash_key": {"S": utils.PDP_TEST_USER_UUID}}, KeyConditionExpression=f"{IDENTITY_HASH_KEY} = :hash_key", ) seeded_tenants = set( [ seeded_tenant["tenant_uuid"]["S"] for seeded_tenant in seeded_global_tenants.get("Items", []) ] ) assert seeded_tenants == {tenant_uuid}, "Seeded tenant not found." # Detach the only role from the tenant, which should delete the identity/tenant item. # noqa: E501 url = f"{config.QA_BASE_URL}/identity/{utils.PDP_TEST_USER_UUID}/tenant/{tenant_uuid}/attach-and-detach/roles/" # noqa: E501 response = requests.put( url, headers={"Authorization": f"Bearer {bearer_token_pdptest_rap_admin_user}"}, json={ "tenant_uuid": tenant_uuid, "tenant_type": "account", "roles_to_attach": [], "roles_to_detach": [{"role": "audience_development_client"}], }, ) assert response.status_code == 200, response.text # Fetch pdp_test_user_uuid's remaining tenants and see that the tenant was deleted. actual_global_remaining = default_boto_client.query( TableName=DYNAMODB_TABLE_IDENTITY, ExpressionAttributeValues={":hash_key": {"S": utils.PDP_TEST_USER_UUID}}, KeyConditionExpression=f"{IDENTITY_HASH_KEY} = :hash_key", ) remaining_tenants = set( [ remaining_tenant["tenant_uuid"]["S"] for remaining_tenant in actual_global_remaining.get("Items", []) ] ) assert tenant_uuid not in remaining_tenants # Verify tombstone records tombstone_records = utils.fetch_tombstone_records(default_boto_client) assert len(tombstone_records) == 1 assert set(tombstone_records.keys()) == set([tenant_uuid]) assert all( [ r.get("identity_uuid") .get("S", "") .startswith(f"TOMBSTONE:{utils.PDP_TEST_USER_UUID}") ] for r in tombstone_records.values() ) @pytest.mark.parametrize( "body, expected_status, expected_response", [ pytest.param( AttachDetachRolesRequest( **{ "tenant_uuid": "573d0372-7f2f-48a6-8deb-c9a6558f9549", "tenant_type": "account", "roles_to_attach": [{"role": "songwhip_read"}], "roles_to_detach": [], } ), 200, { "cursor": {"cursor": None, "shorthand": None}, "errors": {}, "tenants": { "573d0372-7f2f-48a6-8deb-c9a6558f9549": { "tenant_uuid": "573d0372-7f2f-48a6-8deb-c9a6558f9549", "tenant_type": "account", "roles": [ {"role": "audience_development_client"}, {"role": "songwhip_read"}, ], }, }, }, id="songwhip_read role should be attached.", ), pytest.param( AttachDetachRolesRequest( **{ "tenant_uuid": "573d0372-7f2f-48a6-8deb-c9a6558f9549", "tenant_type": "account", "roles_to_attach": [], "roles_to_detach": [{"role": "audience_development_client"}], } ), 200, { "cursor": {"cursor": None, "shorthand": None}, "errors": {}, "tenants": { "573d0372-7f2f-48a6-8deb-c9a6558f9549": { "tenant_uuid": "573d0372-7f2f-48a6-8deb-c9a6558f9549", "tenant_type": "account", "roles": [], }, }, }, id="audience_development_client role should be detached", ), pytest.param( AttachDetachRolesRequest( **{ "tenant_uuid": "573d0372-7f2f-48a6-8deb-c9a6558f9549", "tenant_type": "account", "roles_to_attach": [{"role": "songwhip_read"}], "roles_to_detach": [{"role": "audience_development_client"}], } ), 200, { "cursor": {"cursor": None, "shorthand": None}, "errors": {}, "tenants": { "573d0372-7f2f-48a6-8deb-c9a6558f9549": { "tenant_uuid": "573d0372-7f2f-48a6-8deb-c9a6558f9549", "tenant_type": "account", "roles": [{"role": "songwhip_read"}], }, }, }, id="both attach and detach operations should be executed", ), pytest.param( AttachDetachRolesRequest( **{ "tenant_uuid": "41edd831-fb7d-11ef-8476-0ef8c77b7565", "tenant_type": "account", "roles_to_attach": [{"role": "songwhip_read"}], "roles_to_detach": [{"role": "audience_development_client"}], } ), 200, { "cursor": {"cursor": None, "shorthand": None}, "errors": {}, "tenants": { "41edd831-fb7d-11ef-8476-0ef8c77b7565": { "tenant_uuid": "41edd831-fb7d-11ef-8476-0ef8c77b7565", "tenant_type": "account", "roles": [{"role": "songwhip_read"}], }, }, }, id="both attach and detach operations should be executed with UUIDv1 tenant uuid", # noqa: E501 ), pytest.param( AttachDetachRolesRequest( **{ "tenant_uuid": "573d0372-7f2f-48a6-8deb-c9a6558f9549", "tenant_type": "company_brand", "roles_to_attach": [], "roles_to_detach": [], } ), 400, { "code": "bad_request", "message": "Tenant_type does not match. Identity: {identity_uuid}, Tenant: 573d0372-7f2f-48a6-8deb-c9a6558f9549, Given tenant_type: company_brand, Existing tenant_type: TenantType.TENANT_TYPE_ACCOUNT", # noqa: E501 }, id="mismatched tenant type should cause error", ), ], ) def test_attach_detach_roles_by_identity_tenant( body: AttachDetachRolesRequest, expected_status: int, expected_response: Dict[str, Any], default_boto_client: DynamoDBClient, bearer_token_pdptest_user: str, bearer_token_pdptest_user_identity_uuid: str, ) -> None: """PUT /identity//tenant//attach-and-detach/roles/ handler.""" # noqa: E501 tenant_uuid = body.tenant_uuid seed_test_pp_identity( default_boto_client, bearer_token_pdptest_user_identity_uuid, str(tenant_uuid), "audience_development_client", tenant_type="account", ) token_uuid = bearer_token_pdptest_user_identity_uuid url = f"{config.QA_BASE_URL}/identity/{token_uuid}/tenant/{tenant_uuid}/attach-and-detach/roles/" # noqa: E501 response = requests.put( url, headers={"Authorization": f"Bearer {bearer_token_pdptest_user}"}, json=body.model_dump(mode="json"), ) assert response.status_code == expected_status, f"Response: {response.text}" if response.status_code == 200: assert response.json() == expected_response, response.text else: expected_response["message"] = expected_response["message"].format( identity_uuid=token_uuid, ) assert response.json() == expected_response, response.text @pytest.mark.parametrize( "tenant_type", [ pytest.param("account"), pytest.param("subaccount"), pytest.param("company_brand"), pytest.param("parent_company"), pytest.param("label_participant"), ], ) def test_attach_detach_roles_by_identity_tenant__invalid_tenant( tenant_type: str, default_boto_client: DynamoDBClient, bearer_token_pdptest_user: str, bearer_token_pdptest_user_identity_uuid: str, ) -> None: """Test that the attach-and-detach endpoint returns a 400 error for an invalid tenant uuid.""" # noqa: E501 tenant_uuid = uuid.UUID("ff3604b6-fde3-11ef-b7ed-4a2888760683") body = AttachDetachRolesRequest( **{ "tenant_uuid": "ff3604b6-fde3-11ef-b7ed-4a2888760683", "tenant_type": tenant_type, "roles_to_attach": [{"role": "songwhip_read"}], "roles_to_detach": [], } ) expected_status = 400 expected_response = { "code": "bad_request", "message": f"Found invalid tenants: [TenantType.TENANT_TYPE_{tenant_type.upper()}#{tenant_uuid}]", # noqa: E501 } seed_test_pp_identity( default_boto_client, bearer_token_pdptest_user_identity_uuid, str(tenant_uuid), "audience_development_client", tenant_type="account", ) token_uuid = bearer_token_pdptest_user_identity_uuid url = f"{config.QA_BASE_URL}/identity/{token_uuid}/tenant/{tenant_uuid}/attach-and-detach/roles/" # noqa: E501 response = requests.put( url, headers={"Authorization": f"Bearer {bearer_token_pdptest_user}"}, json=body.model_dump(mode="json"), ) assert response.status_code == expected_status, f"Response: {response.text}" assert response.json() == expected_response, response.text def test_attach_detach_roles_by_identity_tenant_not_rap_admin( default_boto_client: DynamoDBClient, bearer_token_pdptest_not_rap_admin_user: str, ) -> None: """Verify that a non-rap admin user is not authorized to make this request.""" bearer_token_identity_uuid = utils.get_bearer_token_identity_uuid( bearer_token_pdptest_not_rap_admin_user ) url = f"{config.QA_BASE_URL}/identity/{utils.PDP_TEST_USER_UUID}/tenant/573d0372-7f2f-48a6-8deb-c9a6558f9549/attach-and-detach/roles/" # noqa: E501 response = requests.put( url, headers={"Authorization": f"Bearer {bearer_token_pdptest_not_rap_admin_user}"}, json={ "tenant_uuid": "573d0372-7f2f-48a6-8deb-c9a6558f9549", "tenant_type": "account", "roles_to_attach": [{"role": "songwhip_read"}], "roles_to_detach": [], }, ) assert response.status_code == 403, ( "NOT RAP Admin user shouldn't be authorized to attach-and-detach." # noqa: E501 ) assert response.json() == { "code": "bad_request", "message": f"Principal {bearer_token_identity_uuid} not authorized to attach_and_detach_role on identity {utils.PDP_TEST_USER_UUID}", # noqa: E501 }, "NOT RAP Admin user shouldn't be authorized to attach-and-detach." def test_attach_detach_roles_by_identity_tenant_rap_admin( default_boto_client: DynamoDBClient, bearer_token_pdptest_rap_admin_user: str, ) -> None: """Verify that a rap admin user is authorized to administer a user under a tenant that the rap admin has administration privileges for.""" tenant_uuid = "573d0372-7f2f-48a6-8deb-c9a6558f9549" url = f"{config.QA_BASE_URL}/identity/{utils.PDP_TEST_USER_UUID}/tenant/{tenant_uuid}/attach-and-detach/roles/" # noqa: E501 response = requests.put( url, headers={"Authorization": f"Bearer {bearer_token_pdptest_rap_admin_user}"}, json={ "tenant_uuid": tenant_uuid, "tenant_type": "account", "roles_to_attach": [{"role": "songwhip_read"}], "roles_to_detach": [], }, ) assert response.status_code == 200, response.text def test_attach_detach_roles_by_identity_tenant_d3_rap_admin( default_boto_client: DynamoDBClient, identity_uuid: uuid.UUID, bearer_token_pdptest_d3_rap_admin_user: str, ) -> None: """Verify that a D3 rap admin test user is authorized to administer users on a subaccount that is owned by the D3 that the rap admin can administer. The user does not have to be on the subaccount tenant already.""" # This subaccount uuid is for the Fat Wreck Records Text Tenant # This is a tenant owned by the Distribution Inc test D3. # The D3 rap admin test user has admin access to the Distribution Inc tenant. subaccount_tenant_uuid = "ccd55b40-e10e-4059-97f0-aec665c24ec7" url = f"{config.QA_BASE_URL}/identity/{identity_uuid}/tenant/{subaccount_tenant_uuid}/attach-and-detach/roles/" # noqa: E501 response = requests.put( url, headers={"Authorization": f"Bearer {bearer_token_pdptest_d3_rap_admin_user}"}, json={ "tenant_uuid": subaccount_tenant_uuid, "tenant_type": "subaccount", "roles_to_attach": [{"role": "songwhip_read"}], "roles_to_detach": [], }, ) assert response.status_code == 200, response.text def test_attach_detach_roles_by_identity_tenant_d3_rap_admin_existing_identity( default_boto_client: DynamoDBClient, identity_uuid: uuid.UUID, bearer_token_pdptest_d3_rap_admin_user: str, ) -> None: """Verify that a D3 rap admin test user is authorized to administer users on a subaccount that is owned by the D3 that the rap admin can administer. The user can be on the subaccount tenant already.""" # This subaccount uuid is for the Fat Wreck Records Text Tenant # This is a tenant owned by the Distribution Inc test D3. # The D3 rap admin test user has admin access to the Distribution Inc tenant. subaccount_tenant_uuid = "ccd55b40-e10e-4059-97f0-aec665c24ec7" seed_test_pp_identity( default_boto_client, str(identity_uuid), str(subaccount_tenant_uuid), "audience_development_client", tenant_type="subaccount", ) url = f"{config.QA_BASE_URL}/identity/{identity_uuid}/tenant/{subaccount_tenant_uuid}/attach-and-detach/roles/" # noqa: E501 response = requests.put( url, headers={"Authorization": f"Bearer {bearer_token_pdptest_d3_rap_admin_user}"}, json={ "tenant_uuid": subaccount_tenant_uuid, "tenant_type": "subaccount", "roles_to_attach": [{"role": "songwhip_read"}], "roles_to_detach": [], }, ) assert response.status_code == 200, response.text def test_attach_detach_roles_by_identity_tenant_d3_rap_admin_wrong_subaccount( default_boto_client: DynamoDBClient, identity_uuid: uuid.UUID, bearer_token_pdptest_d3_rap_admin_user: str, ) -> None: """Verify that a D3 rap admin test user is NOT authorized to administer users on tenants that are under a subaccount that is NOT owned by the D3 that the rap admin can administer.""" # This subaccount uuid is for the Luondu Music subaccount tenant # This tenant is NOT owned by the Distribution Inc test D3. # The D3 rap admin test user has admin access to the Distribution Inc tenant. subaccount_tenant_uuid = "e369934b-36fb-4ca9-b2bd-4adecc6a8ccc" url = f"{config.QA_BASE_URL}/identity/{identity_uuid}/tenant/{subaccount_tenant_uuid}/attach-and-detach/roles/" # noqa: E501 response = requests.put( url, headers={"Authorization": f"Bearer {bearer_token_pdptest_d3_rap_admin_user}"}, json={ "tenant_uuid": subaccount_tenant_uuid, "tenant_type": "subaccount", "roles_to_attach": [{"role": "songwhip_read"}], "roles_to_detach": [], }, ) assert response.status_code == 403, f"Response: {response.text}" assert ( response.json()["message"] == f"Principal c7ebf40e-f09b-4292-a997-59d25668a72e not authorized to attach_and_detach_role on identity {identity_uuid}" # noqa: E501 ) def test_attach_detach_roles_by_identity_tenant_auto_detach_child_roles( dynamodb_table_identity: str, default_boto_client: DynamoDBClient, bearer_token_pdptest_user: str, bearer_token_pdptest_user_identity_uuid: str, ) -> None: """Test child roles are automatically detached when a parent role is detached.""" tenant_uuid = "573d0372-7f2f-48a6-8deb-c9a6558f9549" default_boto_client.put_item( TableName=dynamodb_table_identity, Item={ IDENTITY_HASH_KEY: {"S": str(bearer_token_pdptest_user_identity_uuid)}, IDENTITY_RANGE_KEY: {"S": str(tenant_uuid)}, "version": {"S": "1"}, "tenant_type": {"S": "account"}, "roles": { "L": [ {"M": {"role": {"S": "fansifter_can_view_fan_data"}}}, {"M": {"role": {"S": "fansifter_can_create_ad_reports"}}}, {"M": {"role": {"S": "fansifter_can_share_ad_campaign_audiences"}}}, {"M": {"role": {"S": "fansifter_can_connect_ad_accounts"}}}, {"M": {"role": {"S": "fansifter_can_view_email_campaigns"}}}, {"M": {"role": {"S": "fansifter_can_create_email_campaigns"}}}, {"M": {"role": {"S": "fansifter_can_view_sms_campaigns"}}}, {"M": {"role": {"S": "fansifter_can_create_sms_campaigns"}}}, {"M": {"role": {"S": "songwhip_read"}}}, ] }, }, ) url = f"{config.QA_BASE_URL}/identity/{bearer_token_pdptest_user_identity_uuid}/roles/" # noqa: E501 confirm_seed_response = requests.get( url, headers={"Authorization": f"Bearer {bearer_token_pdptest_user}"}, ) assert confirm_seed_response.status_code == 200, ( f"Response: {confirm_seed_response.text}" ) assert confirm_seed_response.json() == { "cursor": {"cursor": None, "shorthand": None}, "errors": {}, "tenants": { str(tenant_uuid): { "tenant_uuid": str(tenant_uuid), "tenant_type": "account", "roles": [ {"role": "fansifter_can_view_fan_data"}, {"role": "fansifter_can_create_ad_reports"}, {"role": "fansifter_can_share_ad_campaign_audiences"}, {"role": "fansifter_can_connect_ad_accounts"}, {"role": "fansifter_can_view_email_campaigns"}, {"role": "fansifter_can_create_email_campaigns"}, {"role": "fansifter_can_view_sms_campaigns"}, {"role": "fansifter_can_create_sms_campaigns"}, {"role": "songwhip_read"}, ], }, }, } url = f"{config.QA_BASE_URL}/identity/{bearer_token_pdptest_user_identity_uuid}/tenant/{tenant_uuid}/attach-and-detach/roles/" # noqa: E501 response = requests.put( url, headers={"Authorization": f"Bearer {bearer_token_pdptest_user}"}, json={ "tenant_uuid": str(tenant_uuid), "tenant_type": "account", "roles_to_attach": [], "roles_to_detach": [{"role": "fansifter_can_view_fan_data"}], }, ) assert response.status_code == 200, f"Response: {response.text}" assert response.json() == { "cursor": {"cursor": None, "shorthand": None}, "errors": {}, "tenants": { str(tenant_uuid): { "tenant_uuid": str(tenant_uuid), "tenant_type": "account", "roles": [{"role": "songwhip_read"}], }, }, } def test_attach_detach_roles_by_identity_tenant_keep_supported_roles( dynamodb_table_identity: str, default_boto_client: DynamoDBClient, bearer_token_pdptest_user: str, bearer_token_pdptest_user_identity_uuid: str, ) -> None: """Test only supported roles are attached.""" tenant_uuid = "573d0372-7f2f-48a6-8deb-c9a6558f9549" url = f"{config.QA_BASE_URL}/identity/{bearer_token_pdptest_user_identity_uuid}/tenant/{tenant_uuid}/attach-and-detach/roles/" # noqa: E501 response = requests.put( url, headers={"Authorization": f"Bearer {bearer_token_pdptest_user}"}, json={ "tenant_uuid": str(tenant_uuid), "tenant_type": "account", "roles_to_attach": [ {"role": "hallucinated_role"}, {"role": "songwhip_read"}, ], "roles_to_detach": [], }, ) assert response.status_code == 200, f"Response: {response.text}" assert response.json() == { "cursor": {"cursor": None, "shorthand": None}, "errors": {}, "tenants": { str(tenant_uuid): { "tenant_uuid": str(tenant_uuid), "tenant_type": "account", "roles": [{"role": "songwhip_read"}], }, }, }