""" Unit tests for caching strategies and managers. """ import json import time from datetime import datetime, timedelta from unittest.mock import Mock, patch, MagicMock from django.test import TestCase, override_settings from django.core.cache import cache from django.conf import settings from django.contrib.auth import get_user_model from django.db import connection from django.http import HttpRequest, HttpResponse from django.test import RequestFactory from rest_framework.test import APITestCase from core.caching.cache_manager import ( CacheManager, CacheKeyGenerator, MalaysianDataCache, QueryCache, TenantCacheManager, CacheWarmer ) from core.caching.strategies import ( WriteThroughCache, WriteBehindCache, ReadThroughCache, RefreshAheadCache, CacheAsidePattern, MultiLevelCache, MalaysianCacheStrategies, CacheEvictionPolicy, cache_view_response, cache_query_results ) from core.caching.django_integration import ( TenantCacheMiddleware, CacheMiddleware, DatabaseCacheMiddleware, MalaysianCacheMiddleware, get_cache_config ) from core.caching.config import CacheConfig User = get_user_model() class CacheKeyGeneratorTest(TestCase): """Test cache key generation.""" def setUp(self): self.generator = CacheKeyGenerator() def test_generate_basic_key(self): """Test basic key generation.""" key = self.generator.generate_key("test", "123") self.assertIn("my_sme", key) self.assertIn("test", key) self.assertIn("123", key) def test_generate_key_with_context(self): """Test key generation with context.""" context = {"filter": "active", "sort": "name"} key = self.generator.generate_key("test", "123", context=context) self.assertIn("my_sme", key) self.assertIn("test", key) self.assertIn("123", key) def test_generate_malaysian_key(self): """Test Malaysian-specific key generation.""" key = self.generator.generate_malaysian_key("ic", "1234567890") self.assertIn("my_sme", key) self.assertIn("ic_1234567890", key) self.assertIn("my", key) def test_tenant_prefix_inclusion(self): """Test tenant prefix inclusion in keys.""" key = self.generator.generate_key("test", "123") self.assertIn("tenant_", key) class CacheManagerTest(TestCase): """Test cache manager operations.""" def setUp(self): self.manager = CacheManager() def test_set_and_get(self): """Test basic set and get operations.""" key = "test_key" value = {"data": "test_value"} result = self.manager.set(key, value) self.assertTrue(result) retrieved = self.manager.get(key) self.assertEqual(retrieved, value) def test_get_default_value(self): """Test get with default value.""" key = "nonexistent_key" default = {"default": "value"} result = self.manager.get(key, default) self.assertEqual(result, default) def test_delete_key(self): """Test key deletion.""" key = "test_key" value = "test_value" self.manager.set(key, value) result = self.manager.delete(key) self.assertTrue(result) retrieved = self.manager.get(key) self.assertIsNone(retrieved) def test_clear_tenant_cache(self): """Test tenant cache clearing.""" result = self.manager.clear_tenant_cache() self.assertTrue(result) def test_get_cache_stats(self): """Test cache statistics.""" stats = self.manager.get_cache_stats() self.assertIn("tenant", stats) self.assertIn("redis_available", stats) self.assertIn("default_timeout", stats) @patch('core.caching.cache_manager.get_redis_connection') def test_redis_connection_failure(self, mock_get_redis): """Test graceful handling of Redis connection failure.""" mock_get_redis.side_effect = Exception("Connection failed") manager = CacheManager() self.assertIsNone(manager.redis_client) stats = manager.get_cache_stats() self.assertFalse(stats["redis_available"]) class MalaysianDataCacheTest(TestCase): """Test Malaysian data caching.""" def setUp(self): self.cache_manager = CacheManager() self.malaysian_cache = MalaysianDataCache(self.cache_manager) def test_ic_validation_caching(self): """Test IC validation caching.""" ic_number = "1234567890" validation_result = {"valid": True, "age": 30} result = self.malaysian_cache.set_cached_ic_validation(ic_number, validation_result) self.assertTrue(result) retrieved = self.malaysian_cache.get_cached_ic_validation(ic_number) self.assertEqual(retrieved, validation_result) def test_sst_rate_caching(self): """Test SST rate caching.""" state = "Johor" category = "standard" rate = 0.06 result = self.malaysian_cache.set_cached_sst_rate(state, category, rate) self.assertTrue(result) retrieved = self.malaysian_cache.get_cached_sst_rate(state, category) self.assertEqual(retrieved, rate) def test_postcode_data_caching(self): """Test postcode data caching.""" postcode = "50000" postcode_data = {"city": "Kuala Lumpur", "state": "WP Kuala Lumpur"} result = self.malaysian_cache.set_cached_postcode_data(postcode, postcode_data) self.assertTrue(result) retrieved = self.malaysian_cache.get_cached_postcode_data(postcode) self.assertEqual(retrieved, postcode_data) class QueryCacheTest(TestCase): """Test query caching.""" def setUp(self): self.cache_manager = CacheManager() self.query_cache = QueryCache(self.cache_manager) def test_query_hash_generation(self): """Test query hash generation.""" query = "SELECT * FROM users WHERE id = %s" params = (1,) hash1 = self.query_cache.generate_query_hash(query, params) hash2 = self.query_cache.generate_query_hash(query, params) self.assertEqual(hash1, hash2) # Different params should produce different hash hash3 = self.query_cache.generate_query_hash(query, (2,)) self.assertNotEqual(hash1, hash3) def test_query_result_caching(self): """Test query result caching.""" query = "SELECT * FROM test_table" result = [{"id": 1, "name": "test"}] success = self.query_cache.cache_query_result(query, result) self.assertTrue(success) retrieved = self.query_cache.get_cached_query_result(query) self.assertEqual(retrieved, result) def test_model_cache_invalidation(self): """Test model cache invalidation.""" # Add some query hashes self.query_cache.query_hashes.add("user_query_123") self.query_cache.query_hashes.add("product_query_456") invalidated = self.query_cache.invalidate_model_cache("user") self.assertEqual(invalidated, 1) self.assertIn("product_query_456", self.query_cache.query_hashes) self.assertNotIn("user_query_123", self.query_cache.query_hashes) class TenantCacheManagerTest(TestCase): """Test tenant cache management.""" def setUp(self): self.tenant_manager = TenantCacheManager() def test_get_cache_manager(self): """Test getting cache manager for tenant.""" manager = self.tenant_manager.get_cache_manager(1) self.assertIsInstance(manager, CacheManager) self.assertEqual(manager.config.tenant_prefix, "tenant_1") def test_cache_manager_reuse(self): """Test cache manager reuse for same tenant.""" manager1 = self.tenant_manager.get_cache_manager(1) manager2 = self.tenant_manager.get_cache_manager(1) self.assertIs(manager1, manager2) def test_get_tenant_cache_stats(self): """Test tenant cache statistics.""" self.tenant_manager.get_cache_manager(1) stats = self.tenant_manager.get_tenant_cache_stats() self.assertIn("tenants", stats) self.assertIn("total_tenants", stats) self.assertEqual(stats["total_tenants"], 1) class CacheWarmerTest(TestCase): """Test cache warming.""" def setUp(self): self.cache_manager = CacheManager() self.warmer = CacheWarmer(self.cache_manager) def test_warm_malaysian_data(self): """Test warming Malaysian data.""" result = self.warmer.warm_malaysian_data() self.assertIn("sst_rates", result) self.assertIn("postcodes", result) self.assertGreater(result["sst_rates"], 0) self.assertGreater(result["postcodes"], 0) def test_warm_user_data(self): """Test warming user data.""" user = User.objects.create_user( username="testuser", email="test@example.com", password="testpass123" ) warmed = self.warmer.warm_user_data([user.id]) self.assertEqual(warmed, 1) # Verify user data is cached key = self.cache_manager.key_generator.generate_key("user", str(user.id)) cached_data = self.cache_manager.get(key) self.assertIsNotNone(cached_data) self.assertEqual(cached_data["id"], user.id) class WriteThroughCacheTest(TestCase): """Test write-through caching.""" def setUp(self): self.cache_manager = CacheManager() self.write_through = WriteThroughCache(self.cache_manager) def test_write_through_operation(self): """Test write-through operation.""" key = "test_key" value = "test_value" def db_operation(): return value result = self.write_through.write_through(key, value, db_operation) self.assertEqual(result, value) # Verify cache is populated cached_value = self.cache_manager.get(key) self.assertEqual(cached_value, value) class ReadThroughCacheTest(TestCase): """Test read-through caching.""" def setUp(self): self.cache_manager = CacheManager() self.read_through = ReadThroughCache(self.cache_manager) def test_read_through_operation(self): """Test read-through operation.""" key = "test_key" value = "test_value" def db_operation(): return value # First read - should hit database and cache result1 = self.read_through.read_through(key, db_operation) self.assertEqual(result1, value) # Second read - should hit cache result2 = self.read_through.read_through(key, db_operation) self.assertEqual(result2, value) # Verify cache was populated cached_value = self.cache_manager.get(key) self.assertEqual(cached_value, value) class CacheAsidePatternTest(TestCase): """Test cache-aside pattern.""" def setUp(self): self.cache_manager = CacheManager() self.cache_aside = CacheAsidePattern(self.cache_manager) def test_get_or_set_operation(self): """Test get-or-set operation.""" key = "test_key" value = "test_value" def db_operation(): return value # First call - should set cache result1 = self.cache_aside.get_or_set(key, db_operation) self.assertEqual(result1, value) # Second call - should get from cache result2 = self.cache_aside.get_or_set(key, db_operation) self.assertEqual(result2, value) def test_invalidate_operation(self): """Test cache invalidation.""" key = "test_key" value = "test_value" def db_operation(): return value # Set cache self.cache_aside.get_or_set(key, db_operation) # Invalidate result = self.cache_aside.invalidate(key) self.assertTrue(result) # Verify cache is cleared cached_value = self.cache_manager.get(key) self.assertIsNone(cached_value) class MultiLevelCacheTest(TestCase): """Test multi-level caching.""" def setUp(self): self.l1_cache = CacheManager() self.l2_cache = CacheManager() self.multi_cache = MultiLevelCache(self.l1_cache, self.l2_cache) def test_multi_level_get_set(self): """Test multi-level get and set operations.""" key = "test_key" value = "test_value" # Set value result = self.multi_cache.set(key, value) self.assertTrue(result) # Get from multi-level cache retrieved = self.multi_cache.get(key) self.assertEqual(retrieved, value) def test_l1_promotion(self): """Test L1 cache promotion.""" key = "test_key" value = "test_value" # Set only in L2 cache self.l2_cache.set(key, value) # Get from multi-level cache - should promote to L1 retrieved = self.multi_cache.get(key) self.assertEqual(retrieved, value) # Verify it's now in L1 cache l1_value = self.l1_cache.get(key) self.assertEqual(l1_value, value) def test_cache_statistics(self): """Test cache statistics.""" key = "test_key" value = "test_value" # Initial stats stats = self.multi_cache.get_stats() self.assertEqual(stats["l1_hits"], 0) self.assertEqual(stats["l2_hits"], 0) self.assertEqual(stats["misses"], 0) # Set and get self.multi_cache.set(key, value) self.multi_cache.get(key) # L1 hit stats = self.multi_cache.get_stats() self.assertEqual(stats["l1_hits"], 1) self.assertEqual(stats["misses"], 0) class MalaysianCacheStrategiesTest(TestCase): """Test Malaysian cache strategies.""" def setUp(self): self.cache_manager = CacheManager() self.malaysian_strategies = MalaysianCacheStrategies(self.cache_manager) def test_ic_validation_caching(self): """Test IC validation caching.""" ic_number = "1234567890" def validation_func(ic): return {"valid": True, "age": 30} result = self.malaysian_strategies.cache_ic_validation(ic_number, validation_func) self.assertEqual(result["valid"], True) # Verify cached cached = self.cache_manager.get(f"*:my:ic_validation_{ic_number}") self.assertIsNotNone(cached) def test_sst_calculation_caching(self): """Test SST calculation caching.""" calculation_key = "johor_standard" def calculation_func(): return 0.06 result = self.malaysian_strategies.cache_sst_calculation(calculation_key, calculation_func) self.assertEqual(result, 0.06) def test_postcode_lookup_caching(self): """Test postcode lookup caching.""" postcode = "50000" def lookup_func(pc): return {"city": "Kuala Lumpur", "state": "WP Kuala Lumpur"} result = self.malaysian_strategies.cache_postcode_lookup(postcode, lookup_func) self.assertEqual(result["city"], "Kuala Lumpur") class CacheEvictionPolicyTest(TestCase): """Test cache eviction policies.""" def setUp(self): self.cache_manager = CacheManager() self.eviction_policy = CacheEvictionPolicy(self.cache_manager) def test_lru_eviction(self): """Test LRU eviction.""" keys = ["key1", "key2", "key3"] # Record access with different times self.eviction_policy.record_access("key1") time.sleep(0.1) self.eviction_policy.record_access("key2") time.sleep(0.1) self.eviction_policy.record_access("key3") # LRU should evict key1 (oldest access) evicted = self.eviction_policy.lru_eviction(keys, 1) self.assertEqual(evicted, ["key1"]) def test_lfu_eviction(self): """Test LFU eviction.""" keys = ["key1", "key2", "key3"] # Record different access frequencies self.eviction_policy.record_access("key1") self.eviction_policy.record_access("key2") self.eviction_policy.record_access("key2") # Access twice self.eviction_policy.record_access("key3") self.eviction_policy.record_access("key3") self.eviction_policy.record_access("key3") # Access three times # LFU should evict key1 (least frequent) evicted = self.eviction_policy.lfu_eviction(keys, 1) self.assertEqual(evicted, ["key1"]) def test_fifo_eviction(self): """Test FIFO eviction.""" keys = ["key1", "key2", "key3"] evicted = self.eviction_policy.fifo_eviction(keys, 1) self.assertEqual(evicted, ["key1"]) class CacheMiddlewareTest(TestCase): """Test cache middleware.""" def setUp(self): self.factory = RequestFactory() self.middleware = CacheMiddleware(self.get_response) def get_response(self, request): return HttpResponse("test response") def test_middleware_process_request_cacheable(self): """Test middleware process request for cacheable path.""" request = self.factory.get('/api/products/') request.user = Mock() request.user.is_authenticated = False response = self.middleware.process_request(request) self.assertIsNone(response) # Should not return cached response def test_middleware_process_request_non_cacheable(self): """Test middleware process request for non-cacheable path.""" request = self.factory.get('/api/auth/login/') request.user = Mock() request.user.is_authenticated = False response = self.middleware.process_request(request) self.assertIsNone(response) # Should bypass cache def test_middleware_should_bypass_cache(self): """Test cache bypass logic.""" request = self.factory.get('/api/products/') request.user = Mock() request.user.is_authenticated = True should_bypass = self.middleware._should_bypass_cache(request) self.assertTrue(should_bypass) # Should bypass for authenticated users def test_cache_key_generation(self): """Test cache key generation.""" request = self.factory.get('/api/products/', {'category': 'electronics'}) request.user = Mock() request.user.is_authenticated = False request.tenant = Mock() request.tenant.id = 1 key = self.middleware._generate_cache_key(request) self.assertIn('/api/products/', key) self.assertIn('tenant_1', key) class CacheConfigurationTest(TestCase): """Test cache configuration.""" def test_cache_config_initialization(self): """Test cache configuration initialization.""" config = CacheConfig() self.assertIsInstance(config.default_timeout, int) self.assertIsInstance(config.use_redis, bool) self.assertIsInstance(config.tenant_isolation, bool) def test_get_cache_config(self): """Test getting cache configuration.""" config = get_cache_config() self.assertIn('CACHES', config) self.assertIn('CACHE_MIDDLEWARE_ALIAS', config) self.assertIn('CACHE_MIDDLEWARE_SECONDS', config) class CacheManagementCommandTest(TestCase): """Test cache management command.""" @patch('core.management.commands.cache_management.Command._output_results') def test_command_initialization(self, mock_output): """Test command initialization.""" from core.management.commands.cache_management import Command command = Command() self.assertIsNotNone(command.cache_manager) self.assertIsNotNone(command.malaysian_cache) self.assertIsNotNone(command.query_cache) @patch('core.management.commands.cache_management.Command._output_results') def test_stats_action(self, mock_output): """Test stats action.""" from core.management.commands.cache_management import Command command = Command() command.action = 'stats' command.cache_type = 'all' command.output_format = 'table' command.handle_stats() # Verify _output_results was called mock_output.assert_called_once() @patch('core.management.commands.cache_management.Command._output_results') def test_health_check_action(self, mock_output): """Test health check action.""" from core.management.commands.cache_management import Command command = Command() command.action = 'health-check' command.output_format = 'table' command.handle_health_check() # Verify _output_results was called mock_output.assert_called_once() class CacheIntegrationTest(TestCase): """Integration tests for caching system.""" def test_full_cache_workflow(self): """Test complete cache workflow.""" # Create cache manager cache_manager = CacheManager() # Test Malaysian data caching malaysian_cache = MalaysianDataCache(cache_manager) # Cache IC validation ic_result = {"valid": True, "age": 25} malaysian_cache.set_cached_ic_validation("1234567890", ic_result) # Retrieve cached result cached_result = malaysian_cache.get_cached_ic_validation("1234567890") self.assertEqual(cached_result, ic_result) # Test query caching query_cache = QueryCache(cache_manager) query = "SELECT * FROM users WHERE id = %s" result = [{"id": 1, "name": "test"}] query_cache.cache_query_result(query, result) cached_query_result = query_cache.get_cached_query_result(query) self.assertEqual(cached_query_result, result) # Test tenant isolation tenant_manager = TenantCacheManager() tenant1_cache = tenant_manager.get_cache_manager(1) tenant2_cache = tenant_manager.get_cache_manager(2) # Different tenants should have different cache managers self.assertIsNot(tenant1_cache, tenant2_cache) # Test cache warming cache_warmer = CacheWarmer(cache_manager) warmed = cache_warmer.warm_malaysian_data() self.assertGreater(warmed["sst_rates"], 0) def test_cache_error_handling(self): """Test cache error handling.""" cache_manager = CacheManager() # Test get with non-existent key result = cache_manager.get("nonexistent_key") self.assertIsNone(result) # Test get with default value result = cache_manager.get("nonexistent_key", "default") self.assertEqual(result, "default") # Test error handling in operations with patch.object(cache_manager, 'set', side_effect=Exception("Cache error")): result = cache_manager.set("test_key", "test_value") self.assertFalse(result)