Spaces:
Paused
Paused
File size: 2,219 Bytes
4a2ab42 4ae946d 4a2ab42 4ae946d 4a2ab42 4ae946d 4a2ab42 4ae946d 4a2ab42 | 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 | from unittest.mock import MagicMock, patch
from core.database import DatabaseOptimizer
def test_database_optimizer_indexes():
"""Test that optimizer creates correct indexes using a mocked engine"""
# Mock the engine
mock_engine = MagicMock()
mock_conn = MagicMock()
mock_engine.connect.return_value.__enter__.return_value = mock_conn
# Initialize optimizer with mock engine
optimizer = DatabaseOptimizer(mock_engine)
# Get indexes to be created
indexes = optimizer.create_performance_indexes()
# Verify index structure
assert len(indexes) > 0
assert any("idx_cases_status" in idx for idx in indexes)
assert any("idx_audit_user" in idx for idx in indexes)
def test_connection_pooling_optimization():
"""Test connection pooling configuration"""
mock_engine = MagicMock()
optimizer = DatabaseOptimizer(mock_engine)
config = optimizer.optimize_connection_pooling()
assert config["pool_size"] == 20
assert config["max_overflow"] == 30
assert config["pool_recycle"] == 1800
def test_query_monitoring():
"""Test that query monitoring event listeners are attached"""
# Mock sqlalchemy.event (patched where it is imported)
with patch("core.database.event") as mock_event:
mock_engine = MagicMock()
optimizer = DatabaseOptimizer(mock_engine)
stats = optimizer.enable_query_monitoring()
# Verify monitoring was initialized
assert stats["query_count"] == 0
assert "slow_queries" in stats
# Verify event listeners were registered
assert mock_event.listens_for.call_count >= 2
# Verify calls to listens_for(engine, 'before_execute') and ('after_execute')
# call_args_list returns Call objects, args property is a tuple
calls = [c.args for c in mock_event.listens_for.call_args_list]
# We check if any call args contains 'before_execute'
found_before = False
found_after = False
for args in calls:
if "before_execute" in args:
found_before = True
if "after_execute" in args:
found_after = True
assert found_before
assert found_after
|