"""Performance tests for permission system — visibility filter, caching. Covers: - Visibility filter performance with 1000 mock contacts - Cache hit vs miss performance comparison """ from __future__ import annotations import time import uuid from unittest.mock import AsyncMock, MagicMock, patch import pytest import pytest_asyncio from sqlalchemy.ext.asyncio import AsyncSession from app.models.contact import Contact from app.services import entity_permission_service as eps from tests.conftest import seed_tenant_and_users @pytest.mark.asyncio class TestPermissionPerformance: """Performance tests for permission system.""" async def test_visibility_filter_performance(self, db_session: AsyncSession): """get_visible_ids with 1000 contacts completes in reasonable time.""" seed = await seed_tenant_and_users(db_session) tenant_id = seed["tenant_a"].id user_id = seed["admin_a"].id # Create 1000 contacts owned by the user contacts = [] for i in range(1000): contacts.append(Contact( tenant_id=tenant_id, type="person", firstname=f"Perf{i}", surname=f"Test{i}", owner_id=user_id, created_by=user_id, updated_by=user_id, )) db_session.add_all(contacts) await db_session.commit() # Measure time for get_visible_ids start = time.perf_counter() visible, access_map = await eps.get_visible_ids( db_session, tenant_id, user_id, "contact" ) elapsed = time.perf_counter() - start # Should complete in under 2 seconds for 1000 contacts assert elapsed < 2.0, f"get_visible_ids took {elapsed:.3f}s (expected <2.0s)" assert len(visible) >= 1000, f"Expected >=1000 visible, got {len(visible)}" # All should be at 'owner' level for eid, level in access_map.items(): if eid in [c.id for c in contacts]: assert level == "owner", f"Expected 'owner', got '{level}'" async def test_cache_hit_vs_miss(self, db_session: AsyncSession, redis_client): """Cache hit is significantly faster than cache miss.""" seed = await seed_tenant_and_users(db_session) tenant_id = seed["tenant_a"].id user_id = seed["admin_a"].id # Create 100 contacts owned by the user contacts = [] for i in range(100): contacts.append(Contact( tenant_id=tenant_id, type="person", firstname=f"Cache{i}", surname=f"Test{i}", owner_id=user_id, created_by=user_id, updated_by=user_id, )) db_session.add_all(contacts) await db_session.commit() # First call — cache miss (populates cache) miss_start = time.perf_counter() visible_miss, access_map_miss = await eps.get_cached_visible_ids( db_session, redis_client, tenant_id, user_id, "contact" ) miss_elapsed = time.perf_counter() - miss_start assert len(visible_miss) >= 100, f"Expected >=100 visible, got {len(visible_miss)}" # Second call — cache hit hit_start = time.perf_counter() visible_hit, access_map_hit = await eps.get_cached_visible_ids( db_session, redis_client, tenant_id, user_id, "contact" ) hit_elapsed = time.perf_counter() - hit_start # Cache hit should be faster (at least 2x speedup) assert hit_elapsed < miss_elapsed, f"Cache hit {hit_elapsed:.4f}s should be faster than miss {miss_elapsed:.4f}s" assert len(visible_hit) == len(visible_miss), "Cache hit should return same results" async def test_batch_resolution_performance(self, db_session: AsyncSession): """batch_get_effective_access with 100 entities completes quickly.""" seed = await seed_tenant_and_users(db_session) tenant_id = seed["tenant_a"].id user_id = seed["admin_a"].id # Create 100 contacts contacts = [] for i in range(100): contacts.append(Contact( tenant_id=tenant_id, type="person", firstname=f"Batch{i}", surname=f"Test{i}", owner_id=user_id, created_by=user_id, updated_by=user_id, )) db_session.add_all(contacts) await db_session.commit() entity_ids = [c.id for c in contacts] start = time.perf_counter() result = await eps.batch_get_effective_access( db_session, tenant_id, user_id, "contact", entity_ids ) elapsed = time.perf_counter() - start assert elapsed < 1.0, f"batch_get_effective_access took {elapsed:.3f}s (expected <1.0s)" assert len(result) == 100, f"Expected 100 results, got {len(result)}" # All should be 'owner' for eid, level in result.items(): assert level == "owner", f"Expected 'owner', got '{level}'" async def test_cache_invalidation_performance(self, db_session: AsyncSession, redis_client): """Cache invalidation for all entity types completes quickly.""" seed = await seed_tenant_and_users(db_session) tenant_id = seed["tenant_a"].id user_id = seed["admin_a"].id # Populate cache for multiple entity types for entity_type in ["contact", "company", "attachment", "entity_attachment", "workflow"]: await eps.get_cached_visible_ids( db_session, redis_client, tenant_id, user_id, entity_type ) # Measure invalidation time start = time.perf_counter() await eps.invalidate_all_user_entity_cache(redis_client, tenant_id, user_id) elapsed = time.perf_counter() - start assert elapsed < 1.0, f"Cache invalidation took {elapsed:.3f}s (expected <1.0s)" # Verify cache is cleared for entity_type in ["contact", "company", "attachment", "entity_attachment", "workflow"]: cache_key = f"ep_vis:{user_id}:{tenant_id}:{entity_type}" cached = await redis_client.get(cache_key) assert cached is None, f"Cache for {entity_type} should be cleared" async def test_get_effective_access_performance(self, db_session: AsyncSession): """get_effective_access for a single entity completes quickly.""" seed = await seed_tenant_and_users(db_session) tenant_id = seed["tenant_a"].id user_id = seed["admin_a"].id contact_id = seed["company_a"].id # Set owner_id contact = await db_session.get(Contact, contact_id) contact.owner_id = user_id await db_session.commit() # Measure single access check start = time.perf_counter() for _ in range(100): access = await eps.get_effective_access( db_session, tenant_id, user_id, "contact", contact_id ) elapsed = time.perf_counter() - start avg_ms = (elapsed / 100) * 1000 assert avg_ms < 50, f"Average get_effective_access took {avg_ms:.2f}ms (expected <50ms)" assert access == "owner" async def test_check_entity_access_performance(self, db_session: AsyncSession): """check_entity_access for a single entity completes quickly.""" seed = await seed_tenant_and_users(db_session) tenant_id = seed["tenant_a"].id user_id = seed["admin_a"].id contact_id = seed["company_a"].id # Set owner_id contact = await db_session.get(Contact, contact_id) contact.owner_id = user_id await db_session.commit() # Measure single access check start = time.perf_counter() for _ in range(100): result = await eps.check_entity_access( db_session, tenant_id, user_id, "contact", contact_id, "read" ) elapsed = time.perf_counter() - start avg_ms = (elapsed / 100) * 1000 assert avg_ms < 50, f"Average check_entity_access took {avg_ms:.2f}ms (expected <50ms)" assert result is True async def test_permission_creation_performance(self, db_session: AsyncSession): """Creating permissions in bulk completes quickly.""" seed = await seed_tenant_and_users(db_session) tenant_id = seed["tenant_a"].id owner_id = seed["admin_a"].id viewer_id = seed["viewer_a"].id contact_id = seed["company_a"].id # Set owner_id contact = await db_session.get(Contact, contact_id) contact.owner_id = owner_id await db_session.commit() # Measure bulk permission creation start = time.perf_counter() for level in ["read", "write", "admin", "delete"]: await eps.create_permission( db_session, tenant_id=tenant_id, entity_type="contact", entity_id=str(contact_id), principal_type="user", principal_id=str(viewer_id), permission_level=level, created_by=owner_id, ) elapsed = time.perf_counter() - start avg_ms = (elapsed / 4) * 1000 assert avg_ms < 200, f"Average permission creation took {avg_ms:.2f}ms (expected <200ms)" async def test_visible_ids_with_mixed_ownership(self, db_session: AsyncSession): """get_visible_ids with mixed ownership (owned + tenant + shared) performs well.""" seed = await seed_tenant_and_users(db_session) tenant_id = seed["tenant_a"].id owner_id = seed["admin_a"].id viewer_id = seed["viewer_a"].id # Create 500 owned, 300 tenant-owned, 200 shared contacts contacts = [] for i in range(500): contacts.append(Contact( tenant_id=tenant_id, type="person", firstname=f"Owned{i}", surname=f"Test{i}", owner_id=owner_id, created_by=owner_id, updated_by=owner_id, )) for i in range(300): contacts.append(Contact( tenant_id=tenant_id, type="person", firstname=f"Tenant{i}", surname=f"Test{i}", owner_id=None, created_by=owner_id, updated_by=owner_id, )) db_session.add_all(contacts) await db_session.commit() # Grant viewer access to 200 contacts shared_ids = [c.id for c in contacts[:200]] for cid in shared_ids: await eps.create_permission( db_session, tenant_id=tenant_id, entity_type="contact", entity_id=str(cid), principal_type="user", principal_id=str(viewer_id), permission_level="read", created_by=owner_id, ) # Measure performance for viewer start = time.perf_counter() visible, access_map = await eps.get_visible_ids( db_session, tenant_id, viewer_id, "contact" ) elapsed = time.perf_counter() - start assert elapsed < 3.0, f"get_visible_ids with mixed ownership took {elapsed:.3f}s (expected <3.0s)" # Viewer should see: 200 shared + 300 tenant-owned = 500 assert len(visible) >= 500, f"Expected >=500 visible, got {len(visible)}"