94 lines
3.1 KiB
Python
94 lines
3.1 KiB
Python
"""
|
|
Test script for MongoDB handler with a local MongoDB instance.
|
|
"""
|
|
|
|
import os
|
|
from datetime import datetime
|
|
from .database import MongoDBHandler, CollectionContext
|
|
|
|
# Create a custom handler class for local testing
|
|
class LocalMongoDBHandler(MongoDBHandler):
|
|
"""A MongoDB handler for local testing without authentication."""
|
|
|
|
def __init__(self):
|
|
"""Initialize with a direct MongoDB URI."""
|
|
self._initialized = False
|
|
self.uri = "mongodb://localhost:27017/test"
|
|
self.client_options = {
|
|
"maxPoolSize": 5,
|
|
"minPoolSize": 2,
|
|
"maxIdleTimeMS": 30000,
|
|
"waitQueueTimeoutMS": 2000,
|
|
"serverSelectionTimeoutMS": 5000,
|
|
}
|
|
self._initialized = True
|
|
|
|
|
|
# Create a custom handler for local testing
|
|
def create_local_handler():
|
|
"""Create a MongoDB handler for local testing."""
|
|
# Create a fresh instance with direct MongoDB URI
|
|
handler = LocalMongoDBHandler()
|
|
return handler
|
|
|
|
|
|
def test_connection_monitoring():
|
|
"""Test connection monitoring with the MongoDB handler."""
|
|
print("\nTesting connection monitoring...")
|
|
|
|
# Create a local handler
|
|
local_handler = create_local_handler()
|
|
|
|
# Add connection tracking to the handler
|
|
local_handler._open_connections = 0
|
|
|
|
# Modify the CollectionContext class to track connections
|
|
original_enter = CollectionContext.__enter__
|
|
original_exit = CollectionContext.__exit__
|
|
|
|
def tracked_enter(self):
|
|
result = original_enter(self)
|
|
self.db_handler._open_connections += 1
|
|
print(f"Connection opened. Total open: {self.db_handler._open_connections}")
|
|
return result
|
|
|
|
def tracked_exit(self, exc_type, exc_val, exc_tb):
|
|
self.db_handler._open_connections -= 1
|
|
print(f"Connection closed. Total open: {self.db_handler._open_connections}")
|
|
return original_exit(self, exc_type, exc_val, exc_tb)
|
|
|
|
# Apply the tracking methods
|
|
CollectionContext.__enter__ = tracked_enter
|
|
CollectionContext.__exit__ = tracked_exit
|
|
|
|
try:
|
|
# Test with multiple operations
|
|
for i in range(3):
|
|
print(f"\nTest iteration {i+1}:")
|
|
try:
|
|
with local_handler.collection("test_collection") as collection:
|
|
# Try a simple operation
|
|
try:
|
|
collection.find_one({})
|
|
print("Operation succeeded")
|
|
except Exception as e:
|
|
print(f"Operation failed: {e}")
|
|
except Exception as e:
|
|
print(f"Connection failed: {e}")
|
|
|
|
# Final connection count
|
|
print(f"\nFinal open connections: {local_handler._open_connections}")
|
|
if local_handler._open_connections == 0:
|
|
print("✅ All connections were properly closed")
|
|
else:
|
|
print(f"❌ {local_handler._open_connections} connections remain open")
|
|
|
|
finally:
|
|
# Restore original methods
|
|
CollectionContext.__enter__ = original_enter
|
|
CollectionContext.__exit__ = original_exit
|
|
|
|
|
|
if __name__ == "__main__":
|
|
test_connection_monitoring()
|