7.4Redis Testing

Redis is primarily used as a cache, session store, or message broker. Testing Redis focuses on TTL behavior, cache invalidation, and data structure correctness.

TTL Expiration

import redis
import time
import pytest

@pytest.fixture
def redis_client():
    client = redis.Redis(host="localhost", port=6379, db=1)
    yield client
    client.flushdb()  # Clean up after tests

def test_cache_ttl_expiration(redis_client):
    """Cached data should expire after TTL."""
    redis_client.setex("session:abc", 2, "user-data")  # 2 second TTL

    # Immediately available
    assert redis_client.get("session:abc") is not None

    # Wait for expiration
    time.sleep(3)
    assert redis_client.get("session:abc") is None

def test_cache_ttl_value(redis_client):
    """Verify the TTL is set correctly."""
    redis_client.setex("session:def", 300, "data")
    ttl = redis_client.ttl("session:def")
    assert 298 <= ttl <= 300  # Allow small timing variance

Cache Invalidation

def test_cache_invalidation_on_update(redis_client, api_client):
    """When a user is updated via API, their cache should be invalidated."""
    # Prime the cache
    api_client.get("/users/123")
    assert redis_client.get("user:123") is not None

    # Update the user
    api_client.put("/users/123", json={"name": "Updated Name"})

    # Cache should be invalidated
    assert redis_client.get("user:123") is None

    # Next GET should re-populate cache
    response = api_client.get("/users/123")
    assert response.json()["name"] == "Updated Name"
    assert redis_client.get("user:123") is not None

Redis Data Structures

def test_redis_sorted_set_leaderboard(redis_client):
    """Test a leaderboard using Redis sorted set."""
    redis_client.zadd("leaderboard", {"alice": 100, "bob": 85, "charlie": 92})

    # Get top 2
    top = redis_client.zrevrange("leaderboard", 0, 1, withscores=True)
    assert top[0] == (b"alice", 100.0)
    assert top[1] == (b"charlie", 92.0)

    # Get rank
    rank = redis_client.zrevrank("leaderboard", "bob")
    assert rank == 2  # 0-indexed: alice=0, charlie=1, bob=2

def test_redis_list_as_queue(redis_client):
    """Test a job queue using Redis list."""
    redis_client.rpush("job_queue", "job-1", "job-2", "job-3")
    assert redis_client.llen("job_queue") == 3

    # Process (FIFO)
    job = redis_client.lpop("job_queue")
    assert job == b"job-1"
    assert redis_client.llen("job_queue") == 2