Redis

Cache Invalidation Testing Strategies: Patterns and Anti-Patterns

HelpMeTest

HelpMeTest

— 8 min read
Cache Invalidation Testing Strategies: Patterns and Anti-Patterns

Phil Karlton famously said there are only two hard things in computer science: cache invalidation and naming things. Testing cache invalidation is hard for different reasons—the behavior is time-dependent, state-dependent, and often involves distributed components that are difficult to simulate in isolation.

This guide covers the major cache invalidation patterns and how to test each one reliably.

Why Cache Invalidation Tests Fail

Most cache bugs aren't "cache miss when there should be a hit." They're:

  1. Stale data served — the cache wasn't invalidated when the underlying data changed
  2. Thundering herd — invalidation triggers simultaneous cache rebuilds from multiple consumers
  3. Partial invalidation — some cache keys were updated, others weren't (related data inconsistency)
  4. Race conditions — a write happens between a cache read and a subsequent write-back

Testing these requires controlling time (TTL), controlling order of operations (races), and observing what data was served—not just whether a function was called.

Pattern 1: TTL-Based Expiry

The simplest invalidation strategy: set a TTL, let data expire naturally.

Implementation

# user_service.py
import json
import time
import redis

r = redis.Redis.from_url('redis://localhost:6379', decode_responses=True)

CACHE_TTL = 300  # 5 minutes

def get_user(user_id: int) -> dict:
    key = f'user:{user_id}'
    cached = r.get(key)
    if cached:
        return json.loads(cached)
    
    user = fetch_user_from_db(user_id)  # DB call
    r.setex(key, CACHE_TTL, json.dumps(user))
    return user

def update_user(user_id: int, data: dict) -> dict:
    user = update_user_in_db(user_id, data)  # DB write
    r.delete(f'user:{user_id}')  # Explicit invalidation
    return user

Testing TTL Expiry

Testing real TTL expiry means either waiting (slow) or using fakeredis2 with time control:

# test_user_service_ttl.py
import fakeredis
import pytest
import time
from unittest.mock import patch, MagicMock
from freezegun import freeze_time
import user_service

@pytest.fixture
def fake_redis():
    server = fakeredis.FakeServer()
    client = fakeredis.FakeRedis(server=server, decode_responses=True)
    return client

def test_cache_expires_after_ttl(fake_redis, monkeypatch):
    monkeypatch.setattr(user_service, 'r', fake_redis)
    
    db_call = MagicMock(return_value={'id': 1, 'name': 'Alice'})
    monkeypatch.setattr(user_service, 'fetch_user_from_db', db_call)
    
    # First call: cache miss, DB hit
    user_service.get_user(1)
    assert db_call.call_count == 1
    
    # Within TTL: cache hit, no DB call
    user_service.get_user(1)
    assert db_call.call_count == 1
    
    # Simulate TTL expiry by deleting key directly
    fake_redis.delete('user:1')
    
    # After expiry: cache miss again
    user_service.get_user(1)
    assert db_call.call_count == 2

def test_explicit_invalidation_on_update(fake_redis, monkeypatch):
    monkeypatch.setattr(user_service, 'r', fake_redis)
    
    db_fetch = MagicMock(side_effect=[
        {'id': 1, 'name': 'Alice'},   # first fetch
        {'id': 1, 'name': 'Alice V2'} # after update
    ])
    db_update = MagicMock(return_value={'id': 1, 'name': 'Alice V2'})
    
    monkeypatch.setattr(user_service, 'fetch_user_from_db', db_fetch)
    monkeypatch.setattr(user_service, 'update_user_in_db', db_update)
    
    # Warm the cache
    user1 = user_service.get_user(1)
    assert user1['name'] == 'Alice'
    
    # Update should invalidate
    user_service.update_user(1, {'name': 'Alice V2'})
    
    # Next read should be fresh
    user2 = user_service.get_user(1)
    assert user2['name'] == 'Alice V2'
    assert db_fetch.call_count == 2  # Two DB fetches

Testing with Real TTL via Testcontainers

For tests that must verify actual TTL behavior (e.g., checking that a 1-second TTL actually expires):

# test_real_ttl.py
import pytest
import redis
import time
from testcontainers.redis import RedisContainer

@pytest.fixture(scope='module')
def real_redis():
    with RedisContainer('redis:7-alpine') as container:
        client = redis.Redis(
            host=container.get_container_host_ip(),
            port=container.get_exposed_port(6379),
            decode_responses=True
        )
        yield client

