""" Performance tests for the Atlas AI Chat API Consolidated from: - test_performance_user_auth.py - Performance portions of other test files """ import asyncio from datetime import datetime, timedelta import os import time import psutil import pytest import random from analytics.collectors import create_session, track_message, track_search from analytics.dashboard import get_authenticated_vs_anonymous_metrics, get_basic_stats, get_hourly_message_stats, get_user_analytics, get_user_statistics from analytics.database import get_messages_collection, get_sessions_collection from tests.utilities import MockDataGenerator, PerformanceHelpers, TestHelpers, create_test_user_data, skip_if_no_database TestHelpers, PerformanceHelpers, MockDataGenerator, skip_if_no_database, create_test_user_data ) class TestDatabaseIndexPerformance: """Test performance of database indexes for user_id queries""" @pytest.mark.asyncio @skip_if_no_database() async def test_user_id_index_performance(self): """Test performance of user_id index queries""" # Create test data with multiple users user_ids = [f"perf_user_{i}" for i in range(10)] sessions_per_user = 2 messages_per_session = 3 # Create test data start_time = time.time() for user_id in user_ids: await create_test_user_data(user_id, sessions_per_user, messages_per_session) data_creation_time = time.time() - start_time # Wait for data to be persisted await TestHelpers.wait_for_data_persistence(2.0) # Test query performance sessions_collection = await get_sessions_collection() messages_collection = await get_messages_collection() if sessions_collection and messages_collection: # Test individual user queries start_time = time.time() for user_id in user_ids: user_sessions = await sessions_collection.count_documents({"user_id": user_id}) assert user_sessions >= sessions_per_user user_messages = await messages_collection.count_documents({"user_id": user_id}) assert user_messages >= sessions_per_user * messages_per_session individual_query_time = time.time() - start_time avg_query_time = individual_query_time / len(user_ids) # Performance assertion assert avg_query_time < 0.5, f"Individual queries too slow: {avg_query_time:.4f}s average" # Test bulk queries start_time = time.time() # Query all authenticated sessions auth_sessions = await sessions_collection.count_documents({"user_id": {"$ne": None}}) # Query all authenticated messages auth_messages = await messages_collection.count_documents({"user_id": {"$ne": None}}) bulk_query_time = time.time() - start_time # Performance assertion assert bulk_query_time < 3.0, f"Bulk queries too slow: {bulk_query_time:.2f}s" # Verify we got expected results assert auth_sessions >= len(user_ids) * sessions_per_user assert auth_messages >= len(user_ids) * sessions_per_user * messages_per_session @pytest.mark.asyncio @skip_if_no_database() async def test_compound_index_performance(self): """Test performance of compound (user_id, timestamp) index queries""" # Create test data with timestamps spread over time user_id = "compound_perf_user" session = await create_session(user_id=user_id) num_messages = 20 for i in range(num_messages): await track_message( session_id=session.session_id, prompt_length=50, response_length=100, response_time_ms=1000, user_id=user_id ) # Small delay to ensure different timestamps await asyncio.sleep(0.01) # Wait for data to be persisted await TestHelpers.wait_for_data_persistence(2.0) # Test compound queries messages_collection = await get_messages_collection() if messages_collection: # Test various time range queries time_ranges = [ ("1 hour", timedelta(hours=1)), ("6 hours", timedelta(hours=6)), ("24 hours", timedelta(hours=24)) ] for range_name, time_delta in time_ranges: start_time = time.time() cutoff_time = datetime.utcnow() - time_delta recent_messages = await messages_collection.count_documents({ "user_id": user_id, "timestamp": {"$gte": cutoff_time} }) query_time = time.time() - start_time # Performance assertion assert query_time < 1.0, f"{range_name} query too slow: {query_time:.4f}s" # Should find our messages assert recent_messages >= num_messages @pytest.mark.asyncio @skip_if_no_database() async def test_sparse_index_performance(self): """Test performance of sparse indexes with mixed null/non-null user_id values""" # Create mixed data (authenticated and anonymous) num_auth_users = 5 num_anon_sessions = 10 messages_per_session = 3 # Create authenticated user data for i in range(num_auth_users): user_id = f"sparse_user_{i}" await create_test_user_data(user_id, 1, messages_per_session) # Create anonymous user data for i in range(num_anon_sessions): session = await create_session(user_id=None) for j in range(messages_per_session): await track_message( session_id=session.session_id, prompt_length=50, response_length=100, response_time_ms=1000, user_id=None ) # Wait for data to be persisted await TestHelpers.wait_for_data_persistence(2.0) # Test sparse index queries sessions_collection = await get_sessions_collection() messages_collection = await get_messages_collection() if sessions_collection and messages_collection: # Test authenticated user queries start_time = time.time() auth_session_count = await sessions_collection.count_documents({"user_id": {"$ne": None}}) auth_query_time = time.time() - start_time # Test anonymous user queries start_time = time.time() anon_session_count = await sessions_collection.count_documents({"user_id": None}) anon_query_time = time.time() - start_time # Test specific user queries start_time = time.time() specific_user_sessions = await sessions_collection.count_documents({"user_id": "sparse_user_0"}) specific_query_time = time.time() - start_time # Performance assertions assert auth_query_time < 1.0, f"Auth query too slow: {auth_query_time:.4f}s" assert anon_query_time < 1.0, f"Anon query too slow: {anon_query_time:.4f}s" assert specific_query_time < 0.5, f"Specific query too slow: {specific_query_time:.4f}s" # Verify results assert auth_session_count >= num_auth_users assert anon_session_count >= num_anon_sessions assert specific_user_sessions >= 1 class TestAnalyticsFunctionPerformance: """Test performance of analytics functions with user authentication""" @pytest.mark.asyncio async def test_basic_stats_performance(self): """Test performance of get_basic_stats function""" # Create some test data await self._create_performance_test_data() # Test get_basic_stats performance start_time = time.time() stats = await get_basic_stats() stats_time = time.time() - start_time assert isinstance(stats, dict) assert "total_sessions" in stats assert "total_messages" in stats # Performance assertion assert stats_time < 5.0, f"Basic stats too slow: {stats_time:.4f}s" @pytest.mark.asyncio async def test_user_statistics_performance(self): """Test performance of get_user_statistics function""" # Create test data await self._create_performance_test_data() # Test get_user_statistics performance start_time = time.time() user_stats = await get_user_statistics() stats_time = time.time() - start_time assert isinstance(user_stats, dict) assert "unique_authenticated_users" in user_stats assert "authenticated_sessions" in user_stats # Performance assertion assert stats_time < 8.0, f"User statistics too slow: {stats_time:.4f}s" @pytest.mark.asyncio async def test_user_analytics_performance(self): """Test performance of get_user_analytics function""" # Create test user with substantial data user_id = "analytics_perf_user" await create_test_user_data(user_id, num_sessions=2, messages_per_session=10) # Wait for data to be persisted await TestHelpers.wait_for_data_persistence() # Test get_user_analytics performance start_time = time.time() user_analytics = await get_user_analytics(user_id) analytics_time = time.time() - start_time assert isinstance(user_analytics, dict) assert user_analytics.get("user_id") == user_id # Performance assertion assert analytics_time < 5.0, f"User analytics too slow: {analytics_time:.4f}s" @pytest.mark.asyncio async def test_comparison_metrics_performance(self): """Test performance of get_authenticated_vs_anonymous_metrics function""" # Create mixed test data await self._create_performance_test_data() # Test get_authenticated_vs_anonymous_metrics performance start_time = time.time() comparison_metrics = await get_authenticated_vs_anonymous_metrics() comparison_time = time.time() - start_time assert isinstance(comparison_metrics, dict) assert "authenticated" in comparison_metrics assert "anonymous" in comparison_metrics # Performance assertion assert comparison_time < 8.0, f"Comparison metrics too slow: {comparison_time:.4f}s" @pytest.mark.asyncio async def test_hourly_stats_performance(self): """Test performance of get_hourly_message_stats function""" # Create test data await self._create_performance_test_data() # Test hourly stats performance start_time = time.time() hourly_stats = await get_hourly_message_stats(hours=24) hourly_time = time.time() - start_time assert isinstance(hourly_stats, list) assert len(hourly_stats) == 24 # Performance assertion assert hourly_time < 5.0, f"Hourly stats too slow: {hourly_time:.4f}s" async def _create_performance_test_data(self): """Create test data for performance testing""" # Create authenticated users for i in range(3): user_id = f"perf_test_user_{i}" await create_test_user_data(user_id, num_sessions=1, messages_per_session=5) # Create anonymous users for i in range(2): session = await create_session(user_id=None) # Create messages for anonymous users for j in range(3): await track_message( session_id=session.session_id, prompt_length=random.randint(20, 100), response_length=random.randint(50, 200), response_time_ms=random.randint(500, 2000), used_search=random.choice([True, False]), user_id=None ) class TestConcurrentUserPerformance: """Test performance with concurrent user operations""" @pytest.mark.asyncio async def test_concurrent_user_creation(self): """Test performance of concurrent user session creation""" num_concurrent_users = 10 async def create_user_session(user_id: str): session = await create_session(user_id=user_id) # Create a few messages for each user for i in range(2): await track_message( session_id=session.session_id, prompt_length=50, response_length=100, response_time_ms=1000, user_id=user_id ) return session # Create concurrent tasks start_time = time.time() tasks = [ create_user_session(f"concurrent_user_{i}") for i in range(num_concurrent_users) ] sessions = await asyncio.gather(*tasks) concurrent_time = time.time() - start_time assert len(sessions) == num_concurrent_users # Performance assertion avg_time_per_user = concurrent_time / num_concurrent_users assert avg_time_per_user < 2.0, f"Concurrent creation too slow: {avg_time_per_user:.2f}s per user" @pytest.mark.asyncio async def test_concurrent_user_queries(self): """Test performance of concurrent user-specific queries""" # Create test users first user_ids = [f"query_user_{i}" for i in range(5)] for user_id in user_ids: await create_test_user_data(user_id, num_sessions=1, messages_per_session=2) # Wait for data to be persisted await TestHelpers.wait_for_data_persistence() # Test concurrent queries async def query_user_analytics(user_id: str): return await get_user_analytics(user_id) start_time = time.time() tasks = [query_user_analytics(user_id) for user_id in user_ids] results = await asyncio.gather(*tasks) concurrent_query_time = time.time() - start_time assert len(results) == len(user_ids) for i, result in enumerate(results): assert result.get("user_id") == user_ids[i] # Performance assertion avg_query_time = concurrent_query_time / len(user_ids) assert avg_query_time < 2.0, f"Concurrent queries too slow: {avg_query_time:.2f}s per query" @pytest.mark.asyncio async def test_concurrent_mixed_operations(self): """Test performance of mixed concurrent operations""" # Define different types of operations async def create_user_data(user_id: str): session = await create_session(user_id=user_id) await track_message( session_id=session.session_id, prompt_length=50, response_length=100, response_time_ms=1000, user_id=user_id ) return f"created_{user_id}" async def query_basic_stats(): stats = await get_basic_stats() return f"stats_{stats['total_sessions']}" async def query_user_stats(): stats = await get_user_statistics() return f"user_stats_{stats['total_sessions']}" # Create mixed operations operations = [] # Add user creation operations for i in range(3): operations.append(create_user_data(f"mixed_user_{i}")) # Add query operations operations.append(query_basic_stats()) operations.append(query_user_stats()) # Execute concurrently start_time = time.time() results = await asyncio.gather(*operations) total_time = time.time() - start_time assert len(results) == len(operations) # Performance assertion avg_operation_time = total_time / len(operations) assert avg_operation_time < 3.0, f"Mixed operations too slow: {avg_operation_time:.2f}s per operation" class TestMemoryPerformance: """Test memory usage with user authentication""" @pytest.mark.asyncio async def test_memory_usage_with_users(self): """Test that user_id fields don't significantly increase memory usage""" # Get initial memory usage process = psutil.Process(os.getpid()) initial_memory = process.memory_info().rss / 1024 / 1024 # MB # Create substantial amount of data num_users = 10 messages_per_user = 5 for i in range(num_users): user_id = f"memory_test_user_{i}" await create_test_user_data(user_id, num_sessions=1, messages_per_session=messages_per_user) # Get final memory usage final_memory = process.memory_info().rss / 1024 / 1024 # MB memory_increase = final_memory - initial_memory # Memory increase should be reasonable total_records = num_users * (1 + messages_per_user) # sessions + messages memory_per_record = memory_increase / total_records if total_records > 0 else 0 # Performance assertion (should be less than 2MB per record) assert memory_per_record < 2.0, f"Memory usage too high: {memory_per_record:.3f}MB per record" @pytest.mark.asyncio async def test_memory_usage_with_large_dataset(self): """Test memory usage with larger dataset""" # Get initial memory usage process = psutil.Process(os.getpid()) initial_memory = process.memory_info().rss / 1024 / 1024 # MB # Create larger dataset num_users = 20 for i in range(num_users): user_id = f"large_memory_test_user_{i}" session = await create_session(user_id=user_id) # Create multiple messages per user for j in range(3): await track_message( session_id=session.session_id, prompt_length=random.randint(50, 200), response_length=random.randint(100, 500), response_time_ms=random.randint(500, 3000), user_id=user_id ) # Get final memory usage final_memory = process.memory_info().rss / 1024 / 1024 # MB memory_increase = final_memory - initial_memory # Memory increase should be reasonable for the amount of data total_records = num_users * 4 # 1 session + 3 messages per user memory_per_record = memory_increase / total_records if total_records > 0 else 0 # Performance assertion assert memory_per_record < 3.0, f"Large dataset memory usage too high: {memory_per_record:.3f}MB per record" class TestScalabilityPerformance: """Test scalability with increasing data volumes""" @pytest.mark.asyncio @skip_if_no_database() async def test_query_performance_with_scale(self): """Test that query performance doesn't degrade significantly with more data""" # Create baseline data and measure performance baseline_user = "scale_baseline_user" await create_test_user_data(baseline_user, num_sessions=1, messages_per_session=5) # Wait for data to be persisted await TestHelpers.wait_for_data_persistence() # Measure baseline query performance start_time = time.time() baseline_analytics = await get_user_analytics(baseline_user) baseline_time = time.time() - start_time # Create more data (simulate scale) for i in range(5): scale_user = f"scale_user_{i}" await create_test_user_data(scale_user, num_sessions=2, messages_per_session=10) # Wait for data to be persisted await TestHelpers.wait_for_data_persistence() # Measure performance with more data start_time = time.time() scaled_analytics = await get_user_analytics(baseline_user) scaled_time = time.time() - start_time # Performance should not degrade significantly performance_ratio = scaled_time / baseline_time if baseline_time > 0 else 1 assert performance_ratio < 3.0, f"Performance degraded too much: {performance_ratio:.2f}x slower" # Results should be consistent assert baseline_analytics["user_id"] == scaled_analytics["user_id"] assert baseline_analytics["total_sessions"] == scaled_analytics["total_sessions"] @pytest.mark.asyncio async def test_analytics_performance_with_scale(self): """Test analytics function performance with increasing data""" # Measure performance with small dataset small_users = 2 for i in range(small_users): user_id = f"small_scale_user_{i}" await create_test_user_data(user_id, num_sessions=1, messages_per_session=2) start_time = time.time() small_stats = await get_user_statistics() small_time = time.time() - start_time # Add more data additional_users = 5 for i in range(additional_users): user_id = f"large_scale_user_{i}" await create_test_user_data(user_id, num_sessions=1, messages_per_session=3) # Measure performance with larger dataset start_time = time.time() large_stats = await get_user_statistics() large_time = time.time() - start_time # Performance should scale reasonably data_ratio = (small_users + additional_users) / small_users performance_ratio = large_time / small_time if small_time > 0 else 1 # Performance should not degrade more than linearly with data size assert performance_ratio < data_ratio * 2, f"Performance scaling too poor: {performance_ratio:.2f}x for {data_ratio:.2f}x data" # Results should reflect the additional data assert large_stats["unique_authenticated_users"] >= small_stats["unique_authenticated_users"] assert large_stats["total_sessions"] >= small_stats["total_sessions"] class TestPerformanceBenchmarks: """Benchmark tests for performance regression detection""" @pytest.mark.asyncio async def test_user_creation_benchmark(self): """Benchmark user creation performance""" num_iterations = 10 times = [] for i in range(num_iterations): user_id = f"benchmark_user_{i}" start_time = time.time() session = await create_session(user_id=user_id) await track_message( session_id=session.session_id, prompt_length=50, response_length=100, response_time_ms=1000, user_id=user_id ) end_time = time.time() times.append(end_time - start_time) # Calculate statistics avg_time = sum(times) / len(times) max_time = max(times) min_time = min(times) # Benchmark assertions assert avg_time < 1.0, f"Average user creation too slow: {avg_time:.3f}s" assert max_time < 3.0, f"Worst case user creation too slow: {max_time:.3f}s" assert min_time < 0.5, f"Best case user creation too slow: {min_time:.3f}s" @pytest.mark.asyncio async def test_analytics_query_benchmark(self): """Benchmark analytics query performance""" # Create test data for i in range(3): user_id = f"analytics_benchmark_user_{i}" await create_test_user_data(user_id, num_sessions=1, messages_per_session=3) # Wait for data to be persisted await TestHelpers.wait_for_data_persistence() # Benchmark different analytics functions functions_to_test = [ ("basic_stats", get_basic_stats), ("user_statistics", get_user_statistics), ] for func_name, func in functions_to_test: times = [] # Run multiple iterations for i in range(5): start_time = time.time() result = await func() end_time = time.time() times.append(end_time - start_time) assert isinstance(result, dict) # Verify function works # Calculate statistics avg_time = sum(times) / len(times) max_time = max(times) # Benchmark assertions assert avg_time < 3.0, f"{func_name} average too slow: {avg_time:.3f}s" assert max_time < 8.0, f"{func_name} worst case too slow: {max_time:.3f}s" if __name__ == "__main__": # Run tests manually for debugging async def run_basic_tests(): test_index = TestDatabaseIndexPerformance() print("✅ Database index performance tests defined") test_analytics = TestAnalyticsFunctionPerformance() await test_analytics.test_basic_stats_performance() print("✅ Analytics function performance tests passed") test_concurrent = TestConcurrentUserPerformance() print("✅ Concurrent user performance tests defined") asyncio.run(run_basic_tests())