AHMET YILMAZ
b3fff546e9
Some checks failed
System Monitoring / Health Checks (push) Has been cancelled
System Monitoring / Performance Monitoring (push) Has been cancelled
System Monitoring / Database Monitoring (push) Has been cancelled
System Monitoring / Cache Monitoring (push) Has been cancelled
System Monitoring / Log Monitoring (push) Has been cancelled
System Monitoring / Resource Monitoring (push) Has been cancelled
System Monitoring / Uptime Monitoring (push) Has been cancelled
System Monitoring / Backup Monitoring (push) Has been cancelled
System Monitoring / Security Monitoring (push) Has been cancelled
System Monitoring / Monitoring Dashboard (push) Has been cancelled
System Monitoring / Alerting (push) Has been cancelled
Security Scanning / Dependency Scanning (push) Has been cancelled
Security Scanning / Code Security Scanning (push) Has been cancelled
Security Scanning / Secrets Scanning (push) Has been cancelled
Security Scanning / Container Security Scanning (push) Has been cancelled
Security Scanning / Compliance Checking (push) Has been cancelled
Security Scanning / Security Dashboard (push) Has been cancelled
Security Scanning / Security Remediation (push) Has been cancelled
682 lines
26 KiB
Python
682 lines
26 KiB
Python
"""
|
|
Unit tests for database optimization components.
|
|
|
|
This module tests the database optimization functionality including query optimization,
|
|
index management, configuration management, and performance monitoring specifically
|
|
designed for the multi-tenant SaaS platform with Malaysian market requirements.
|
|
"""
|
|
|
|
import unittest
|
|
from unittest.mock import Mock, patch, MagicMock
|
|
from django.test import TestCase, override_settings
|
|
from django.db import connection, models
|
|
from django.core.cache import cache
|
|
from django.utils import timezone
|
|
from django.contrib.auth import get_user_model
|
|
from django_tenants.utils import schema_context
|
|
|
|
from core.optimization.query_optimization import (
|
|
DatabaseOptimizer,
|
|
QueryOptimizer,
|
|
CacheManager,
|
|
DatabaseMaintenance,
|
|
OptimizationLevel,
|
|
QueryMetrics,
|
|
IndexRecommendation
|
|
)
|
|
from core.optimization.index_manager import (
|
|
IndexManager,
|
|
IndexType,
|
|
IndexStatus,
|
|
IndexInfo,
|
|
IndexRecommendation as IndexRec
|
|
)
|
|
from core.optimization.config import (
|
|
DatabaseConfig,
|
|
ConnectionPoolConfig,
|
|
QueryOptimizationConfig,
|
|
CacheConfig,
|
|
MultiTenantConfig,
|
|
MalaysianConfig,
|
|
PerformanceConfig,
|
|
get_config,
|
|
validate_environment_config
|
|
)
|
|
|
|
User = get_user_model()
|
|
|
|
|
|
class DatabaseOptimizerTests(TestCase):
|
|
"""Test cases for DatabaseOptimizer class."""
|
|
|
|
def setUp(self):
|
|
"""Set up test environment."""
|
|
self.optimizer = DatabaseOptimizer()
|
|
self.test_tenant = "test_tenant"
|
|
|
|
def test_init(self):
|
|
"""Test DatabaseOptimizer initialization."""
|
|
optimizer = DatabaseOptimizer(self.test_tenant)
|
|
self.assertEqual(optimizer.tenant_schema, self.test_tenant)
|
|
self.assertIsInstance(optimizer.query_history, list)
|
|
self.assertIsInstance(optimizer.optimization_stats, dict)
|
|
|
|
@patch('core.optimization.query_optimization.connection')
|
|
def test_monitor_query_context_manager(self, mock_connection):
|
|
"""Test query monitoring context manager."""
|
|
mock_cursor = Mock()
|
|
mock_connection.cursor.return_value.__enter__.return_value = mock_cursor
|
|
mock_cursor.fetchone.return_value = ('test_query', 1, 0.5, 10, 1)
|
|
|
|
with self.optimizer.monitor_query("test query"):
|
|
pass
|
|
|
|
self.assertEqual(len(self.optimizer.query_history), 1)
|
|
self.assertEqual(self.optimizer.optimization_stats['queries_analyzed'], 1)
|
|
|
|
@patch('core.optimization.query_optimization.connection')
|
|
def test_optimize_tenant_queries(self, mock_connection):
|
|
"""Test tenant query optimization."""
|
|
mock_cursor = Mock()
|
|
mock_connection.cursor.return_value.__enter__.return_value = mock_cursor
|
|
mock_cursor.fetchone.return_value = (5,)
|
|
|
|
# Create a mock model
|
|
class TestModel(models.Model):
|
|
class Meta:
|
|
app_label = 'test'
|
|
|
|
results = self.optimizer.optimize_tenant_queries(TestModel, self.test_tenant)
|
|
|
|
self.assertIn('tenant', results)
|
|
self.assertIn('queries_optimized', results)
|
|
|
|
def test_optimize_malaysian_queries(self):
|
|
"""Test Malaysian query optimization."""
|
|
with patch.object(self.optimizer, '_optimize_sst_queries', return_value=3):
|
|
with patch.object(self.optimizer, '_optimize_ic_validation', return_value=True):
|
|
with patch.object(self.optimizer, '_optimize_address_queries', return_value=2):
|
|
results = self.optimizer.optimize_malaysian_queries()
|
|
|
|
self.assertEqual(results['sst_queries_optimized'], 3)
|
|
self.assertTrue(results['ic_validation_optimized'])
|
|
self.assertEqual(results['address_queries_optimized'], 2)
|
|
self.assertIn('localization_improvements', results)
|
|
|
|
@patch('core.optimization.query_optimization.connection')
|
|
def test_analyze_query_performance(self, mock_connection):
|
|
"""Test query performance analysis."""
|
|
mock_cursor = Mock()
|
|
mock_connection.cursor.return_value.__enter__.return_value = mock_cursor
|
|
mock_cursor.fetchall.return_value = [
|
|
(100, 0.5, 2),
|
|
[('public', 'test_table', 10, 100, 5, 50)]
|
|
]
|
|
|
|
analysis = self.optimizer.analyze_query_performance(24)
|
|
|
|
self.assertEqual(analysis['total_queries'], 100)
|
|
self.assertEqual(analysis['slow_queries'], 2)
|
|
self.assertEqual(len(analysis['most_used_tables']), 1)
|
|
|
|
def test_get_optimization_report(self):
|
|
"""Test optimization report generation."""
|
|
with patch.object(self.optimizer, 'optimize_malaysian_queries', return_value={}):
|
|
with patch.object(self.optimizer, 'analyze_query_performance', return_value={}):
|
|
with patch.object(self.optimizer, '_get_suggested_actions', return_value=[]):
|
|
report = self.optimizer.get_optimization_report()
|
|
|
|
self.assertIn('optimization_statistics', report)
|
|
self.assertIn('malaysian_optimizations', report)
|
|
self.assertIn('suggested_actions', report)
|
|
|
|
def test_clear_optimization_history(self):
|
|
"""Test clearing optimization history."""
|
|
self.optimizer.query_history = [Mock()]
|
|
self.optimizer.optimization_stats['queries_analyzed'] = 5
|
|
|
|
self.optimizer.clear_optimization_history()
|
|
|
|
self.assertEqual(len(self.optimizer.query_history), 0)
|
|
self.assertEqual(self.optimizer.optimization_stats['queries_analyzed'], 0)
|
|
|
|
|
|
class QueryOptimizerTests(TestCase):
|
|
"""Test cases for QueryOptimizer static methods."""
|
|
|
|
def test_optimize_tenant_filter(self):
|
|
"""Test tenant filter optimization."""
|
|
queryset = Mock()
|
|
optimized = QueryOptimizer.optimize_tenant_filter(queryset, 1)
|
|
|
|
queryset.filter.assert_called_once_with(tenant_id=1)
|
|
queryset.select_related.assert_called_once_with('tenant')
|
|
|
|
def test_optimize_pagination(self):
|
|
"""Test pagination optimization."""
|
|
queryset = Mock()
|
|
optimized = QueryOptimizer.optimize_pagination(queryset, 25)
|
|
|
|
queryset.order_by.assert_called_once_with('id')
|
|
queryset.__getitem__.assert_called_once_with(slice(0, 25))
|
|
|
|
def test_optimize_foreign_key_query(self):
|
|
"""Test foreign key query optimization."""
|
|
queryset = Mock()
|
|
optimized = QueryOptimizer.optimize_foreign_key_query(queryset, ['user', 'profile'])
|
|
|
|
queryset.select_related.assert_called_once_with('user', 'profile')
|
|
|
|
def test_optimize_many_to_many_query(self):
|
|
"""Test many-to-many query optimization."""
|
|
queryset = Mock()
|
|
optimized = QueryOptimizer.optimize_many_to_many_query(queryset, ['tags', 'categories'])
|
|
|
|
queryset.prefetch_related.assert_called_once_with('tags', 'categories')
|
|
|
|
def test_optimize_date_range_query(self):
|
|
"""Test date range query optimization."""
|
|
queryset = Mock()
|
|
start_date = timezone.now() - timezone.timedelta(days=7)
|
|
end_date = timezone.now()
|
|
|
|
optimized = QueryOptimizer.optimize_date_range_query(
|
|
queryset, 'created_at', start_date, end_date
|
|
)
|
|
|
|
expected_filter = {
|
|
'created_at__gte': start_date,
|
|
'created_at__lte': end_date
|
|
}
|
|
queryset.filter.assert_called_once_with(**expected_filter)
|
|
queryset.order_by.assert_called_once_with('created_at')
|
|
|
|
@patch('core.optimization.query_optimization.SearchVector')
|
|
@patch('core.optimization.query_optimization.SearchQuery')
|
|
@patch('core.optimization.query_optimization.SearchRank')
|
|
def test_optimize_full_text_search(self, mock_search_rank, mock_search_query, mock_search_vector):
|
|
"""Test full-text search optimization."""
|
|
queryset = Mock()
|
|
mock_search_vector.return_value = Mock()
|
|
mock_search_query.return_value = Mock()
|
|
mock_search_rank.return_value = Mock()
|
|
|
|
optimized = QueryOptimizer.optimize_full_text_search(
|
|
queryset, ['title', 'content'], 'search term'
|
|
)
|
|
|
|
queryset.annotate.assert_called()
|
|
queryset.filter.assert_called()
|
|
queryset.order_by.assert_called()
|
|
|
|
|
|
class CacheManagerTests(TestCase):
|
|
"""Test cases for CacheManager class."""
|
|
|
|
def test_get_cache_key(self):
|
|
"""Test cache key generation."""
|
|
key = CacheManager.get_cache_key("prefix", "arg1", "arg2", 123)
|
|
self.assertEqual(key, "prefix_arg1_arg2_123")
|
|
|
|
def test_cache_query_result(self):
|
|
"""Test caching query results."""
|
|
cache_key = "test_key"
|
|
query_result = {"data": "test"}
|
|
|
|
CacheManager.cache_query_result(cache_key, query_result, 3600)
|
|
|
|
# Mock cache.get to return cached result
|
|
with patch.object(cache, 'get', return_value=query_result):
|
|
cached_result = CacheManager.get_cached_result(cache_key)
|
|
self.assertEqual(cached_result, query_result)
|
|
|
|
@patch('core.optimization.query_optimization.cache')
|
|
def test_invalidate_cache_pattern(self, mock_cache):
|
|
"""Test cache invalidation by pattern."""
|
|
mock_cache.keys.return_value = ['prefix_1', 'prefix_2', 'other_key']
|
|
|
|
CacheManager.invalidate_cache_pattern('prefix_*')
|
|
|
|
mock_cache.delete_many.assert_called_once_with(['prefix_1', 'prefix_2'])
|
|
|
|
|
|
class DatabaseMaintenanceTests(TestCase):
|
|
"""Test cases for DatabaseMaintenance class."""
|
|
|
|
@patch('core.optimization.query_optimization.connection')
|
|
def test_analyze_tables(self, mock_connection):
|
|
"""Test table analysis."""
|
|
mock_cursor = Mock()
|
|
mock_connection.cursor.return_value.__enter__.return_value = mock_cursor
|
|
mock_cursor.fetchall.return_value = [
|
|
('public', 'test_table1'),
|
|
('public', 'test_table2')
|
|
]
|
|
|
|
DatabaseMaintenance.analyze_tables()
|
|
|
|
self.assertEqual(mock_cursor.execute.call_count, 2) # SELECT + ANALYZE
|
|
mock_cursor.execute.assert_any_call("ANALYZE public.test_table1")
|
|
mock_cursor.execute.assert_any_call("ANALYZE public.test_table2")
|
|
|
|
@patch('core.optimization.query_optimization.connection')
|
|
def test_vacuum_tables(self, mock_connection):
|
|
"""Test table vacuuming."""
|
|
mock_cursor = Mock()
|
|
mock_connection.cursor.return_value.__enter__.return_value = mock_cursor
|
|
mock_cursor.fetchall.return_value = [
|
|
('public', 'test_table1'),
|
|
('public', 'test_table2')
|
|
]
|
|
|
|
DatabaseMaintenance.vacuum_tables()
|
|
|
|
self.assertEqual(mock_cursor.execute.call_count, 2) # SELECT + VACUUM
|
|
mock_cursor.execute.assert_any_call("VACUUM ANALYZE public.test_table1")
|
|
mock_cursor.execute.assert_any_call("VACUUM ANALYZE public.test_table2")
|
|
|
|
@patch('core.optimization.query_optimization.connection')
|
|
def test_get_table_sizes(self, mock_connection):
|
|
"""Test getting table sizes."""
|
|
mock_cursor = Mock()
|
|
mock_connection.cursor.return_value.__enter__.return_value = mock_cursor
|
|
mock_cursor.fetchall.return_value = [
|
|
('public', 'test_table1', '10 MB', 10485760),
|
|
('public', 'test_table2', '5 MB', 5242880)
|
|
]
|
|
|
|
sizes = DatabaseMaintenance.get_table_sizes()
|
|
|
|
self.assertEqual(len(sizes), 2)
|
|
self.assertEqual(sizes[0]['table'], 'test_table1')
|
|
self.assertEqual(sizes[0]['size'], '10 MB')
|
|
self.assertEqual(sizes[0]['size_bytes'], 10485760)
|
|
|
|
|
|
class IndexManagerTests(TestCase):
|
|
"""Test cases for IndexManager class."""
|
|
|
|
def setUp(self):
|
|
"""Set up test environment."""
|
|
self.manager = IndexManager(self.test_tenant)
|
|
|
|
@patch('core.optimization.index_manager.connection')
|
|
def test_get_all_indexes(self, mock_connection):
|
|
"""Test getting all indexes."""
|
|
mock_cursor = Mock()
|
|
mock_connection.cursor.return_value.__enter__.return_value = mock_cursor
|
|
mock_cursor.fetchall.return_value = [
|
|
('idx_test', 'test_table', 'btree', False, False, 'CREATE INDEX idx_test ON test_table (id)', 1024, 'test_tenant')
|
|
]
|
|
|
|
indexes = self.manager.get_all_indexes()
|
|
|
|
self.assertEqual(len(indexes), 1)
|
|
self.assertIsInstance(indexes[0], IndexInfo)
|
|
self.assertEqual(indexes[0].name, 'idx_test')
|
|
self.assertEqual(indexes[0].table_name, 'test_table')
|
|
|
|
def test_extract_column_names(self):
|
|
"""Test extracting column names from index definition."""
|
|
definition = "CREATE INDEX idx_test ON test_table (id, name, created_at)"
|
|
columns = self.manager._extract_column_names(definition)
|
|
|
|
self.assertEqual(columns, ['id', 'name', 'created_at'])
|
|
|
|
@patch('core.optimization.index_manager.connection')
|
|
def test_analyze_index_performance(self, mock_connection):
|
|
"""Test index performance analysis."""
|
|
mock_cursor = Mock()
|
|
mock_connection.cursor.return_value.__enter__.return_value = mock_cursor
|
|
mock_cursor.fetchall.return_value = [
|
|
('test_table1', 5000, 100000, 1024 * 1024 * 10),
|
|
('test_table2', 1000, 50000, 1024 * 1024 * 5)
|
|
]
|
|
|
|
analysis = self.manager.analyze_index_performance()
|
|
|
|
self.assertIn('total_indexes', analysis)
|
|
self.assertIn('unused_indexes', analysis)
|
|
self.assertIn('recommendations', analysis)
|
|
|
|
@patch('core.optimization.index_manager.connection')
|
|
def test_create_index(self, mock_connection):
|
|
"""Test index creation."""
|
|
mock_cursor = Mock()
|
|
mock_connection.cursor.return_value.__enter__.return_value = mock_cursor
|
|
|
|
index_name = self.manager.create_index(
|
|
table_name='test_table',
|
|
columns=['id', 'name'],
|
|
index_type=IndexType.BTREE,
|
|
unique=True
|
|
)
|
|
|
|
self.assertEqual(index_name, 'unq_test_table_id_name')
|
|
mock_cursor.execute.assert_called_once()
|
|
self.assertEqual(self.manager.stats['indexes_created'], 1)
|
|
|
|
@patch('core.optimization.index_manager.connection')
|
|
def test_drop_index(self, mock_connection):
|
|
"""Test index dropping."""
|
|
mock_cursor = Mock()
|
|
mock_connection.cursor.return_value.__enter__.return_value = mock_cursor
|
|
|
|
result = self.manager.drop_index('test_index')
|
|
|
|
self.assertTrue(result)
|
|
mock_cursor.execute.assert_called_once()
|
|
self.assertEqual(self.manager.stats['indexes_dropped'], 1)
|
|
|
|
@patch('core.optimization.index_manager.connection')
|
|
def test_rebuild_index(self, mock_connection):
|
|
"""Test index rebuilding."""
|
|
mock_cursor = Mock()
|
|
mock_connection.cursor.return_value.__enter__.return_value = mock_cursor
|
|
|
|
result = self.manager.rebuild_index('test_index')
|
|
|
|
self.assertTrue(result)
|
|
mock_cursor.execute.assert_called_once_with("REINDEX INDEX test_index")
|
|
self.assertEqual(self.manager.stats['indexes_rebuilt'], 1)
|
|
|
|
@patch('core.optimization.index_manager.connection')
|
|
def test_create_malaysian_indexes(self, mock_connection):
|
|
"""Test creating Malaysian-specific indexes."""
|
|
mock_cursor = Mock()
|
|
mock_connection.cursor.return_value.__enter__.return_value = mock_cursor
|
|
|
|
created = self.manager.create_malaysian_indexes()
|
|
|
|
self.assertIsInstance(created, list)
|
|
# Should create multiple Malaysian indexes
|
|
self.assertGreater(len(created), 0)
|
|
|
|
@patch('core.optimization.index_manager.connection')
|
|
def test_get_index_statistics(self, mock_connection):
|
|
"""Test getting index statistics."""
|
|
mock_cursor = Mock()
|
|
mock_connection.cursor.return_value.__enter__.return_value = mock_cursor
|
|
mock_cursor.fetchall.return_value = [
|
|
('btree', 5),
|
|
('hash', 2),
|
|
('active', 6),
|
|
('inactive', 1)
|
|
]
|
|
|
|
stats = self.manager.get_index_statistics()
|
|
|
|
self.assertIn('total_indexes', stats)
|
|
self.assertIn('index_types', stats)
|
|
self.assertIn('status_distribution', stats)
|
|
self.assertEqual(stats['index_types']['btree'], 5)
|
|
self.assertEqual(stats['index_types']['hash'], 2)
|
|
|
|
|
|
class DatabaseConfigTests(TestCase):
|
|
"""Test cases for DatabaseConfig class."""
|
|
|
|
def test_production_config(self):
|
|
"""Test production configuration."""
|
|
config = DatabaseConfig("production")
|
|
|
|
self.assertEqual(config.environment, "production")
|
|
self.assertIsInstance(config.connection_pool, ConnectionPoolConfig)
|
|
self.assertIsInstance(config.query_optimization, QueryOptimizationConfig)
|
|
self.assertIsInstance(config.cache, CacheConfig)
|
|
self.assertIsInstance(config.multi_tenant, MultiTenantConfig)
|
|
self.assertIsInstance(config.malaysian, MalaysianConfig)
|
|
self.assertIsInstance(config.performance, PerformanceConfig)
|
|
|
|
# Check production-specific settings
|
|
self.assertGreater(config.connection_pool.max_connections, 50)
|
|
self.assertTrue(config.performance.enable_connection_pooling)
|
|
self.assertTrue(config.performance.enable_query_optimization)
|
|
|
|
def test_staging_config(self):
|
|
"""Test staging configuration."""
|
|
config = DatabaseConfig("staging")
|
|
|
|
self.assertEqual(config.environment, "staging")
|
|
# Should be less aggressive than production
|
|
self.assertLess(config.connection_pool.max_connections, 200)
|
|
self.assertGreater(config.query_optimization.slow_query_threshold, 0.5)
|
|
|
|
def test_development_config(self):
|
|
"""Test development configuration."""
|
|
config = DatabaseConfig("development")
|
|
|
|
self.assertEqual(config.environment, "development")
|
|
# Should have minimal optimization for development
|
|
self.assertFalse(config.performance.enable_connection_pooling)
|
|
self.assertFalse(config.performance.enable_query_optimization)
|
|
|
|
def test_get_django_database_config(self):
|
|
"""Test Django database configuration generation."""
|
|
config = DatabaseConfig("production")
|
|
db_config = config.get_django_database_config()
|
|
|
|
self.assertIn('default', db_config)
|
|
self.assertIn('ENGINE', db_config['default'])
|
|
self.assertIn('OPTIONS', db_config['default'])
|
|
self.assertEqual(db_config['default']['ENGINE'], 'django_tenants.postgresql_backend')
|
|
|
|
def test_get_django_cache_config(self):
|
|
"""Test Django cache configuration generation."""
|
|
config = DatabaseConfig("production")
|
|
cache_config = config.get_django_cache_config()
|
|
|
|
self.assertIn('default', cache_config)
|
|
self.assertIn('tenant_cache', cache_config)
|
|
self.assertIn('malaysian_cache', cache_config)
|
|
|
|
def test_get_postgresql_settings(self):
|
|
"""Test PostgreSQL settings generation."""
|
|
config = DatabaseConfig("production")
|
|
settings = config.get_postgresql_settings()
|
|
|
|
self.assertIsInstance(settings, list)
|
|
self.assertGreater(len(settings), 0)
|
|
# Should contain performance-related settings
|
|
settings_str = ' '.join(settings)
|
|
self.assertIn('shared_buffers', settings_str)
|
|
self.assertIn('effective_cache_size', settings_str)
|
|
|
|
def test_validate_configuration(self):
|
|
"""Test configuration validation."""
|
|
config = DatabaseConfig("production")
|
|
warnings = config.validate_configuration()
|
|
|
|
self.assertIsInstance(warnings, list)
|
|
# Should not have warnings for valid config
|
|
# But will accept empty list as valid
|
|
|
|
def test_get_performance_recommendations(self):
|
|
"""Test performance recommendations."""
|
|
config = DatabaseConfig("production")
|
|
recommendations = config.get_performance_recommendations()
|
|
|
|
self.assertIsInstance(recommendations, list)
|
|
# Should have recommendations for production
|
|
self.assertGreater(len(recommendations), 0)
|
|
|
|
|
|
class ConfigFactoryTests(TestCase):
|
|
"""Test cases for configuration factory functions."""
|
|
|
|
def test_get_config(self):
|
|
"""Test configuration factory function."""
|
|
config = get_config("production")
|
|
self.assertIsInstance(config, DatabaseConfig)
|
|
self.assertEqual(config.environment, "production")
|
|
|
|
def test_get_production_config(self):
|
|
"""Test production configuration factory."""
|
|
config = get_production_config()
|
|
self.assertIsInstance(config, DatabaseConfig)
|
|
self.assertEqual(config.environment, "production")
|
|
|
|
def test_get_staging_config(self):
|
|
"""Test staging configuration factory."""
|
|
config = get_staging_config()
|
|
self.assertIsInstance(config, DatabaseConfig)
|
|
self.assertEqual(config.environment, "staging")
|
|
|
|
def test_get_development_config(self):
|
|
"""Test development configuration factory."""
|
|
config = get_development_config()
|
|
self.assertIsInstance(config, DatabaseConfig)
|
|
self.assertEqual(config.environment, "development")
|
|
|
|
@patch('core.optimization.config.get_config')
|
|
def test_validate_environment_config(self, mock_get_config):
|
|
"""Test environment configuration validation."""
|
|
mock_config = Mock()
|
|
mock_config.validate_configuration.return_value = []
|
|
mock_get_config.return_value = mock_config
|
|
|
|
result = validate_environment_config("production")
|
|
|
|
self.assertTrue(result)
|
|
mock_config.validate_configuration.assert_called_once()
|
|
|
|
|
|
class IntegrationTests(TestCase):
|
|
"""Integration tests for optimization components."""
|
|
|
|
@override_settings(CACHES={
|
|
'default': {
|
|
'BACKEND': 'django.core.cache.backends.dummy.DummyCache'
|
|
}
|
|
})
|
|
def test_cache_manager_integration(self):
|
|
"""Test CacheManager integration with Django cache."""
|
|
cache_key = CacheManager.get_cache_key("test", "integration")
|
|
test_data = {"key": "value"}
|
|
|
|
CacheManager.cache_query_result(cache_key, test_data)
|
|
cached_data = CacheManager.get_cached_result(cache_key)
|
|
|
|
self.assertEqual(cached_data, test_data)
|
|
|
|
@patch('core.optimization.query_optimization.connection')
|
|
def test_database_optimizer_integration(self, mock_connection):
|
|
"""Test DatabaseOptimizer integration."""
|
|
mock_cursor = Mock()
|
|
mock_connection.cursor.return_value.__enter__.return_value = mock_cursor
|
|
mock_cursor.fetchall.return_value = [
|
|
(100, 0.5, 2),
|
|
[('public', 'test_table', 10, 100, 5, 50)]
|
|
]
|
|
|
|
optimizer = DatabaseOptimizer()
|
|
analysis = optimizer.analyze_query_performance()
|
|
|
|
self.assertEqual(analysis['total_queries'], 100)
|
|
self.assertEqual(analysis['slow_queries'], 2)
|
|
|
|
def test_query_optimizer_integration(self):
|
|
"""Test QueryOptimizer integration with mock querysets."""
|
|
# This test uses mock querysets to test optimization logic
|
|
queryset = Mock()
|
|
optimized = QueryOptimizer.optimize_tenant_filter(queryset, 1)
|
|
|
|
queryset.filter.assert_called_with(tenant_id=1)
|
|
queryset.select_related.assert_called_with('tenant')
|
|
|
|
|
|
class MalaysianOptimizationTests(TestCase):
|
|
"""Test cases for Malaysian-specific optimizations."""
|
|
|
|
def setUp(self):
|
|
"""Set up test environment."""
|
|
self.optimizer = DatabaseOptimizer()
|
|
|
|
@patch('core.optimization.query_optimization.connection')
|
|
def test_malaysian_sst_optimization(self, mock_connection):
|
|
"""Test SST optimization for Malaysian market."""
|
|
mock_cursor = Mock()
|
|
mock_connection.cursor.return_value.__enter__.return_value = mock_cursor
|
|
|
|
result = self.optimizer._optimize_sst_queries()
|
|
|
|
self.assertIsInstance(result, int)
|
|
self.assertGreaterEqual(result, 0)
|
|
|
|
@patch('core.optimization.query_optimization.connection')
|
|
def test_malaysian_ic_validation_optimization(self, mock_connection):
|
|
"""Test IC validation optimization for Malaysian market."""
|
|
mock_cursor = Mock()
|
|
mock_connection.cursor.return_value.__enter__.return_value = mock_cursor
|
|
|
|
result = self.optimizer._optimize_ic_validation()
|
|
|
|
self.assertIsInstance(result, bool)
|
|
|
|
@patch('core.optimization.query_optimization.connection')
|
|
def test_malaysian_address_optimization(self, mock_connection):
|
|
"""Test address optimization for Malaysian market."""
|
|
mock_cursor = Mock()
|
|
mock_connection.cursor.return_value.__enter__.return_value = mock_cursor
|
|
|
|
result = self.optimizer._optimize_address_queries()
|
|
|
|
self.assertIsInstance(result, int)
|
|
self.assertGreaterEqual(result, 0)
|
|
|
|
def test_malaysian_config(self):
|
|
"""Test Malaysian configuration settings."""
|
|
config = DatabaseConfig("production")
|
|
|
|
self.assertEqual(config.malaysian.timezone, "Asia/Kuala_Lumpur")
|
|
self.assertEqual(config.malaysian.locale, "ms_MY")
|
|
self.assertEqual(config.malaysian.currency, "MYR")
|
|
self.assertTrue(config.malaysian.enable_local_caching)
|
|
self.assertTrue(config.malaysian.malaysian_indexes_enabled)
|
|
|
|
|
|
class PerformanceTests(TestCase):
|
|
"""Performance tests for optimization components."""
|
|
|
|
@patch('core.optimization.query_optimization.connection')
|
|
def test_query_monitoring_performance(self, mock_connection):
|
|
"""Test performance of query monitoring."""
|
|
mock_cursor = Mock()
|
|
mock_connection.cursor.return_value.__enter__.return_value = mock_cursor
|
|
mock_cursor.fetchone.return_value = ('test_query', 1, 0.1, 10, 1)
|
|
|
|
import time
|
|
start_time = time.time()
|
|
|
|
# Monitor multiple queries
|
|
for i in range(100):
|
|
with self.optimizer.monitor_query(f"test query {i}"):
|
|
pass
|
|
|
|
end_time = time.time()
|
|
execution_time = end_time - start_time
|
|
|
|
# Should be fast (less than 1 second for 100 queries)
|
|
self.assertLess(execution_time, 1.0)
|
|
self.assertEqual(len(self.optimizer.query_history), 100)
|
|
|
|
@patch('core.optimization.query_optimization.connection')
|
|
def test_cache_manager_performance(self, mock_connection):
|
|
"""Test performance of cache operations."""
|
|
import time
|
|
start_time = time.time()
|
|
|
|
# Perform multiple cache operations
|
|
for i in range(1000):
|
|
key = CacheManager.get_cache_key("perf_test", i)
|
|
CacheManager.cache_query_result(key, f"value_{i}")
|
|
|
|
end_time = time.time()
|
|
execution_time = end_time - start_time
|
|
|
|
# Should be fast (less than 1 second for 1000 operations)
|
|
self.assertLess(execution_time, 1.0)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main() |