def test_actual_ttl_expiry(real_redis):
    real_redis.setex('temp_key', 1, 'value')  # 1-second TTL
    
    assert real_redis.get('temp_key') == 'value'
    
    time.sleep(1.1)  # Wait for expiry
    
    assert real_redis.get('temp_key') is None

def test_ttl_refreshed_on_access(real_redis):
    # Some implementations refresh TTL on read (sliding expiry)
    real_redis.setex('sliding_key', 2, 'value')
    
    time.sleep(1)
    real_redis.expire('sliding_key', 2)  # Reset TTL
    time.sleep(1)
    
    # Should still exist (reset 1s before expiry)
    assert real_redis.get('sliding_key') == 'value'

Pattern 2: Event-Driven Invalidation

Update the cache when events occur. The source of truth changes → invalidate the corresponding cache entries.

Implementation

// event-driven-cache.js
const EventEmitter = require('events');
const redis = require('redis');

const events = new EventEmitter();
const client = redis.createClient();

// Cache invalidation subscriptions
events.on('user.updated', async ({ userId }) => {
  await client.del(`user:${userId}`);
  await client.del(`user:${userId}:profile`);
  await client.del(`user:${userId}:permissions`);
});

events.on('order.placed', async ({ userId, orderId }) => {
  await client.del(`user:${userId}:orders`);
  await client.del(`order:${orderId}`);
  // Invalidate aggregate cache
  await client.del('stats:daily:orders');
});

module.exports = { events, client };

Testing Event-Driven Invalidation

// event-driven-cache.test.js
const { events, client } = require('./event-driven-cache');
const redisMock = require('redis-mock');

jest.mock('redis', () => redisMock);

describe('Cache invalidation on events', () => {
  beforeEach(async () => {
    await client.flushAll();
  });

  test('user.updated invalidates all user cache keys', async () => {
    // Seed cache
    await client.set('user:42', JSON.stringify({ id: 42, name: 'Bob' }));
    await client.set('user:42:profile', JSON.stringify({ bio: '...' }));
    await client.set('user:42:permissions', JSON.stringify(['read']));
    
    // Verify seeds
    expect(await client.get('user:42')).not.toBeNull();
    expect(await client.get('user:42:profile')).not.toBeNull();
    
    // Emit event
    events.emit('user.updated', { userId: 42 });
    
    // Allow event handlers to complete
    await new Promise(resolve => setImmediate(resolve));
    
    // All keys invalidated
    expect(await client.get('user:42')).toBeNull();
    expect(await client.get('user:42:profile')).toBeNull();
    expect(await client.get('user:42:permissions')).toBeNull();
  });

  test('user.updated does not affect other user cache', async () => {
    await client.set('user:1', JSON.stringify({ id: 1 }));
    await client.set('user:2', JSON.stringify({ id: 2 }));
    
    events.emit('user.updated', { userId: 1 });
    await new Promise(resolve => setImmediate(resolve));
    
    expect(await client.get('user:1')).toBeNull();
    expect(await client.get('user:2')).not.toBeNull(); // Unaffected
  });

  test('order.placed invalidates order and daily stats', async () => {
    await client.set('user:5:orders', JSON.stringify([]));
    await client.set('stats:daily:orders', '42');
    
    events.emit('order.placed', { userId: 5, orderId: 'ord-123' });
    await new Promise(resolve => setImmediate(resolve));
    
    expect(await client.get('user:5:orders')).toBeNull();
    expect(await client.get('stats:daily:orders')).toBeNull();
  });
});

Pattern 3: Write-Through Cache

Every write updates both the database and the cache atomically. No explicit invalidation needed—the cache always has fresh data.

Implementation

# write_through_cache.py
import json
import redis
import psycopg2

r = redis.Redis.from_url('redis://localhost')

def update_product_price(product_id: int, new_price: float) -> dict:
    # Update DB first
    with get_db_connection() as conn:
        with conn.cursor() as cur:
            cur.execute(
                'UPDATE products SET price = %s WHERE id = %s RETURNING *',
                (new_price, product_id)
            )
            product = dict(zip([d[0] for d in cur.description], cur.fetchone()))
            conn.commit()
    
    # Write through to cache
    r.setex(
        f'product:{product_id}',
        3600,
        json.dumps(product)
    )
    
    return product

