Spaces:
Sleeping
Sleeping
| """ | |
| Integration tests for database operations | |
| Consolidated from: | |
| - test_mongo_connection.py | |
| - test_mongodb_connection.py | |
| - Database-related portions of other test files | |
| """ | |
| import asyncio | |
| from datetime import datetime, timedelta | |
| import time | |
| import pytest | |
| from analytics.collectors import create_session, track_message, track_search | |
| from analytics.database import connect_to_database, get_database, get_messages_collection, get_search_analytics_collection, get_sessions_collection, test_connection | |
| from tests.utilities import MockDataGenerator, TestHelpers, skip_if_no_database | |
| # Create some test data | |
| user_id = "perf_test_user" | |
| session = await create_session(user_id=user_id) | |
| for i in range(5): | |
| await track_message( | |
| session_id=session.session_id, | |
| prompt_length=50, | |
| response_length=100, | |
| response_time_ms=1000, | |
| user_id=user_id | |
| ) | |
| # Wait for data to be persisted | |
| await TestHelpers.wait_for_data_persistence() | |
| # Test query performance | |
| sessions_collection = await get_sessions_collection() | |
| messages_collection = await get_messages_collection() | |
| # Time session query | |
| start_time = time.time() | |
| session_count = await sessions_collection.count_documents({"user_id": user_id}) | |
| session_query_time = time.time() - start_time | |
| # Time message query | |
| start_time = time.time() | |
| message_count = await messages_collection.count_documents({"user_id": user_id}) | |
| message_query_time = time.time() - start_time | |
| # Verify results | |
| assert session_count >= 1 | |
| assert message_count >= 5 | |
| # Performance assertions (should be fast) | |
| assert session_query_time < 2.0, f"Session query too slow: {session_query_time:.2f}s" | |
| assert message_query_time < 2.0, f"Message query too slow: {message_query_time:.2f}s" | |
| class TestDatabaseErrorHandling: | |
| """Test database error handling""" | |
| async def test_database_unavailable_handling(self): | |
| """Test handling when database is unavailable""" | |
| # This test verifies that the system gracefully handles database unavailability | |
| # The actual behavior depends on the implementation (JSON fallback, etc.) | |
| # Try to get collections when database might not be available | |
| sessions_collection = await get_sessions_collection() | |
| messages_collection = await get_messages_collection() | |
| search_collection = await get_search_analytics_collection() | |
| # Should return None or valid collection objects | |
| assert sessions_collection is None or hasattr(sessions_collection, 'count_documents') | |
| assert messages_collection is None or hasattr(messages_collection, 'count_documents') | |
| assert search_collection is None or hasattr(search_collection, 'count_documents') | |
| async def test_invalid_query_handling(self): | |
| """Test handling of invalid queries""" | |
| sessions_collection = await get_sessions_collection() | |
| # Test with invalid query structure | |
| try: | |
| # This should either work or raise a proper exception | |
| result = await sessions_collection.count_documents({"$invalid": "query"}) | |
| # If it works, result should be a number | |
| assert isinstance(result, int) | |
| except Exception as e: | |
| # If it fails, should be a proper database exception | |
| assert isinstance(e, Exception) | |
| if __name__ == "__main__": | |
| # Run tests manually for debugging | |
| async def run_basic_tests(): | |
| test_connection = TestDatabaseConnection() | |
| await test_connection.test_database_connection() | |
| print("✅ Database connection tests passed") | |
| test_collections = TestDatabaseCollections() | |
| await test_collections.test_get_sessions_collection() | |
| print("✅ Database collections tests passed") | |
| asyncio.run(run_basic_tests()) |