"""Test the POST /v2/identities endpoint.""" import random import pytest import requests from tests.integration import config, utils account_base_roles = [ 'COLLABORATORS_BASE_ROLE', 'INSIGHTS_BASE_ROLE', 'CUSTOMER_ACCOUNTING_BASE_ROLE', 'WORKSTATION_ANALYTICS_BASE_ROLE', 'WORKSTATION_CATALOG_ROLE', 'WORKSTATION_MARKETING_ROLE', 'WORKSTATION_ADVERTISING_ROLE', ] subaccount_base_roles = [ 'INSIGHTS_BASE_ROLE', 'CUSTOMER_ACCOUNTING_BASE_ROLE', 'WORKSTATION_CATALOG_ROLE', ] @pytest.mark.parametrize( ('test_roles', 'expected_label_profile_roles'), [ # Both SETTINGS_BASE_ROLE + WORKSTATION_ADMIN_ROLE pytest.param( ['SETTINGS_BASE_ROLE', 'WORKSTATION_ADMIN_ROLE'], ['administrator', 'advertising', 'analytics', 'catalog', 'marketing'], id='both_settings_and_workstation_admin', ), # Only SETTINGS_BASE_ROLE pytest.param( ['SETTINGS_BASE_ROLE'], ['advertising', 'analytics', 'catalog', 'marketing'], id='only_settings', ), # Only WORKSTATION_ADMIN_ROLE pytest.param( ['WORKSTATION_ADMIN_ROLE'], ['administrator', 'advertising', 'analytics', 'catalog', 'marketing'], id='only_workstation_admin', ), # Neither role pytest.param( [], ['advertising', 'analytics', 'catalog', 'marketing'], id='neither_settings_nor_workstation_admin', ), ], ) def test_create_identity_v2_account( bearer_token_user_with_vendor_star: str, test_roles: list[str], expected_label_profile_roles: list[str], ) -> None: """Integration test for create_identity_v2 endpoint with account tenant.""" roles = account_base_roles + test_roles expected_profiles = { 'InsightsProfile': ['analytics'], 'LabelProfile': expected_label_profile_roles, 'SettingsProfile': [], 'CollaboratorsProfile': ['royalties'], 'MoneyhubProfile': ['accounting'], } create_identity_data = { 'first_name': 'Foo', 'last_name': 'Bar', 'email': 'test_idt+' + str(random.randint(24, 4142355)) + '@sonymusic-pde.com', 'roles_to_attach': roles, 'tenant': {'tenant_type': 'account', 'tenant_uuid': 'dffedd4d-b88d-444d-a9eb-6ce89aa4d2f6'}, } response = requests.post( f'{config.QA_BASE_URL}/v2/identities', json=create_identity_data, headers={ 'Authorization': f'Bearer {bearer_token_user_with_vendor_star}', 'Orchard-Profile-Type': 'SettingsProfile', 'Orchard-Profile-Id': utils.OWS_ACCOUNT_TEST_USER_SETTINGS_PROFILE_ID, 'Orchard-Identity-Id': utils.OWS_ACCOUNT_TEST_USER_ID, }, ) assert response.status_code == 200 body = response.json() identity_id = body['id'] # GET tenants by identity_id to verify that new identity is associated with the tenant res = requests.get( f'{config.QA_BASE_URL}/v2/identity/{identity_id}/tenants', headers={ 'Authorization': f'Bearer {bearer_token_user_with_vendor_star}', 'Orchard-Profile-Type': 'SettingsProfile', 'Orchard-Profile-Id': utils.OWS_ACCOUNT_TEST_USER_SETTINGS_PROFILE_ID, 'Orchard-Identity-Id': utils.OWS_ACCOUNT_TEST_USER_ID, }, ) assert res.status_code == 200 tenants = res.json()['tenants'] assert tenants[0]['tenant_uuid'] == create_identity_data['tenant']['tenant_uuid'] assert tenants[0]['tenant_type'] == create_identity_data['tenant']['tenant_type'] assert set(tenants[0]['roles']) == set(account_base_roles + test_roles) # GET profiles from ows-users res = requests.get( f'{config.QA_OWS_USERS_URL}/users/identity/{identity_id}/profiles', headers={ 'Authorization': f'Bearer {bearer_token_user_with_vendor_star}', 'Orchard-Profile-Type': 'SettingsProfile', 'Orchard-Profile-Id': utils.OWS_ACCOUNT_TEST_USER_SETTINGS_PROFILE_ID, 'Orchard-Identity-Id': utils.OWS_ACCOUNT_TEST_USER_ID, }, ) profile_data = res.json() assert ( len(profile_data['items']) == profile_data['pagination']['total_records'] == len(expected_profiles) ) actual_profiles = { item['profile_type']: sorted(item['roles']) for item in profile_data['items'] } for profile_type, roles in expected_profiles.items(): assert profile_type in actual_profiles assert sorted(roles) == actual_profiles[profile_type] profile_ids = sorted(item['profile_id'] for item in profile_data['items']) assert len(profile_ids) == len(set(profile_ids)) # All IDs are unique for i in range(1, len(profile_ids)): assert profile_ids[i] > profile_ids[i - 1] # IDs are increasing def create_identity_v2_songwhip_access( bearer_token_user_with_vendor_star: str, ) -> None: """Test create_identity_v2 endpoint with songwhip access.""" create_identity_data = { 'first_name': 'Foo', 'last_name': 'Bar', 'email': 'test_idt+' + str(random.randint(24, 4142355)) + '@sonymusic-pde.com', 'roles_to_attach': ['SONGWHIP_READ_ROLE'], # Test Label (7123) 'tenant': {'tenant_type': 'account', 'tenant_uuid': '573d0372-7f2f-48a6-8deb-c9a6558f9549'}, } expected_profiles = { 'SettingsProfile': [], 'SongwhipProfile': ['songwhip_read'], } response = requests.post( f'{config.QA_BASE_URL}/v2/identities', json=create_identity_data, headers={ 'Authorization': f'Bearer {bearer_token_user_with_vendor_star}', 'Orchard-Profile-Type': 'SettingsProfile', 'Orchard-Profile-Id': utils.OWS_ACCOUNT_TEST_USER_SETTINGS_PROFILE_ID, 'Orchard-Identity-Id': utils.OWS_ACCOUNT_TEST_USER_ID, }, ) assert response.status_code == 200 body = response.json() identity_id = body['id'] # GET profiles from ows-users res = requests.get( f'{config.QA_OWS_USERS_URL}/users/identity/{identity_id}/profiles', headers={ 'Authorization': f'Bearer {bearer_token_user_with_vendor_star}', 'Orchard-Profile-Type': 'SettingsProfile', 'Orchard-Profile-Id': utils.OWS_ACCOUNT_TEST_USER_SETTINGS_PROFILE_ID, 'Orchard-Identity-Id': utils.OWS_ACCOUNT_TEST_USER_ID, }, ) profile_data = res.json() actual_profiles = { item['profile_type']: sorted(item['roles']) for item in profile_data['items'] } assert len(profile_data['items']) == 2 for profile_type, roles in expected_profiles.items(): assert profile_type in actual_profiles assert sorted(roles) == actual_profiles[profile_type] @pytest.mark.parametrize( ('test_roles', 'expected_label_profile_roles'), [ # Both SETTINGS_BASE_ROLE + WORKSTATION_ADMIN_ROLE pytest.param( ['SETTINGS_BASE_ROLE', 'WORKSTATION_ADMIN_ROLE'], ['administrator', 'catalog'], id='both_settings_and_workstation_admin', ), # Only SETTINGS_BASE_ROLE pytest.param(['SETTINGS_BASE_ROLE'], ['catalog'], id='only_settings'), # Only WORKSTATION_ADMIN_ROLE pytest.param( ['WORKSTATION_ADMIN_ROLE'], ['administrator', 'catalog'], id='only_workstation_admin' ), # Neither role pytest.param([], ['catalog'], id='neither_settings_nor_workstation_admin'), ], ) def test_create_identity_v2_subaccount( bearer_token_user_with_vendor_star: str, test_roles: list[str], expected_label_profile_roles: list[str], ) -> None: """Integration test for create_identity_v2 endpoint with subaccount tenant.""" roles = subaccount_base_roles + test_roles expected_profiles = { 'InsightsProfile': ['analytics'], 'LabelProfile': expected_label_profile_roles, 'SettingsProfile': [], 'MoneyhubProfile': ['accounting'], } create_identity_data = { 'first_name': 'Foo', 'last_name': 'Bar', 'email': 'test_idt+' + str(random.randint(24, 4142355)) + '@sonymusic-pde.com', 'roles_to_attach': roles, 'tenant': { 'tenant_type': 'subaccount', 'tenant_uuid': '1fab3fe8-c9a6-488a-9a96-959734a93997', }, } response = requests.post( f'{config.QA_BASE_URL}/v2/identities', json=create_identity_data, headers={ 'Authorization': f'Bearer {bearer_token_user_with_vendor_star}', 'Orchard-Profile-Type': 'SettingsProfile', 'Orchard-Profile-Id': utils.OWS_ACCOUNT_TEST_USER_SETTINGS_PROFILE_ID, 'Orchard-Identity-Id': utils.OWS_ACCOUNT_TEST_USER_ID, }, ) assert response.status_code == 200 body = response.json() identity_id = body['id'] # GET tenants by identity_id to verify that new identity is associated with the tenant res = requests.get( f'{config.QA_BASE_URL}/v2/identity/{identity_id}/tenants', headers={ 'Authorization': f'Bearer {bearer_token_user_with_vendor_star}', 'Orchard-Profile-Type': 'SettingsProfile', 'Orchard-Profile-Id': utils.OWS_ACCOUNT_TEST_USER_SETTINGS_PROFILE_ID, 'Orchard-Identity-Id': utils.OWS_ACCOUNT_TEST_USER_ID, }, ) assert res.status_code == 200 tenants = res.json()['tenants'] assert tenants[0]['tenant_uuid'] == create_identity_data['tenant']['tenant_uuid'] assert tenants[0]['tenant_type'] == create_identity_data['tenant']['tenant_type'] assert set(tenants[0]['roles']) == set(subaccount_base_roles + test_roles) # GET profiles from ows-users res = requests.get( f'{config.QA_OWS_USERS_URL}/users/identity/{identity_id}/profiles', headers={ 'Authorization': f'Bearer {bearer_token_user_with_vendor_star}', 'Orchard-Profile-Type': 'SettingsProfile', 'Orchard-Profile-Id': utils.OWS_ACCOUNT_TEST_USER_SETTINGS_PROFILE_ID, 'Orchard-Identity-Id': utils.OWS_ACCOUNT_TEST_USER_ID, }, ) profile_data = res.json() assert ( len(profile_data['items']) == profile_data['pagination']['total_records'] == len(expected_profiles) ) # noqa: E501 actual_profiles = { item['profile_type']: sorted(item['roles']) for item in profile_data['items'] } for profile_type, roles in expected_profiles.items(): assert profile_type in actual_profiles assert sorted(roles) == actual_profiles[profile_type] profile_ids = sorted(item['profile_id'] for item in profile_data['items']) assert len(profile_ids) == len(set(profile_ids)) # All IDs are unique for i in range(1, len(profile_ids)): assert profile_ids[i] > profile_ids[i - 1] # IDs are increasing @pytest.mark.parametrize( 'roles', [ pytest.param(['fansifter_can_view_fan_data'], id='Fansifter'), pytest.param(['songwhip_read'], id='Songwhip'), ], ) def test_create_identity_with_unsupported_roles_for_subaccount( bearer_token_user_with_vendor_star: str, create_identity_with_subaccount: dict, roles: list[str] ) -> None: """Test fails to give access to fansifter for non-Account tenant type.""" create_identity_with_subaccount['roles_to_attach'] = roles response = requests.post( f'{config.QA_BASE_URL}/v2/identities', json=create_identity_with_subaccount, headers={ 'Authorization': f'Bearer {bearer_token_user_with_vendor_star}', 'Orchard-Profile-Type': 'SettingsProfile', 'Orchard-Profile-Id': utils.OWS_ACCOUNT_TEST_USER_SETTINGS_PROFILE_ID, 'Orchard-Identity-Id': utils.OWS_ACCOUNT_TEST_USER_ID, }, ) assert response.status_code == 400 assert response.json()['message']['_schema'] == [ f'Role {roles[0]} is not supported by the subaccount tenant type.' ] def test_create_identity_fails_vendor_star_bad_email_domain( bearer_token_user_with_vendor_star, create_identity_json ): """Test fails to create identity with vendor star for bad email domain.""" create_identity_json['tenant']['tenant_uuid'] = '053a1a75-acc5-4cd8-9206-a194335d2afa' create_identity_json['email'] = create_identity_json['email'].replace( '@sonymusic-pde.com', '@bar.com', ) response = requests.post( f'{config.QA_BASE_URL}/v2/identities', json=create_identity_json, headers={ 'Authorization': f'Bearer {bearer_token_user_with_vendor_star}', 'Orchard-Profile-Type': 'SettingsProfile', 'Orchard-Profile-Id': utils.OWS_ACCOUNT_TEST_USER_SETTINGS_PROFILE_ID, 'Orchard-Identity-Id': utils.OWS_ACCOUNT_TEST_USER_ID, }, ) assert response.status_code == 403, response.text assert response.json() == { 'code': 'bad_request', 'message': 'Cannot assign resource to this identity', }, response.text def test_create_identity_fails_invalid_roles( bearer_token_user_with_vendor_star, create_identity_json ): """Test fails to create identity when an unknown role is passed (schema validation).""" create_identity_json['roles_to_attach'].append('WORKSTATION_ACCOUNTING_BASE_ROLE') response = requests.post( f'{config.QA_BASE_URL}/v2/identities', json=create_identity_json, headers={ 'Authorization': f'Bearer {bearer_token_user_with_vendor_star}', 'Orchard-Profile-Type': 'SettingsProfile', 'Orchard-Profile-Id': utils.OWS_ACCOUNT_TEST_USER_SETTINGS_PROFILE_ID, 'Orchard-Identity-Id': utils.OWS_ACCOUNT_TEST_USER_ID, }, ) assert response.status_code == 400, response.text assert response.json()['code'] == 'validation_error' @pytest.mark.parametrize( 'email, expected_message', [ pytest.param( 'a' * 65 + '@sonymusic-pde.com', 'Email local part exceeds maximum length of 64 characters.', id='Local part exceeds 64 characters', ), pytest.param( 'a' * 243 + '@sonymusic-pde.com', 'Email exceeds maximum length of 254 characters.', id='Total length exceeds 254 characters', ), ], ) def test_create_identity_fails_email_too_long( bearer_token_user_with_vendor_star, create_identity_json, email: str, expected_message: str, ): """Test that emails violating Auth0 length rules are rejected before downstream calls.""" create_identity_json['email'] = email response = requests.post( f'{config.QA_BASE_URL}/v2/identities', json=create_identity_json, headers={ 'Authorization': f'Bearer {bearer_token_user_with_vendor_star}', 'Orchard-Profile-Type': 'SettingsProfile', 'Orchard-Profile-Id': utils.OWS_ACCOUNT_TEST_USER_SETTINGS_PROFILE_ID, 'Orchard-Identity-Id': utils.OWS_ACCOUNT_TEST_USER_ID, }, ) assert response.status_code == 400, response.text assert response.json()['code'] == 'validation_error' assert expected_message in response.json()['message']['email'] def test_create_identity_fails_role_not_enabled_for_tenant( bearer_token_user_with_vendor_star, create_identity_json ): """Test fails with 422 when a valid role is not enabled for the tenant's configuration.""" create_identity_json['roles_to_attach'].append('fansifter_can_view_fan_data') response = requests.post( f'{config.QA_BASE_URL}/v2/identities', json=create_identity_json, headers={ 'Authorization': f'Bearer {bearer_token_user_with_vendor_star}', 'Orchard-Profile-Type': 'SettingsProfile', 'Orchard-Profile-Id': utils.OWS_ACCOUNT_TEST_USER_SETTINGS_PROFILE_ID, 'Orchard-Identity-Id': utils.OWS_ACCOUNT_TEST_USER_ID, }, ) assert response.status_code == 422, response.text assert response.json() == { 'code': 'Unable to process the request because it contains invalid data.', 'message': 'Error creating identity.', } def test_create_identity_collaborator( bearer_token_user_vendor_star_not_superadmin: str, create_identity_json: dict[str, str | list[str] | dict[str, str]], ) -> None: """Test create identity with access to collaborator tenant.""" create_identity_json['tenant'] = { 'tenant_type': 'collaborator', # If this collaborator stops working, replace with another with a dpEnabledDate 'tenant_uuid': 'c4e63299-1383-4bde-8d6a-9fbdfec34e87', } create_identity_json['roles_to_attach'] = [ 'CUSTOMER_ACCOUNTING_BASE_ROLE', 'BANKING_TAX_BASE_ROLE', ] response = requests.post( f'{config.QA_BASE_URL}/v2/identities', json=create_identity_json, headers={ 'Authorization': f'Bearer {bearer_token_user_vendor_star_not_superadmin}', 'Orchard-Profile-Type': 'SettingsProfile', 'Orchard-Profile-Id': utils.OWS_PERMISSIONS_VENDOR_STAR_USER_SETTINGS_PROFILE_ID, 'Orchard-Identity-Id': utils.OWS_PERMISSIONS_VENDOR_STAR_USER_ID, }, ) assert response.status_code == 200 body = response.json() identity_id = body['id'] # GET tenants by identity_id to verify that new identity is associated with the tenant res = requests.get( f'{config.QA_BASE_URL}/v2/identity/{identity_id}/tenants', headers={ 'Authorization': f'Bearer {bearer_token_user_vendor_star_not_superadmin}', 'Orchard-Profile-Type': 'SettingsProfile', 'Orchard-Profile-Id': utils.OWS_PERMISSIONS_VENDOR_STAR_USER_SETTINGS_PROFILE_ID, 'Orchard-Identity-Id': utils.OWS_PERMISSIONS_VENDOR_STAR_USER_ID, }, ) assert res.status_code == 200 tenants = res.json()['tenants'] assert tenants[0]['tenant_uuid'] == create_identity_json['tenant']['tenant_uuid'] assert tenants[0]['tenant_type'] == create_identity_json['tenant']['tenant_type'] assert set(tenants[0]['roles']) == set(create_identity_json['roles_to_attach']) def test_create_identity_tenant_not_found( bearer_token_user_vendor_star_not_superadmin: str, create_identity_json: dict ) -> None: """Test create identity returns 404 when tenant doesn't exist.""" create_identity_data = { **create_identity_json, 'tenant': {'tenant_type': 'account', 'tenant_uuid': '00000000-0000-0000-0000-000000000000'}, } response = requests.post( f'{config.QA_BASE_URL}/v2/identities', json=create_identity_data, headers={ 'Authorization': f'Bearer {bearer_token_user_vendor_star_not_superadmin}', 'Orchard-Profile-Type': 'SettingsProfile', 'Orchard-Profile-Id': utils.OWS_PERMISSIONS_VENDOR_STAR_USER_SETTINGS_PROFILE_ID, 'Orchard-Identity-Id': utils.OWS_PERMISSIONS_VENDOR_STAR_USER_ID, }, ) assert response.status_code == 404 assert response.json()['message'] == 'Tenant not found.'