def get_product(product_id: int) -> dict | None:
    cached = r.get(f'product:{product_id}')
    if cached:
        return json.loads(cached)
    
    # Cache miss — load from DB
    with get_db_connection() as conn:
        with conn.cursor() as cur:
            cur.execute('SELECT * FROM products WHERE id = %s', (product_id,))
            row = cur.fetchone()
            if not row:
                return None
            product = dict(zip([d[0] for d in cur.description], row))
    
    r.setex(f'product:{product_id}', 3600, json.dumps(product))
    return product

Testing Write-Through Consistency

# test_write_through.py
import fakeredis
import pytest
from unittest.mock import patch, MagicMock, call
import write_through_cache

@pytest.fixture
def setup(monkeypatch):
    fake_r = fakeredis.FakeRedis(decode_responses=True)
    monkeypatch.setattr(write_through_cache, 'r', fake_r)
    return fake_r

def test_update_writes_to_cache(setup, monkeypatch):
    fake_r = setup
    mock_product = {'id': 1, 'name': 'Widget', 'price': 9.99}
    
    mock_conn = MagicMock()
    mock_cur = mock_conn.__enter__().cursor().__enter__()
    mock_cur.description = [('id',), ('name',), ('price',)]
    mock_cur.fetchone.return_value = (1, 'Widget', 9.99)
    
    monkeypatch.setattr(write_through_cache, 'get_db_connection', lambda: mock_conn)
    
    result = write_through_cache.update_product_price(1, 9.99)
    
    # Verify cache was written
    cached = fake_r.get('product:1')
    assert cached is not None
    import json
    assert json.loads(cached)['price'] == 9.99

def test_get_uses_cache_after_update(setup, monkeypatch):
    fake_r = setup
    
    mock_conn = MagicMock()
    mock_cur = mock_conn.__enter__().cursor().__enter__()
    mock_cur.description = [('id',), ('name',), ('price',)]
    mock_cur.fetchone.return_value = (1, 'Widget', 19.99)
    
    monkeypatch.setattr(write_through_cache, 'get_db_connection', lambda: mock_conn)
    
    # Update writes to cache
    write_through_cache.update_product_price(1, 19.99)
    
    # Reset mock to detect if DB is called again
    mock_cur.fetchone.reset_mock()
    
    # Next read should come from cache, not DB
    product = write_through_cache.get_product(1)
    
    assert product['price'] == 19.99
    mock_cur.fetchone.assert_not_called()

Pattern 4: Cache-Aside with Stampede Protection

When many requests hit a cold cache simultaneously, all might trigger DB fetches. Test that stampede protection works.

Implementation

# stampede_protected_cache.py
import json
import time
import redis
import threading

r = redis.Redis.from_url('redis://localhost', decode_responses=True)

LOCK_TIMEOUT = 5

def get_with_lock(key: str, fetch_fn, ttl: int = 300):
    cached = r.get(key)
    if cached:
        return json.loads(cached)
    
    lock_key = f'lock:{key}'
    lock = r.set(lock_key, '1', nx=True, ex=LOCK_TIMEOUT)
    
    if lock:
        # We got the lock — fetch and populate
        try:
            value = fetch_fn()
            r.setex(key, ttl, json.dumps(value))
            return value
        finally:
            r.delete(lock_key)
    else:
        # Another process is fetching — wait and retry
        deadline = time.time() + LOCK_TIMEOUT
        while time.time() < deadline:
            time.sleep(0.05)
            cached = r.get(key)
            if cached:
                return json.loads(cached)
        
        # Fallback: fetch directly if lock holder died
        return fetch_fn()

Testing Stampede Protection

# test_stampede_protection.py
import threading
import pytest
import fakeredis
from unittest.mock import MagicMock
import stampede_protected_cache

