"""Unit tests for tenant model.""" import textwrap import uuid from unittest import mock from unittest.mock import MagicMock, Mock, patch import pytest from neo4j.graph import Graph, Node from pytest_mock import MockerFixture from permissions.connectors import neo4j from permissions.models import tenant from permissions.models.tenant import AdminableTenant from permissions.types import AccessibleTenant, ProfileInfo, Tenant, TenantType from tests.unit.conftest import get_session_mock, get_transactional_session_mock EXPECTED_QUERY_ADMIN_EMPLOYEE = textwrap.dedent(""" MATCH (tenant)<-[ :HAS_ACCESS_TO|HAS_ADMIN_ACCESS_TO ]-(up:Profile) <-[:HAS_PROFILE]-(user:Identity) WHERE ( tenant:Vendor OR tenant:SubAccount OR tenant:Collaborator OR tenant:LabelParticipant ) AND user.id = $userIdentityId WITH tenant, COLLECT( { profileType: up.profileType, profileId: up.profileId, roles: up.roles, uuid:up.uuid } ) as profiles RETURN DISTINCT tenant, profiles SKIP $offset LIMIT $limit """) EXPECTED_QUERY_ADMIN_EMPLOYEE_WITH_DELETED_ACCESS = textwrap.dedent(""" MATCH (tenant)<-[ :HAS_ACCESS_TO|HAS_ADMIN_ACCESS_TO|DELETED_HAS_ACCESS_TO|DELETED_HAS_ADMIN_ACCESS_TO ]-(up:Profile) <-[:HAS_PROFILE]-(user:Identity) WHERE ( tenant:Vendor OR tenant:SubAccount OR tenant:Collaborator OR tenant:LabelParticipant ) AND user.id = $userIdentityId WITH tenant, COLLECT( { profileType: up.profileType, profileId: up.profileId, roles: up.roles, uuid:up.uuid } ) as profiles RETURN DISTINCT tenant, profiles SKIP $offset LIMIT $limit """) EXPECTED_QUERY_ADMIN_NON_EMPLOYEE = textwrap.dedent(""" MATCH (admin:Identity)-[:HAS_PROFILE]->(sp:Profile {profileType: $profileType}) -[:HAS_ADMIN_ACCESS_TO]->(t)-[*0..1]->(tenant) <-[ :HAS_ACCESS_TO|HAS_ADMIN_ACCESS_TO ]-(up:Profile)<-[:HAS_PROFILE]-(user:Identity) USING JOIN ON t WHERE ( tenant:Vendor OR tenant:SubAccount OR tenant:Collaborator OR tenant:LabelParticipant ) AND admin.id = $adminIdentityId AND sp.profileId = $adminProfileId AND user.id = $userIdentityId WITH tenant, COLLECT( { profileType: up.profileType, profileId: up.profileId, roles: up.roles, uuid:up.uuid } ) as profiles RETURN DISTINCT tenant, profiles SKIP $offset LIMIT $limit """) EXPECTED_QUERY_ADMIN_NON_EMPLOYEE_WITH_DELETED_ACCESS = textwrap.dedent(""" MATCH (admin:Identity)-[:HAS_PROFILE]->(sp:Profile {profileType: $profileType}) -[:HAS_ADMIN_ACCESS_TO]->(t)-[*0..1]->(tenant) <-[ :HAS_ACCESS_TO|HAS_ADMIN_ACCESS_TO|DELETED_HAS_ACCESS_TO|DELETED_HAS_ADMIN_ACCESS_TO ]-(up:Profile)<-[:HAS_PROFILE]-(user:Identity) USING JOIN ON t WHERE ( tenant:Vendor OR tenant:SubAccount OR tenant:Collaborator OR tenant:LabelParticipant ) AND admin.id = $adminIdentityId AND sp.profileId = $adminProfileId AND user.id = $userIdentityId WITH tenant, COLLECT( { profileType: up.profileType, profileId: up.profileId, roles: up.roles, uuid:up.uuid } ) as profiles RETURN DISTINCT tenant, profiles SKIP $offset LIMIT $limit """) @pytest.mark.parametrize('include_deleted_tenants', [False, True]) @mock.patch('permissions.models.tenant.format_adminable_tenants_result') @mock.patch('permissions.models.profile.check_vendor_star_access') @mock.patch('permissions.models.tenant.get_adminable_tenants_query') @mock.patch('permissions.connectors.neo4j.db_session') def test_get_adminable_tenants_for_identity_employee_admin( db_session_mock, get_query_mock, access_check_mock, format_mock, include_deleted_tenants ): """Test the employee admin case of the get_adminable_tenants_for_identity method.""" admin_context = {'identity_id': 'aaa', 'profile_id': 111} session_mock = get_session_mock() db_session_mock.return_value = session_mock query_result = [mock.Mock(), mock.Mock()] session_mock.__enter__.return_value.run.return_value = query_result access_check_mock.return_value = True format_result = [ {'tenant': {'key': 'val'}, 'profiles': [{'id': 8}]}, {'tenant': {'key': 'va1'}, 'profiles': [{'id': 9}, {'id': 10}]}, ] format_mock.return_value = format_result result = tenant.get_adminable_tenants_for_identity( admin_context=admin_context, identity_id='bbb', limit=200, offset=0, include_deleted_tenants=include_deleted_tenants, ) assert result == format_result db_session_mock.assert_called_with(access_mode='READ') access_check_mock.assert_called_with( identity_id='aaa', profile_type='SettingsProfile', profile_id=111, session=session_mock.__enter__.return_value, ) get_query_mock.assert_called_with(include_deleted_tenants, True) session_mock.__enter__.return_value.run.assert_called_with( get_query_mock.return_value, profileType='SettingsProfile', adminIdentityId='aaa', adminProfileId=111, userIdentityId='bbb', offset=0, limit=200, ) format_mock.assert_called_with(query_result) @pytest.mark.parametrize('include_deleted_tenants', [False, True]) @mock.patch('permissions.models.tenant.format_adminable_tenants_result') @mock.patch('permissions.models.profile.check_vendor_star_access') @mock.patch('permissions.models.tenant.get_adminable_tenants_query') @mock.patch('permissions.connectors.neo4j.db_session') def test_get_adminable_tenants_for_identity_nonemployee_admin( db_session_mock, get_query_mock, access_check_mock, _, include_deleted_tenants ): """Test the nonemployee admin case of the get_adminable_tenants_for_identity method.""" admin_context = {'identity_id': 'aaa', 'profile_id': 111} session_mock = get_session_mock() db_session_mock.return_value = session_mock query_result = [mock.Mock(), mock.Mock()] session_mock.__enter__.return_value.run.return_value = query_result access_check_mock.return_value = False tenant.get_adminable_tenants_for_identity( admin_context=admin_context, identity_id='bbb', limit=200, offset=0, include_deleted_tenants=include_deleted_tenants, ) get_query_mock.assert_called_with(include_deleted_tenants, False) session_mock.__enter__.return_value.run.assert_called_with( get_query_mock.return_value, profileType='SettingsProfile', adminIdentityId='aaa', adminProfileId=111, userIdentityId='bbb', offset=0, limit=200, ) def test_format_adminable_tenants_result(make_graph_node): """Test method takes neo4j records and returns a list of AdminableTenant.""" tenant_node = make_graph_node( node_id=1, labels=frozenset(['Vendor', 'Label']), data={'vendorId': 30, 'uuid': 'abc-123'} ) profile_node1 = {'profileId': 2, 'profileType': 'SettingsProfile', 'roles': [], 'uuid': 'abc'} profile_node2 = { 'profileId': 3, 'profileType': 'InsightsProfile', 'roles': ['analytics'], 'uuid': 'def', } query_result = [{'tenant': tenant_node, 'profiles': [profile_node1, profile_node2]}] result = tenant.format_adminable_tenants_result(query_result) assert result == [ AdminableTenant( tenant=Tenant(tenant_type=TenantType.ACCOUNT, tenant_uuid='abc-123'), profiles=[ ProfileInfo(profile_id=2, profile_type='SettingsProfile', roles=[], uuid='abc'), ProfileInfo( profile_id=3, profile_type='InsightsProfile', roles=['analytics'], uuid='def' ), ], ), ] @pytest.mark.parametrize('include_deleted_tenants', [False, True]) def test_get_adminable_tenants_query_employee_admin(include_deleted_tenants): """Test get_adminable_tenants_query behavior when admin is an employee.""" result = tenant.get_adminable_tenants_query( include_deleted_tenants=include_deleted_tenants, has_full_catalog_access=True ) if include_deleted_tenants: assert result == EXPECTED_QUERY_ADMIN_EMPLOYEE_WITH_DELETED_ACCESS else: assert result == EXPECTED_QUERY_ADMIN_EMPLOYEE @pytest.mark.parametrize('include_deleted_tenants', [False, True]) def test_get_adminable_tenants_query_nonemployee_admin(include_deleted_tenants): """Test get_adminable_tenants_query behavior when admin is not an employee.""" result = tenant.get_adminable_tenants_query( include_deleted_tenants=include_deleted_tenants, has_full_catalog_access=False ) if include_deleted_tenants: assert result == EXPECTED_QUERY_ADMIN_NON_EMPLOYEE_WITH_DELETED_ACCESS else: assert result == EXPECTED_QUERY_ADMIN_NON_EMPLOYEE def test_get_admin_tenant_type_count(app_context): """Test get_admin_tenant_type_count.""" session_mock = get_transactional_session_mock() identity_id = 'test-admin-uuid-123' with patch.object(neo4j, '_get_neo4j_session', return_value=session_mock), patch.object( tenant, '_get_tenant_type_label_count', return_value=3 ), patch.object( tenant, '_get_tenant_type_label_participant_count', return_value=2 ), patch.object(tenant, '_get_tenant_type_collaborator_count', return_value=0): actual = tenant.get_admin_tenant_type_count(identity_id) assert actual assert actual.message == [ {'tenant_type': 'account', 'tenant_count': 3}, {'tenant_type': 'label_participant', 'tenant_count': 2}, ] assert tenant._get_tenant_type_label_count.call_count == 1 assert tenant._get_tenant_type_label_participant_count.call_count == 1 assert tenant._get_tenant_type_collaborator_count.call_count == 1 def test_check_admin_access_to_tenants(test_logic_tenants, test_settings_profile): """Test check_admin_access_to_tenants.""" with patch('neo4j.Session') as MockSession: mock_session_instance = MockSession.return_value mock_session_instance.run.side_effect = [ [ {'type': 'Vendor', 'uuid': 'abc-123', 'tenant': dict({'properties': dict()})}, {'type': 'Vendor', 'uuid': 'mno-345', 'tenant': None}, ], [ {'type': 'Subaccount', 'uuid': 'def-456', 'tenant': None}, {'type': 'Subaccount', 'uuid': 'pqr-678', 'tenant': dict({'properties': dict()})}, ], [ {'type': 'Collaborator', 'uuid': 'ghi-789', 'tenant': dict({'properties': dict()})}, {'type': 'Collaborator', 'uuid': 'stu-901', 'tenant': None}, ], [ { 'type': 'LabelParticipant', 'uuid': 'jkl-012', 'tenant': dict({'properties': dict()}), }, {'type': 'LabelParticipant', 'uuid': 'vwx-234', 'tenant': None}, ], ] actual = tenant.check_admin_access_to_tenants( session=mock_session_instance, tenants=test_logic_tenants, settings_profile=test_settings_profile, ) assert actual == [ AccessibleTenant(tenant_uuid='abc-123', tenant_type=TenantType.ACCOUNT, access=True), AccessibleTenant(tenant_uuid='mno-345', tenant_type=TenantType.ACCOUNT, access=False), AccessibleTenant( tenant_uuid='def-456', tenant_type=TenantType.SUBACCOUNT, access=False ), AccessibleTenant(tenant_uuid='pqr-678', tenant_type=TenantType.SUBACCOUNT, access=True), AccessibleTenant( tenant_uuid='ghi-789', tenant_type=TenantType.COLLABORATOR, access=True ), AccessibleTenant( tenant_uuid='stu-901', tenant_type=TenantType.COLLABORATOR, access=False ), AccessibleTenant( tenant_uuid='jkl-012', tenant_type=TenantType.LABEL_PARTICIPANT, access=True ), AccessibleTenant( tenant_uuid='vwx-234', tenant_type=TenantType.LABEL_PARTICIPANT, access=False ), ] vendor_call = mock_session_instance.run.call_args_list[0] expected_query = textwrap.dedent(""" MATCH (p:Profile { profileType: 'SettingsProfile', profileId: $profileId }) UNWIND $tenant_uuids as tenant_uuid OPTIONAL MATCH (p)-[:HAS_ADMIN_ACCESS_TO]->(tenant:Vendor { uuid: tenant_uuid }) RETURN tenant_uuid as uuid, 'Vendor' as type, tenant """) assert vendor_call[1]['query'] == expected_query assert vendor_call[1]['profileId'] == 123 assert vendor_call[1]['tenant_uuids'] == ['abc-123', 'mno-345'] subaccount_call = mock_session_instance.run.call_args_list[1] expected_query = textwrap.dedent(""" MATCH (p:Profile { profileType: 'SettingsProfile', profileId: $profileId }) UNWIND $tenant_uuids as tenant_uuid OPTIONAL MATCH (tenant:Subaccount { uuid: tenant_uuid }) WHERE (p)-[:HAS_ADMIN_ACCESS_TO]->(tenant) OR (p)-[:HAS_ADMIN_ACCESS_TO]->(:Vendor)-[:OWNS]->(tenant) RETURN tenant_uuid as uuid, 'Subaccount' as type, tenant """) assert subaccount_call[1]['query'] == expected_query assert subaccount_call[1]['profileId'] == 123 assert subaccount_call[1]['tenant_uuids'] == ['def-456', 'pqr-678'] collaborator_call = mock_session_instance.run.call_args_list[2] expected_query = textwrap.dedent(""" MATCH (p:Profile { profileType: 'SettingsProfile', profileId: $profileId }) UNWIND $tenant_uuids as tenant_uuid OPTIONAL MATCH (p)-[:HAS_ADMIN_ACCESS_TO]->(:Vendor) -[:OWNS]->(tenant:Collaborator { uuid: tenant_uuid }) RETURN tenant_uuid as uuid, 'Collaborator' as type, tenant """) assert collaborator_call[1]['query'] == expected_query assert collaborator_call[1]['profileId'] == 123 assert collaborator_call[1]['tenant_uuids'] == ['ghi-789', 'stu-901'] label_participant_call = mock_session_instance.run.call_args_list[3] expected_query = textwrap.dedent(""" MATCH (p:Profile {profileType: 'SettingsProfile', profileId: $profileId}) UNWIND $tenant_uuids as tenant_uuid OPTIONAL MATCH (tenant:LabelParticipant { uuid: tenant_uuid }) WHERE (p)-[:HAS_ADMIN_ACCESS_TO]->(:Label)-[:HAS_LABEL_PARTICIPANT]->(tenant) OR (p)-[:HAS_ADMIN_ACCESS_TO]->(tenant) RETURN tenant_uuid as uuid, 'LabelParticipant' as type, tenant """) assert label_participant_call[1]['query'] == expected_query assert label_participant_call[1]['profileId'] == 123 assert label_participant_call[1]['tenant_uuids'] == ['jkl-012', 'vwx-234'] def test_get_parent_company_brand_for_tenant(make_graph_node: str, mocker: MockerFixture) -> None: """Test get_parent_company_brand_for_tenant.""" expected_query = textwrap.dedent(""" RETURN CASE $tenant_type WHEN 'label_participant' THEN [(l:LabelParticipant {uuid:$tenant_uuid})<-[:HAS_LABEL_PARTICIPANT]-(v:Vendor) <-[:HAS_LABEL]-(cb:CompanyBrand)| cb.name ] WHEN 'account' THEN [(v:Vendor {uuid:$tenant_uuid})<-[:HAS_LABEL]-(cb:CompanyBrand) | cb.name ] WHEN 'subaccount' THEN [(s:Subaccount {uuid:$tenant_uuid})<-[:OWNS]-(v:Vendor) <-[:HAS_LABEL]-(cb:CompanyBrand) | cb.name ] WHEN 'collaborator' THEN [(c:Collaborator {uuid: $tenant_uuid}) <-[:OWNS]-(v:Vendor)<-[:HAS_LABEL]-(cb:CompanyBrand) | cb.name ] END AS brand""") with patch('neo4j.Session') as MockSession: mock_session_instance = MockSession.return_value mock_node = Node(Graph(), '1', 1, n_labels=['CompanyBrand'], properties={'brand': 'awal'}) mock_session_instance.run.return_value.single.return_value = mock_node tenant.get_parent_company_brand_for_tenant( tx=mock_session_instance, tenant_type=TenantType.ACCOUNT, tenant_uuid='user123' ) account_call = mock_session_instance.run.call_args_list[0] assert account_call[0][0] == expected_query def expected_soft_delete_access_to_tenant_for_identity_query() -> str: """Get expected query for soft delete all access to tenant.""" return textwrap.dedent(""" MATCH (tenant)<-[rel:HAS_ACCESS_TO|HAS_ADMIN_ACCESS_TO]-(p:Profile) <-[:HAS_PROFILE]-(i:Identity) WHERE $tenantType in LABELS(tenant) AND i.id = $identityId AND tenant.uuid = $tenantUUID SET rel.deletedAt = datetime(), rel.deletedBy = $deletedBy WITH tenant, p, rel CALL apoc.refactor.setType(rel, apoc.text.join(['DELETED', apoc.rel.type(rel)], '_')) YIELD input, output RETURN true""") @mock.patch('permissions.connectors.neo4j.db_session') def test_soft_delete_access_to_tenant_for_identity(db_session_mock): """Test soft delete all access to a tenant for identity.""" identity_id = 'test-user-123' admin_identity_id = 'admin-user-123' tenant_mock = Tenant(tenant_uuid='abc-123', tenant_type=TenantType.ACCOUNT) session_mock = get_session_mock() db_session_mock.return_value = session_mock query_result = [mock.Mock(), mock.Mock()] session_mock.__enter__.return_value.run.return_value = query_result tenant.soft_delete_access_to_tenant_for_identity( session_mock.__enter__.return_value, admin_identity_id=admin_identity_id, identity_id=identity_id, tenant=tenant_mock, ) deleted_by = f'ows-permissions/revoke-all-access-to-single-tenant/{admin_identity_id}' session_mock.__enter__.return_value.run.assert_called_with( expected_soft_delete_access_to_tenant_for_identity_query(), identityId=identity_id, deletedBy=deleted_by, tenantUUID=tenant_mock.tenant_uuid, tenantType='Vendor', ) @mock.patch('permissions.connectors.neo4j.db_session') def test_get_identity_tenant_count(db_session_mock): """Test get_identity_tenant_count.""" expected_query = textwrap.dedent(""" MATCH (user:Identity)-[:HAS_PROFILE]->(p:Profile)-[:HAS_ACCESS_TO|HAS_ADMIN_ACCESS_TO]->(x) WHERE user.id = $identityId RETURN COUNT(DISTINCT x) as tenant_count""") mock_session = Mock() mock_result = Mock() mock_result.get.return_value = 3 mock_session.run.return_value.single.return_value = mock_result db_session_mock.return_value.__enter__.return_value = mock_session db_session_mock.return_value.__exit__.return_value = None actual = tenant.get_identity_tenant_count(identity_id='test-user-123') assert actual == 3 mock_session.run.assert_called_once() call_args = mock_session.run.call_args assert call_args[0][0] == expected_query assert call_args[1] == {'identityId': 'test-user-123'} @mock.patch('permissions.connectors.neo4j.db_session') def test_get_brands_for_identity(db_session_mock): """Test get_brands_for_identity returns unique brands from Vendor and Subaccount tenants.""" expected_query = textwrap.dedent(""" MATCH (:Identity {id: $identityId})-[:HAS_PROFILE]->(:Profile) -[:HAS_ACCESS_TO|HAS_ADMIN_ACCESS_TO]->(tenant) WHERE tenant:Vendor OR tenant:Subaccount OR tenant:LabelParticipant OR tenant:Collaborator WITH DISTINCT tenant, // brands if tenant is a Vendor [(tenant)<-[:HAS_LABEL]-(cb:CompanyBrand) WHERE tenant:Vendor | cb.name] AS vendorBrands, // brands if tenant is a Subaccount or Collaborator (both owned by a Vendor) [(tenant)<-[:OWNS]-(:Vendor)<-[:HAS_LABEL]-(cb2:CompanyBrand) WHERE tenant:Subaccount OR tenant:Collaborator | cb2.name] AS ownedBrands, // brands if tenant is a LabelParticipant (linked via HAS_LABEL_PARTICIPANT to a Vendor) [(tenant)<-[:HAS_LABEL_PARTICIPANT]-(:Vendor)<-[:HAS_LABEL]-(cb3:CompanyBrand) WHERE tenant:LabelParticipant | cb3.name] AS lpBrands WITH vendorBrands + ownedBrands + lpBrands AS allBrands UNWIND allBrands AS brand RETURN DISTINCT brand ORDER BY brand; """) mock_session = Mock() mock_records = [ Mock(get=Mock(return_value='sme')), Mock(get=Mock(return_value='theorchard')), Mock(get=Mock(return_value='awal')), ] mock_session.run.return_value = mock_records db_session_mock.return_value.__enter__.return_value = mock_session db_session_mock.return_value.__exit__.return_value = None actual = tenant.get_brands_for_identity(identity_id='test-user-123') assert actual == ['sme', 'theorchard', 'awal'] mock_session.run.assert_called_once() call_args = mock_session.run.call_args assert call_args[0][0] == expected_query assert call_args[1] == {'identityId': 'test-user-123'} @mock.patch('permissions.connectors.neo4j.db_session') def test_get_brands_for_identity_empty_result(db_session_mock): """Test get_brands_for_identity returns empty list when user has no brand access.""" mock_session = Mock() mock_session.run.return_value = [] db_session_mock.return_value.__enter__.return_value = mock_session db_session_mock.return_value.__exit__.return_value = None actual = tenant.get_brands_for_identity(identity_id='test-user-456') assert actual == [] mock_session.run.assert_called_once() @mock.patch('permissions.connectors.neo4j.db_session') def test_get_brands_for_identity_filters_none_values(db_session_mock): """Test get_brands_for_identity filters out None values from results.""" mock_session = Mock() mock_records = [ Mock(get=Mock(return_value='Sony Music')), Mock(get=Mock(return_value=None)), Mock(get=Mock(return_value='RCA Records')), ] mock_session.run.return_value = mock_records db_session_mock.return_value.__enter__.return_value = mock_session db_session_mock.return_value.__exit__.return_value = None actual = tenant.get_brands_for_identity(identity_id='test-user-789') assert actual == ['Sony Music', 'RCA Records'] assert None not in actual def test_soft_delete_access_to_multiple_tenants_for_identity(): """Test soft_delete_access_to_multiple_tenants_for_identity...""" expected_query = textwrap.dedent(""" MATCH (i:Identity {id: $identityId})-[:HAS_PROFILE]->(p:Profile) -[rel:HAS_ACCESS_TO|HAS_ADMIN_ACCESS_TO]->(tenant) WHERE tenant.uuid IN [t IN $tenants | t.uuid] AND ANY(t IN $tenants WHERE t.uuid = tenant.uuid AND t.type IN LABELS(tenant)) SET rel.deletedAt = datetime(), rel.deletedBy = $deletedBy WITH tenant, p, rel CALL apoc.refactor.setType( rel, apoc.text.join(['DELETED', apoc.rel.type(rel)], '_') ) YIELD input, output RETURN true """) mock_tx = mock.Mock() tenants = [ Tenant(tenant_type=TenantType.SUBACCOUNT, tenant_uuid='a-uuid'), Tenant(tenant_type=TenantType.ACCOUNT, tenant_uuid='b-uuid'), ] tenant.soft_delete_access_to_multiple_tenants_for_identity( tx=mock_tx, tenants=tenants, admin_id='uuid-of-admin', identity_id='uuid-of-identity', ) mock_tx.run.assert_called_with( expected_query, { 'identityId': 'uuid-of-identity', 'tenants': [ {'type': 'Subaccount', 'uuid': 'a-uuid'}, {'type': 'Vendor', 'uuid': 'b-uuid'}, ], 'deletedBy': 'ows-permissions/revoke-all-access-to-tenants/uuid-of-admin', }, ) def test_soft_delete_all_access_to_tenants_for_identity() -> None: """Test soft_delete_all_access_to_tenants_for_identity.""" identity_id = 'test-user-123' admin_id = 'admin-user-123' session_mock = mock.Mock() expected_query = textwrap.dedent(""" MATCH (i:Identity {id: $identityId})-[:HAS_PROFILE]->(p:Profile) -[rel:HAS_ACCESS_TO|HAS_ADMIN_ACCESS_TO]->(tenant) WHERE tenant:Vendor OR tenant:Subaccount OR tenant:Collaborator OR tenant:LabelParticipant OR tenant:ParentCompany SET rel.deletedAt = datetime(), rel.deletedBy = $deletedBy WITH tenant, p, rel CALL apoc.refactor.setType( rel, apoc.text.join(['DELETED', apoc.rel.type(rel)], '_') ) YIELD input, output RETURN true """) tenant.soft_delete_all_access_to_tenants_for_identity( session=session_mock, admin_id=admin_id, identity_id=identity_id ) session_mock.run.assert_called_with( expected_query, { 'identityId': identity_id, 'deletedBy': f'ows-permissions/revoke-all-access/{admin_id}', }, ) @pytest.mark.parametrize( ('tenant_type', 'expected_label'), [ (TenantType.ACCOUNT, 'Vendor'), (TenantType.SUBACCOUNT, 'Subaccount'), (TenantType.COLLABORATOR, 'Collaborator'), (TenantType.LABEL_PARTICIPANT, 'LabelParticipant'), ], ) @mock.patch('permissions.connectors.neo4j.db_session') def test_get_tenant_by_uuid_and_type( db_session_mock: MagicMock, make_graph_node: MagicMock, tenant_type: TenantType, expected_label: str, ) -> None: """Test get_tenant_by_uuid_and_type.""" tenant_uuid = 'test-tenant-uuid' expected_query = textwrap.dedent(f""" MATCH (t:{expected_label}) WHERE t.uuid = $tenantUuid RETURN DISTINCT t LIMIT 1 """) session_mock = get_session_mock() db_session_mock.return_value = session_mock session_mock.__enter__.return_value.run.return_value.single.return_value = make_graph_node( node_id=1, labels=frozenset([expected_label]), data={'t': {'uuid': tenant_uuid, 'name': 'Test Tenant'}}, ) result = tenant.get_tenant_by_uuid_and_type(tenant_uuid=tenant_uuid, tenant_type=tenant_type) assert result.tenant_uuid == tenant_uuid assert result.tenant_type == tenant_type assert result.tenant_name == 'Test Tenant' args = session_mock.__enter__().run.call_args assert str(args[0][0]) == expected_query @pytest.mark.parametrize( ( 'test_scenario', 'tenant_type', 'tenant_name', 'parent_labels', 'expected_parent_type', 'expected_uuid', 'expected_name', ), [ pytest.param( 'labelparticipant_with_subaccount_parent', TenantType.LABEL_PARTICIPANT, 'Test Label Participant', ['Subaccount'], TenantType.SUBACCOUNT, 'subaccount-parent-uuid', 'Test Subaccount Parent', id='labelparticipant_with_subaccount_parent', ), pytest.param( 'labelparticipant_with_vendor_parent_only', TenantType.LABEL_PARTICIPANT, 'Test Label Participant', ['Vendor'], TenantType.ACCOUNT, 'vendor-parent-uuid', 'Test Vendor Parent', id='labelparticipant_with_vendor_parent_only', ), pytest.param( 'subaccount_with_vendor_owner', TenantType.SUBACCOUNT, 'Test Subaccount', ['Vendor'], TenantType.ACCOUNT, 'vendor-owner-uuid', 'Test Vendor Owner', id='subaccount_with_vendor_owner', ), pytest.param( 'collaborator_with_vendor_owner', TenantType.COLLABORATOR, 'Test Collaborator', ['Vendor'], TenantType.ACCOUNT, 'vendor-owner-uuid', 'Test Vendor Owner', id='collaborator_with_vendor_owner', ), ], ) @mock.patch('permissions.connectors.neo4j.db_session') def test_get_tenant_parent_of_tenant_success_cases( db_session_mock: MagicMock, make_graph_node: MagicMock, test_scenario: str, tenant_type: TenantType, tenant_name: str, parent_labels: list[str], expected_parent_type: TenantType, expected_uuid: str, expected_name: str, ) -> None: """Test get_tenant_parent_of_tenant for various successful parent-child relationships.""" from permissions.types import TenantWithName tenant_input = TenantWithName( tenant_uuid=f'test-{tenant_type.value}-uuid', tenant_type=tenant_type, tenant_name=tenant_name, ) # Mock parent node parent_node = make_graph_node( node_id=100 + hash(test_scenario) % 1000, # Generate unique node id labels=frozenset(parent_labels), data={'uuid': expected_uuid, 'name': expected_name}, ) session_mock = get_session_mock() db_session_mock.return_value = session_mock session_mock.__enter__.return_value.run.return_value.single.return_value = { 'parent': parent_node } result = tenant.get_tenant_parent_of_tenant(tenant_input) assert result.tenant_uuid == expected_uuid assert result.tenant_type == expected_parent_type assert result.tenant_name == expected_name @pytest.mark.parametrize( ( 'error_scenario', 'tenant_type', 'mock_return', 'expected_exception', 'expected_error_pattern', ), [ pytest.param( 'no_parent_found', TenantType.ACCOUNT, {'parent': None}, 'IncompleteResultError', 'Parent for tenant with uuid: test-uuid, type: account not found', id='no_parent_found', ), pytest.param( 'parent_with_unrecognized_labels', TenantType.SUBACCOUNT, 'unrecognized_labels', # Special marker for creating mock node with unknown labels 'IncompleteResultError', 'Could not determine tenant type for parent with labels', id='parent_with_unrecognized_labels', ), ], ) @mock.patch('permissions.connectors.neo4j.db_session') def test_get_tenant_parent_of_tenant_error_cases( db_session_mock: MagicMock, make_graph_node: MagicMock, error_scenario: str, tenant_type: TenantType, mock_return: dict | str, expected_exception: str, expected_error_pattern: str, ) -> None: """Test get_tenant_parent_of_tenant error cases.""" from permissions.exceptions.incomplete_result_error import IncompleteResultError from permissions.types import TenantWithName tenant_input = TenantWithName( tenant_uuid='test-uuid', tenant_type=tenant_type, tenant_name='Test Tenant' ) session_mock = get_session_mock() db_session_mock.return_value = session_mock if mock_return == 'unrecognized_labels': # Mock parent node with unrecognized labels parent_node = make_graph_node( node_id=104, labels=frozenset(['UnknownLabel', 'AnotherUnknownLabel']), data={'uuid': 'parent-uuid', 'name': 'Test Parent'}, ) session_mock.__enter__.return_value.run.return_value.single.return_value = { 'parent': parent_node } else: session_mock.__enter__.return_value.run.return_value.single.return_value = mock_return with pytest.raises(IncompleteResultError, match=expected_error_pattern): tenant.get_tenant_parent_of_tenant(tenant_input) def test_get_tenant_parent_of_tenant_invalid_tenant_type_raises_error(): """Test get_tenant_parent_of_tenant with invalid tenant type - should raise ValueError.""" from permissions.types import TenantWithName tenant_input = TenantWithName( tenant_uuid='test-uuid', tenant_type='invalid_type', # type: ignore[arg-type] tenant_name='Test Tenant', ) with pytest.raises(ValueError, match='Invalid tenant type'): tenant.get_tenant_parent_of_tenant(tenant_input) # Tests for get_adminable_tenants_for_identities (batch/dataloader version) EXPECTED_QUERY_IDENTITIES_FULL_CATALOG_ACCESS = textwrap.dedent(""" UNWIND $identityUuids as identityUuid MATCH (tenant)<-[:HAS_ACCESS_TO|HAS_ADMIN_ACCESS_TO]-(up:Profile) <-[:HAS_PROFILE]-(user:Identity {id: identityUuid}) WHERE tenant:Vendor OR tenant:SubAccount OR tenant:Collaborator OR tenant:LabelParticipant WITH identityUuid, tenant, COLLECT(DISTINCT { profileType: up.profileType, profileId: up.profileId, roles: up.roles, uuid: up.uuid }) as profiles WITH identityUuid, COLLECT(DISTINCT { tenant_uuid: tenant.uuid, tenant_type: CASE WHEN tenant:Vendor THEN 'account' WHEN tenant:SubAccount THEN 'subaccount' WHEN tenant:Collaborator THEN 'collaborator' WHEN tenant:LabelParticipant THEN 'label_participant' END, profiles: profiles }) as tenants RETURN identityUuid as identity_uuid, tenants """) EXPECTED_QUERY_SEATER_ADMINABLE_TENANTS = textwrap.dedent(""" UNWIND $identityUuids as identityUuid MATCH (tenant)<-[:HAS_ACCESS_TO|HAS_ADMIN_ACCESS_TO]-(up:Profile) <-[:HAS_PROFILE]-(user:Identity {id: identityUuid}) WHERE tenant:Vendor OR tenant:SubAccount OR tenant:Collaborator OR tenant:LabelParticipant OR tenant:ParentCompany WITH identityUuid, tenant, COLLECT(DISTINCT { profileType: up.profileType, profileId: up.profileId, roles: up.roles, uuid: up.uuid }) as profiles WITH identityUuid, COLLECT(DISTINCT { tenant_uuid: tenant.uuid, tenant_type: CASE WHEN tenant:Vendor THEN 'account' WHEN tenant:SubAccount THEN 'subaccount' WHEN tenant:Collaborator THEN 'collaborator' WHEN tenant:LabelParticipant THEN 'label_participant' WHEN tenant:ParentCompany THEN 'parent_company' END, profiles: profiles }) as tenants RETURN identityUuid as identity_uuid, tenants """) EXPECTED_QUERY_IDENTITIES_LIMITED_ACCESS = textwrap.dedent(""" MATCH (admin:Identity)-[:HAS_PROFILE]->(sp:Profile {profileType: $profileType}) -[:HAS_ADMIN_ACCESS_TO]->(t) WHERE admin.id = $adminIdentityId AND sp.profileId = $adminProfileId WITH t, sp UNWIND $identityUuids as identityUuid MATCH (t)-[*0..1]->(tenant) <-[:HAS_ACCESS_TO|HAS_ADMIN_ACCESS_TO]-(up:Profile)<-[:HAS_PROFILE]-(user:Identity {id: identityUuid}) WHERE tenant:Vendor OR tenant:SubAccount OR tenant:Collaborator OR tenant:LabelParticipant WITH identityUuid, tenant, COLLECT(DISTINCT { profileType: up.profileType, profileId: up.profileId, roles: up.roles, uuid: up.uuid }) as profiles WITH identityUuid, COLLECT(DISTINCT { tenant_uuid: tenant.uuid, tenant_type: CASE WHEN tenant:Vendor THEN 'account' WHEN tenant:SubAccount THEN 'subaccount' WHEN tenant:Collaborator THEN 'collaborator' WHEN tenant:LabelParticipant THEN 'label_participant' END, profiles: profiles }) as tenants RETURN identityUuid as identity_uuid, tenants """) @mock.patch('permissions.models.tenant.format_adminable_tenants_for_identities_result') @mock.patch('permissions.models.profile.check_vendor_star_access') @mock.patch('permissions.models.tenant.get_adminable_tenants_for_identities_query') @mock.patch('permissions.connectors.neo4j.db_session') def test_get_adminable_tenants_for_identities_full_catalog_access( db_session_mock, get_query_mock, access_check_mock, format_mock ): """Test get_adminable_tenants_for_identities when admin has full catalog access.""" admin_context = {'identity_id': 'admin-123', 'profile_id': 456} identity_uuids = [ uuid.UUID('fd7b4385-6f42-4205-add6-4c90cc6ec086'), uuid.UUID('97fba315-fa3d-4133-bc49-fc01fddf5671'), uuid.UUID('a1b2c3d4-e5f6-4a5b-8c9d-0e1f2a3b4c5d'), ] session_mock = get_session_mock() db_session_mock.return_value = session_mock query_result = [mock.Mock(), mock.Mock(), mock.Mock()] session_mock.__enter__.return_value.run.return_value = query_result access_check_mock.return_value = True format_result = [ {'identity_uuid': 'user-aaa', 'tenants': [{'tenant_uuid': 'tenant-1'}]}, {'identity_uuid': 'user-bbb', 'tenants': []}, {'identity_uuid': 'user-ccc', 'tenants': [{'tenant_uuid': 'tenant-2'}]}, ] format_mock.return_value = format_result result = tenant.get_adminable_tenants_for_identities( identity_uuids=identity_uuids, admin_context=admin_context, has_full_catalog_access=True, ) assert result == format_result db_session_mock.assert_called_with(access_mode='READ') access_check_mock.assert_not_called() # Should not check access when admin has full catalog access get_query_mock.assert_called_with(True) # Verify that UUID objects are converted to strings before being passed to the query session_mock.__enter__.return_value.run.assert_called_with( get_query_mock.return_value, profileType='SettingsProfile', adminIdentityId='admin-123', adminProfileId=456, identityUuids=[str(uuid_obj) for uuid_obj in identity_uuids], ) format_mock.assert_called_with(query_result) @mock.patch('permissions.models.tenant.format_adminable_tenants_for_identities_result') @mock.patch('permissions.models.profile.check_vendor_star_access') @mock.patch('permissions.models.tenant.get_adminable_tenants_for_identities_query') @mock.patch('permissions.connectors.neo4j.db_session') def test_get_adminable_tenants_for_identities_limited_access( db_session_mock, get_query_mock, access_check_mock, format_mock ): """Test get_adminable_tenants_for_identities when admin has limited access.""" admin_context = {'identity_id': 'admin-123', 'profile_id': 456} identity_uuids = [ uuid.UUID('fd7b4385-6f42-4205-add6-4c90cc6ec086'), uuid.UUID('97fba315-fa3d-4133-bc49-fc01fddf5671'), ] session_mock = get_session_mock() db_session_mock.return_value = session_mock query_result = [mock.Mock(), mock.Mock()] session_mock.__enter__.return_value.run.return_value = query_result access_check_mock.return_value = False tenant.get_adminable_tenants_for_identities( identity_uuids=identity_uuids, admin_context=admin_context, has_full_catalog_access=False, ) get_query_mock.assert_called_with(False) # Verify that UUID objects are converted to strings before being passed to the query session_mock.__enter__.return_value.run.assert_called_with( get_query_mock.return_value, profileType='SettingsProfile', adminIdentityId='admin-123', adminProfileId=456, identityUuids=[str(uuid_obj) for uuid_obj in identity_uuids], ) def test_get_adminable_tenants_for_identities_query_full_catalog_access(): """Test get_adminable_tenants_for_identities_query when admin has full catalog access.""" result = tenant.get_adminable_tenants_for_identities_query(has_full_catalog_access=True) assert result == EXPECTED_QUERY_IDENTITIES_FULL_CATALOG_ACCESS def test_get_adminable_tenants_for_identities_query_limited_access(): """Test get_adminable_tenants_for_identities_query when admin has limited access.""" result = tenant.get_adminable_tenants_for_identities_query(has_full_catalog_access=False) assert result == EXPECTED_QUERY_IDENTITIES_LIMITED_ACCESS @mock.patch('permissions.models.tenant.format_adminable_tenants_for_identities_result') @mock.patch('permissions.connectors.neo4j.db_session') def test_get_seater_adminable_tenants_for_identities(db_session_mock, format_mock): """Test get_seater_adminable_tenants_for_identities runs the seater query.""" admin_context = {'identity_id': 'admin-123', 'profile_id': 456} identity_uuids = [ uuid.UUID('fd7b4385-6f42-4205-add6-4c90cc6ec086'), uuid.UUID('97fba315-fa3d-4133-bc49-fc01fddf5671'), ] session_mock = get_session_mock() db_session_mock.return_value = session_mock query_result = [mock.Mock(), mock.Mock()] session_mock.__enter__.return_value.run.return_value = query_result format_result = [ {'identity_uuid': 'user-aaa', 'tenants': [{'tenant_uuid': 'tenant-1'}]}, {'identity_uuid': 'user-bbb', 'tenants': []}, ] format_mock.return_value = format_result result = tenant.get_seater_adminable_tenants_for_identities( identity_uuids=identity_uuids, admin_context=admin_context, ) assert result == format_result db_session_mock.assert_called_with(access_mode='READ') session_mock.__enter__.return_value.run.assert_called_with( EXPECTED_QUERY_SEATER_ADMINABLE_TENANTS, profileType='SettingsProfile', adminIdentityId='admin-123', adminProfileId=456, identityUuids=[str(uid) for uid in identity_uuids], ) format_mock.assert_called_with(query_result) def test_format_adminable_tenants_for_identities_result(): """Test format_adminable_tenants_for_identities_result formats query result correctly.""" query_result = [ { 'identity_uuid': '550e8400-e29b-41d4-a716-446655440000', 'tenants': [ { 'tenant_uuid': 'tenant-123', 'tenant_type': 'account', 'profiles': [ { 'profileId': 1, 'profileType': 'SettingsProfile', 'roles': ['admin'], 'uuid': 'profile-abc', }, { 'profileId': 2, 'profileType': 'InsightsProfile', 'roles': ['analytics'], 'uuid': 'profile-def', }, ], }, { 'tenant_uuid': 'd4cabe5c-54e0-4967-84ea-f494244219a9', 'tenant_type': 'subaccount', 'profiles': [ { 'profileId': 3, 'profileType': 'SettingsProfile', 'roles': [], 'uuid': 'profile-ghi', }, ], }, ], }, { 'identity_uuid': '28ee7930-18ec-4459-964e-a95dd151933a', 'tenants': [ { 'tenant_uuid': 'tenant-789', 'tenant_type': 'collaborator', 'profiles': [ { 'profileId': 4, 'profileType': 'InsightsProfile', 'roles': ['viewer'], 'uuid': 'profile-jkl', }, ], }, ], }, ] result = tenant.format_adminable_tenants_for_identities_result(query_result) assert len(result) == 2 assert result[0].identity_uuid == '550e8400-e29b-41d4-a716-446655440000' assert len(result[0].tenants) == 2 assert result[0].tenants[0].tenant.tenant_uuid == 'tenant-123' assert result[0].tenants[0].tenant.tenant_type == 'account' assert len(result[0].tenants[0].profiles) == 2 assert result[0].tenants[0].profiles[0] == ProfileInfo( profile_id=1, profile_type='SettingsProfile', roles=['admin'], uuid='profile-abc' ) assert result[0].tenants[0].profiles[1] == ProfileInfo( profile_id=2, profile_type='InsightsProfile', roles=['analytics'], uuid='profile-def' ) assert result[0].tenants[1].tenant.tenant_uuid == 'd4cabe5c-54e0-4967-84ea-f494244219a9' assert result[0].tenants[1].tenant.tenant_type == 'subaccount' assert result[1].identity_uuid == '28ee7930-18ec-4459-964e-a95dd151933a' assert len(result[1].tenants) == 1 assert result[1].tenants[0].tenant.tenant_uuid == 'tenant-789' assert result[1].tenants[0].tenant.tenant_type == 'collaborator' def test_format_adminable_tenants_for_identities_result_filters_null_tenants(): """Test that format_adminable_tenants_for_identities_result filters out null tenant_uuids.""" query_result = [ { 'identity_uuid': 'a1b2c3d4-e5f6-4a5b-8c9d-0e1f2a3b4c5d', 'tenants': [ { 'tenant_uuid': 'tenant-123', 'tenant_type': 'account', 'profiles': [ { 'profileId': 1, 'profileType': 'SettingsProfile', 'roles': [], 'uuid': 'profile-abc', }, ], }, { 'tenant_uuid': None, # Should be filtered out 'tenant_type': None, 'profiles': [], }, ], }, { 'identity_uuid': 'f6e5d4c3-b2a1-4f5e-9d8c-7b6a5e4d3c2b', 'tenants': [ { 'tenant_uuid': None, # All tenants null, should result in empty list 'tenant_type': None, 'profiles': [], }, ], }, ] result = tenant.format_adminable_tenants_for_identities_result(query_result) assert len(result) == 2 assert result[0].identity_uuid == 'a1b2c3d4-e5f6-4a5b-8c9d-0e1f2a3b4c5d' assert len(result[0].tenants) == 1 assert result[0].tenants[0].tenant.tenant_uuid == 'tenant-123' assert result[1].identity_uuid == 'f6e5d4c3-b2a1-4f5e-9d8c-7b6a5e4d3c2b' assert len(result[1].tenants) == 0 def test_format_adminable_tenants_for_identities_result_filters_null_profiles(): """Test that format_adminable_tenants_for_identities_result filters out profiles with null profileId.""" query_result = [ { 'identity_uuid': '9e8d7c6b-5a4f-4e3d-9c2b-1a0e9d8c7b6a', 'tenants': [ { 'tenant_uuid': 'tenant-123', 'tenant_type': 'account', 'profiles': [ { 'profileId': 1, 'profileType': 'SettingsProfile', 'roles': [], 'uuid': 'profile-abc', }, { 'profileId': None, # Should be filtered out 'profileType': None, 'roles': None, 'uuid': None, }, ], }, ], }, ] result = tenant.format_adminable_tenants_for_identities_result(query_result) assert len(result) == 1 assert result[0].identity_uuid == '9e8d7c6b-5a4f-4e3d-9c2b-1a0e9d8c7b6a' assert len(result[0].tenants) == 1 assert len(result[0].tenants[0].profiles) == 1 assert result[0].tenants[0].profiles[0].profile_id == 1 def test_format_adminable_tenants_for_identities_result_empty_input(): """Test format_adminable_tenants_for_identities_result with empty input.""" result = tenant.format_adminable_tenants_for_identities_result([]) assert result == [] @mock.patch('permissions.models.tenant.neo4j_connector') def test_seat_get_tenant_by_uuid_returns_none_when_not_found(neo4j_connector_mock) -> None: """Test seat_get_tenant_by_uuid returns None when tenant is not found.""" identity_id = 'test-identity-id' tenant_uuid = 'nonexistent-tenant-uuid' session_mock = mock.Mock() session_mock.run.return_value.single.return_value = None neo4j_connector_mock.db_session.return_value.__enter__.return_value = session_mock result = tenant.seat_get_tenant_by_uuid(identity_id, tenant_uuid) expected_query = textwrap.dedent(""" MATCH (tenant)<-[:HAS_ACCESS_TO|HAS_ADMIN_ACCESS_TO]-(up:Profile) <-[:HAS_PROFILE]-(user:Identity {id: $identityId}) WHERE tenant.uuid = $tenantUuid AND ( tenant:Vendor OR tenant:CompanyBrand OR tenant:ParentCompany ) WITH tenant, COLLECT(DISTINCT { profileType: up.profileType, profileId: up.profileId, roles: up.roles, uuid: up.uuid }) as profiles RETURN DISTINCT tenant, profiles LIMIT 1 """) session_mock.run.assert_called_once_with( expected_query, identityId=identity_id, tenantUuid=tenant_uuid ) assert result is None @mock.patch('permissions.models.tenant.neo4j_connector') def test_seat_get_tenant_by_uuid(neo4j_connector_mock) -> None: """Test seat_get_tenant_by_uuid returns AdminableTenant for ParentCompany.""" identity_id = 'test-identity-id' tenant_uuid = 'parent-company-uuid' tenant_node = mock.Mock(labels=['ParentCompany']) tenant_node.get.return_value = tenant_uuid profiles_data = [ { 'profileId': 789, 'profileType': 'SettingsProfile', 'roles': [], 'uuid': 'profile-uuid-3', }, ] session_mock = mock.Mock() session_mock.run.return_value.single.return_value = { 'tenant': tenant_node, 'profiles': profiles_data, } neo4j_connector_mock.db_session.return_value.__enter__.return_value = session_mock result = tenant.seat_get_tenant_by_uuid(identity_id, tenant_uuid) assert result.tenant.tenant_type == TenantType.PARENT_COMPANY assert result.tenant.tenant_uuid == tenant_uuid assert len(result.profiles) == 1 assert result.profiles[0].profile_id == 789