Spaces:
Sleeping
Sleeping
File size: 4,173 Bytes
f0b765c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 | """
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"""
@pytest.mark.asyncio
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')
@pytest.mark.asyncio
@skip_if_no_database()
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()) |