def test_only_one_fetch_under_concurrent_load(monkeypatch):
    fake_r = fakeredis.FakeRedis(decode_responses=True)
    monkeypatch.setattr(stampede_protected_cache, 'r', fake_r)
    
    call_count = 0
    event = threading.Event()
    
    def slow_fetch():
        nonlocal call_count
        call_count += 1
        event.wait(timeout=0.1)  # Simulate slow DB query
        return {'data': 'expensive_result'}
    
    results = []
    errors = []
    
    def make_request():
        try:
            result = stampede_protected_cache.get_with_lock('expensive_key', slow_fetch)
            results.append(result)
        except Exception as e:
            errors.append(e)
    
    threads = [threading.Thread(target=make_request) for _ in range(10)]
    for t in threads:
        t.start()
    
    event.set()  # Allow fetch to complete
    
    for t in threads:
        t.join(timeout=10)
    
    assert len(errors) == 0
    assert len(results) == 10
    assert all(r == {'data': 'expensive_result'} for r in results)
    
    # Only 1 actual DB fetch despite 10 concurrent requests
    assert call_count == 1

Pattern 5: Batch Invalidation with Key Patterns

Invalidating all keys matching a pattern (e.g., all cache entries for a given tenant).

Implementation

# batch_invalidation.py
import redis

r = redis.Redis.from_url('redis://localhost')

def invalidate_tenant_cache(tenant_id: str):
    """Invalidate all cache keys for a tenant."""
    pattern = f'tenant:{tenant_id}:*'
    cursor = 0
    deleted = 0
    
    while True:
        cursor, keys = r.scan(cursor=cursor, match=pattern, count=100)
        if keys:
            r.delete(*keys)
            deleted += len(keys)
        if cursor == 0:
            break
    
    return deleted

Testing Batch Invalidation

# test_batch_invalidation.py
import fakeredis
import pytest
import batch_invalidation

def test_invalidates_all_tenant_keys(monkeypatch):
    fake_r = fakeredis.FakeRedis()
    monkeypatch.setattr(batch_invalidation, 'r', fake_r)
    
    # Seed keys for tenant A
    for i in range(50):
        fake_r.set(f'tenant:A:user:{i}', f'data_{i}')
        fake_r.set(f'tenant:A:order:{i}', f'order_{i}')
    
    # Seed keys for tenant B (should not be affected)
    for i in range(10):
        fake_r.set(f'tenant:B:user:{i}', f'data_{i}')
    
    deleted = batch_invalidation.invalidate_tenant_cache('A')
    
    assert deleted == 100  # 50 users + 50 orders
    
    # All tenant A keys gone
    assert fake_r.keys('tenant:A:*') == []
    
    # Tenant B keys intact
    assert len(fake_r.keys('tenant:B:*')) == 10

def test_returns_zero_for_nonexistent_tenant(monkeypatch):
    fake_r = fakeredis.FakeRedis()
    monkeypatch.setattr(batch_invalidation, 'r', fake_r)
    
    deleted = batch_invalidation.invalidate_tenant_cache('nonexistent')
    assert deleted == 0

Testing Anti-Patterns to Avoid

Anti-Pattern 1: Asserting calls instead of state

# BAD: tests that invalidation was called, not that cache is actually empty
def test_bad_invalidation():
    with patch.object(redis_client, 'delete') as mock_delete:
        update_user(1, {'name': 'New'})
        mock_delete.assert_called_once_with('user:1')
    # This passes even if delete() is a no-op that doesn't work

# GOOD: test the actual state
def test_good_invalidation(fake_redis):
    fake_redis.set('user:1', json.dumps({'id': 1, 'name': 'Old'}))
    
    update_user(1, {'name': 'New'})
    
    assert fake_redis.get('user:1') is None  # Cache is actually empty

Anti-Pattern 2: Not testing the stale data case

Most cache tests only test that invalidation happens after an update. They don't test that stale data is served if invalidation is missing. Always add a regression test:

def test_stale_data_not_served_after_update(fake_redis):
    # Warm cache with old data
    fake_redis.setex('user:1', 3600, json.dumps({'name': 'Old Name'}))
    
    # Update in DB
    update_user_in_db(1, {'name': 'New Name'})
    
    # Without explicit cache invalidation, this would return stale data
    # This test documents that the invalidation MUST happen
    result = get_user(1)
    assert result['name'] == 'New Name', "Cache was not invalidated after DB update"

Anti-Pattern 3: Time.sleep in tests

Never sleep to wait for TTL expiry in tests. It makes tests slow and flaky. Use fakeredis2 with time control or delete keys directly to simulate expiry.

Conclusion

Cache invalidation tests must verify state, not method calls. Use fakeredis/redis-mock for fast unit tests of invalidation logic, Testcontainers when you need real Redis behavior. Always test the stale data case—what happens if invalidation fails—not just the happy path where it succeeds.

Read more

Start now free