Cache Testing: Testing Redis Cache Behavior Under Various Conditions
In this tutorial, you will learn about Cache Testing: Testing Redis Cache Behavior Under Various Conditions. We cover key concepts, practical examples, and best practices to help you master this topic.
Cache testing validates that Redis behaves correctly under various conditions including normal operation, memory pressure, node failures, and traffic spikes, ensuring cache hit rates meet targets and fallback mechanisms work as designed.
flowchart TD
Test[Cache Test Suite] --> Unit[Unit Tests]
Test --> Integration[Integration Tests]
Test --> Performance[Benchmarks]
Test --> Chaos[Chaos Tests]
Unit -->|Mock Redis| Behavior[Cache Logic]
Integration -->|Real Redis| Correctness[Cache Behavior]
Performance -->|Load Test| Throughput[Latency + Throughput]
Chaos -->|Failures| Resilience[Failover + Recovery]
What You'll Learn
- Unit Testing cache logic with mocked Redis
- Integration Testing with real Redis instances
- Load testing for latency and throughput benchmarks
- Chaos testing for failover and recovery validation
Why It Matters
An untested cache is a liability. A misconfigured eviction policy, incorrect TTL, or broken fallback can cause production incidents. Cache testing catches these issues before they affect users, ensuring the cache improves reliability rather than threatening it.
Real-World Use
DodaTech's cache test suite runs 200+ tests in CI on every Pull Request. It caught a regression where a new feature used a cache key prefix that conflicted with an existing key pattern, causing 30% of cache reads to return wrong data. The test failed in CI, preventing the deployment from reaching production.
Unit Testing with Mocked Redis
Test cache logic without a real Redis instance:
import unittest
from unittest.mock import Mock, patch
import json
import time
class CacheService:
def __init__(self, redis_client):
self.redis = redis_client
def get_user(self, user_id):
key = f"user:{user_id}"
data = self.redis.get(key)
if data:
return {"source": "cache", "data": json.loads(data)}
return {"source": "miss"}
def set_user(self, user_id, data, ttl=3600):
key = f"user:{user_id}"
self.redis.setex(key, ttl, json.dumps(data))
return {"cached": True}
def invalidate_user(self, user_id):
key = f"user:{user_id}"
self.redis.delete(key)
return {"invalidated": True}
class TestCacheService(unittest.TestCase):
def setUp(self):
self.mock_redis = Mock()
self.cache = CacheService(self.mock_redis)
def test_get_user_cache_hit(self):
self.mock_redis.get.return_value = json.dumps({"name": "Alice"})
result = self.cache.get_user(42)
self.assertEqual(result["source"], "cache")
self.assertEqual(result["data"]["name"], "Alice")
self.mock_redis.get.assert_called_with("user:42")
def test_get_user_cache_miss(self):
self.mock_redis.get.return_value = None
result = self.cache.get_user(99)
self.assertEqual(result["source"], "miss")
self.assertIsNone(result.get("data"))
def test_set_user(self):
data = {"name": "Bob"}
result = self.cache.set_user(7, data, ttl=3600)
self.assertTrue(result["cached"])
self.mock_redis.setex.assert_called_with(
"user:7", 3600, json.dumps(data)
)
def test_invalidate_user(self):
result = self.cache.invalidate_user(42)
self.assertTrue(result["invalidated"])
self.mock_redis.delete.assert_called_with("user:42")
suite = unittest.TestLoader().loadTestsFromTestCase(TestCacheService)
runner = unittest.TextTestRunner(verbosity=2)
runner.run(suite)
Expected output:
test_get_user_cache_hit ... ok
test_get_user_cache_miss ... ok
test_invalidate_user ... ok
test_set_user ... ok
----------------------------------------------------------------------
Ran 4 tests in 0.002s
OK
Integration Testing with Real Redis
Test against a real Redis instance in test mode:
import redis
import json
import time
import unittest
class CacheIntegrationTest(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.r = redis.Redis(decode_responses=True)
cls.r.flushdb()
@classmethod
def tearDownClass(cls):
cls.r.flushdb()
cls.r.close()
def test_set_and_get(self):
self.r.setex("test:key", 60, "test_value")
value = self.r.get("test:key")
self.assertEqual(value, "test_value")
def test_ttl(self):
self.r.setex("test:ttl", 60, "value")
ttl = self.r.ttl("test:ttl")
self.assertGreater(ttl, 0)
self.assertLessEqual(ttl, 60)
def test_missing_key(self):
value = self.r.get("nonexistent")
self.assertIsNone(value)
def test_expiration(self):
self.r.setex("test:expire", 1, "value")
time.sleep(1.5)
value = self.r.get("test:expire")
self.assertIsNone(value)
def test_cache_hit_rate(self):
for i in range(100):
self.r.setex(f"hit:test:{i}", 3600, f"value_{i}")
hits = 0
misses = 0
for i in range(200):
key = f"hit:test:{i}"
if self.r.get(key):
hits += 1
else:
misses += 1
hit_rate = hits / (hits + misses) * 100
self.assertGreaterEqual(hit_rate, 49)
suite = unittest.TestLoader().loadTestsFromTestCase(CacheIntegrationTest)
runner = unittest.TextTestRunner(verbosity=2)
runner.run(suite)
Expected output:
test_cache_hit_rate ... ok
test_expiration ... ok
test_missing_key ... ok
test_set_and_get ... ok
test_ttl ... ok
----------------------------------------------------------------------
Ran 5 tests in 1.502s
OK
Load Testing
Benchmark cache latency and throughput:
import redis
import time
import statistics
class CacheBenchmark:
def __init__(self, redis_client):
self.r = redis_client
def benchmark_get(self, key_count=10000):
"""Benchmark GET operations."""
for i in range(key_count):
self.r.setex(f"bench:get:{i}", 3600, f"value_{i}")
latencies = []
for i in range(key_count):
start = time.perf_counter()
self.r.get(f"bench:get:{i}")
latencies.append((time.perf_counter() - start) * 1000)
return self._analyze(latencies, f"GET {key_count}")
def benchmark_set(self, key_count=10000):
"""Benchmark SET operations."""
latencies = []
for i in range(key_count):
start = time.perf_counter()
self.r.setex(f"bench:set:{i}", 3600, f"value_{i}")
latencies.append((time.perf_counter() - start) * 1000)
return self._analyze(latencies, f"SET {key_count}")
def _analyze(self, latencies, label):
result = {
"operation": label,
"p50_ms": round(statistics.median(latencies), 3),
"p95_ms": round(sorted(latencies)[int(len(latencies) * 0.95)], 3),
"p99_ms": round(sorted(latencies)[int(len(latencies) * 0.99)], 3),
"max_ms": round(max(latencies), 3),
"avg_ms": round(statistics.mean(latencies), 3),
"throughput": round(len(latencies) / sum(latencies) * 1000),
}
print(f" {label:15s}: p50={result['p50_ms']}ms "
f"p95={result['p95_ms']}ms "
f"p99={result['p99_ms']}ms "
f"max={result['max_ms']}ms "
f"throughput={result['throughput']:,}/s")
return result
r = redis.Redis(decode_responses=True)
bench = CacheBenchmark(r)
print("Cache Benchmark Results:")
bench.benchmark_set(5000)
bench.benchmark_get(5000)
Expected output:
Cache Benchmark Results:
SET 5000 : p50=0.25ms p95=0.58ms p99=1.12ms max=2.34ms throughput=40,000/s
GET 5000 : p50=0.22ms p95=0.52ms p99=1.05ms max=1.89ms throughput=45,000/s
Common Mistakes
- Testing with mocked Redis but not testing against real Redis — mocked tests verify logic but not real Redis behavior. Always run integration tests against a real Redis instance in CI.
- Not testing cache expiration — TTL behavior is critical. Test that keys actually expire at the expected time and that your application handles expiry correctly.
- Testing with empty cache only — test with a warm cache (hit path), cold cache (miss path), and maximum-capacity cache (eviction path).
- Ignoring network latency in tests — running tests against localhost Redis doesn't reveal network timeout issues. Add a test that simulates network latency with tc (traffic control).
- Not testing concurrent access — concurrent cache operations can reveal race conditions. Use threads in test code to simulate concurrent access patterns.
Practice Questions
- What is the difference between unit testing with mocked Redis and integration testing with real Redis?
- Why should you test cache behavior at maximum capacity?
- How do you test cache expiration timing?
- What concurrency issues can arise with cache operations and how do you test them?
- How do you benchmark cache latency and throughput?
Challenge
Build a cache test framework that: (1) starts a local Redis instance for testing (docker-compose or embedded), (2) seeds the cache with test data at various sizes (10 keys, 1000 keys, 100000 keys), (3) runs test suites for: basic CRUD, TTL expiration, eviction policy behavior (fill cache and observe eviction), concurrent access (10 threads doing 1000 ops each), network failure simulation (stop Redis and verify fallback), and performance baselines, (4) generates a test report with pass/fail and performance metrics, and (5) can be integrated into CI/CD pipelines.
FAQ
Mini Project
Build a comprehensive cache test suite that: (1) unit tests with mocked Redis for all cache service methods, (2) integration tests with real Redis covering hit, miss, TTL, and expiration, (3) eviction tests that fill Redis to maxmemory and verify eviction policy behavior, (4) concurrency tests with 10 parallel workers doing 1000 operations each, (5) fallback tests that simulate Redis downtime and verify database fallback works, (6) a benchmark that measures p50/p95/p99 latency and throughput, and (7) a CI configuration that runs all tests on every pull request.
What's Next
Continue with Cache-Friendly API Design to learn how to design APIs that maximize cache efficiency, or explore Cache Security for securing your cache layer.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro