Spaces:
Paused
Paused
| 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 | |