Redis Test Data Setup for Cache and Session Testing

Testing code that talks to Redis is rarely a “set‑and‑forget” activity.
Cache layers and session stores have different lifecycles, eviction policies, and consistency expectations.
If the test data you feed them does not reflect those realities, the test suite will either hide bugs or surface flaky failures that waste engineering time.

This guide walks through a repeatable, evidence‑driven process for building Redis test data that works for both cache and session scenarios.
It covers prerequisites, design decisions, a worked example, validation steps, and the most common failure modes we see in production‑grade test suites.

---

1. Why Redis Test Data Is Different

Aspect	Cache	Session
Typical TTL	Seconds‑to‑minutes, often refreshed on read	Minutes‑to‑hours, refreshed on each request
Eviction policy	LRU, LFU, random, or custom	Usually TTL‑only; eviction = logout
Write pattern	Bulk pre‑warming, periodic refresh	Per‑user write on login, update on activity
Read pattern	High‑throughput, many keys per request	One key per request (session ID)
Consistency requirement	Eventual is acceptable	Strong – a stale session is a security bug

Because of these differences, a single “dump‑everything‑into‑Redis” script rarely satisfies both use cases.
You need two data‑generation pipelines that share a common infrastructure (connection handling, seeding, cleanup) but diverge on key design, TTL strategy, and payload shape.

---

2. Prerequisites

Item	Minimum version	Reason
Redis server	6.2+	Supports KEYS/SCAN, EXPIRE, SETNX, Lua scripting
Test runner	Jest, PyTest, Go test, etc.	Must be able to start/stop a Redis instance per test suite
Container runtime	Docker 20+	Guarantees identical Redis version across CI and local
Language client	ioredis (Node), redis-py, go-redis	Provides pipeline & transaction APIs
Data‑generation helper	QA3 free test data generator – https://qa3.io/tools/test-data-generator	Generates realistic JSON payloads, UUIDs, timestamps, and custom distributions without writing boilerplate

Tip: Spin up Redis in a disposable container for each test file (or each test case if you need isolation).
Example docker-compose.yml snippet:

services:
  redis:
    image: redis:7-alpine
    ports: ["6379:6379"]
    tmpfs: /data   # avoids persisting between runs

---

3. Data‑Model Decisions

3.1 Key Naming Conventions

Domain	Pattern	Example
Cache	cache:{domain}:{entity}:{id}	cache:product:details:42
Session	session:{tenant}:{userId}:{sessionId}	session:acme:u123:sess-9f1a

Use a colon‑delimited hierarchy. It makes SCAN patterns (cache:product:*) predictable and lets you namespace eviction scripts.

3.2 Payload Shape

Field	Cache	Session
data	Domain object (JSON)	Minimal user context (id, roles, csrf)
meta	{createdAt, version, source}	{createdAt, lastAccess, ip, ua}
ttl	Stored separately via EXPIRE	Stored separately via EXPIRE

Keep the session payload ≤ 1 KB. Large session blobs increase memory pressure and latency.
Cache payloads can be larger (up to a few MB) but should be compressed (e.g., gzip + base64) if they exceed 100 KB.

3.3 TTL Strategy

Strategy	When to use	Implementation
Fixed TTL per key type	Cache entries that never change	SET key value EX 300
Sliding TTL on read	Sessions, hot cache entries	EXPIRE key 1800 after each GET
Write‑through refresh	Cache that mirrors a DB table	Background job SET key value EX 600 every 5 min

Trade‑off: Sliding TTL adds a write on every read (extra RTT). Fixed TTL is cheaper but can serve stale data longer than acceptable.

---

4. Generation Pipelines

4.1 Shared Infrastructure

# conftest.py (pytest)


import redis, os, json
from qa3_test_data import generate  # wrapper around QA3 generator


REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")


@pytest.fixture(scope="session")
def redis_client():
    client = redis.Redis.from_url(REDIS_URL, decode_responses=True)
    yield client
    client.flushdb()   # clean slate after the whole suite

Why a session‑scoped fixture?
Spawning a new connection per test adds latency. A single client with flushdb at the end gives deterministic isolation without container churn.

4.2 Cache Data Generator

def seed_cache(redis_client, count=500):
    """Populate product‑detail cache with realistic payloads."""
    pipe = redis_client.pipeline()
    for _ in range(count):
        product = generate(
            schema={
                "id": "uuid",
                "name": "sentence",
                "price": {"type": "float", "min": 1, "max": 5000},
                "tags": {"type": "array", "items": "word", "max_len": 5},
                "inventory": {"type": "int", "min": 0, "max": 10000},
            },
            count=1
        )[0]


key = f"cache:product:details:{product['id']}"
        value = json.dumps({"data": product, "meta": {"createdAt": now_iso(), "version": 1}})
        pipe.set(key, value, ex=300)   # 5‑min fixed TTL
    pipe.execute()

Key points

• Pipeline – batches 500 SET commands into a single round‑trip.
• QA3 generator – produces varied, schema‑conformant JSON without hand‑crafted factories.
• Fixed TTL – matches the “cache‑warm” pattern; no sliding refresh needed.

4.3 Session Data Generator

def seed_sessions(redis_client, user_count=200, sessions_per_user=3):
    """Create realistic session records with sliding TTL."""
    pipe = redis_client.pipeline()
    for _ in range(user_count):
        user_id = f"u{generate({'type':'int','min':1,'max':100000})[0]}"
        for _ in range(sessions_per_user):
            sess_id = generate({"type":"uuid"})[0]
            key = f"session:acme:{user_id}:{sess_id}"
            payload = {
                "data": {
                    "userId": user_id,
                    "roles": generate({"type":"array","items":{"type":"choice","values":["admin","editor","viewer"]},"max_len":3})[0],
                    "csrf": generate({"type":"hex","length":32})[0],
                },
                "meta": {
                    "createdAt": now_iso(),
                    "lastAccess": now_iso(),
                    "ip": generate({"type":"ipv4"})[0],
                    "ua": generate({"type":"user_agent"})[0],
                },
            }
            pipe.set(key, json.dumps(payload), ex=1800)   # 30‑min initial TTL
    pipe.execute()

Differences from cache pipeline

• Per‑user grouping – enables SCAN session:acme:u123:* for logout‑all logic.
• Sliding TTL – the test harness will call EXPIRE on each simulated request (see validation section).
• Richer metadata – IP, user‑agent, CSRF token; useful for security‑focused tests.

---

5. Worked Example: End‑to‑End Test Flow

Below is a complete, runnable Python test module that demonstrates the whole lifecycle: seed → exercise → assert → cleanup.

# test_redis_cache_and_session.py


import json, time
import pytest
from redis import Redis


# ---------- helpers ----------


def now_iso():
    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())


def simulate_request(redis_client, session_key):
    """Mimic a web request: read session, slide TTL, return payload."""
    raw = redis_client.get(session_key)
    assert raw, "session missing"
    session = json.loads(raw)
    # slide TTL
    redis_client.expire(session_key, 1800)
    # update lastAccess
    session["meta"]["lastAccess"] = now_iso()
    redis_client.set(session_key, json.dumps(session), xx=True, ex=1800)
    return session["data"]


# ---------- fixtures ----------


@pytest.fixture(scope="module")
def redis_client():
    client = Redis.from_url("redis://localhost:6379/0", decode_responses=True)
    yield client
    client.flushdb()


@pytest.fixture(scope="module", autouse=True)
def seed_data(redis_client):
    seed_cache(redis_client, count=200)
    seed_sessions(redis_client, user_count=50, sessions_per_user=2)


# ---------- tests ----------


def test_cache_hit_rate(redis_client):
    """Verify that 95 % of random product IDs are cached."""
    import random
    keys = [f"cache:product:details:{i}" for i in range(200)]
    hits = 0
    for _ in range(100):
        key = random.choice(keys)
        if redis_client.exists(key):
            hits += 1
    assert hits / 100 >= 0.95


def test_session_sliding_ttl(redis_client):
    """Session TTL should be refreshed on each simulated request."""
    # pick a known session
    session_keys = redis_client.keys("session:acme:u1:*")
    assert session_keys
    key = session_keys[0]


# initial TTL
    ttl_before = redis_client.ttl(key)
    assert 1700 < ttl_before <= 1800   # allow small clock drift


# simulate two requests spaced 1 s apart
    simulate_request(redis_client, key)
    time.sleep(1)
    simulate_request(redis_client, key)


ttl_after = redis_client.ttl(key)
    # TTL should be reset to ~1800 again
    assert 1700 < ttl_after <= 1800


def test_logout_all_clears_user_sessions(redis_client):
    """Deleting by pattern removes only that user's sessions."""
    user_keys = redis_client.keys("session:acme:u42:*")
    assert len(user_keys) == 2
    redis_client.delete(*user_keys)
    remaining = redis_client.keys("session:acme:u42:*")
    assert remaining == []
    # other users untouched
    assert redis_client.keys("session:acme:u43:*") == 2

What the example proves

Test	What it validates
test_cache_hit_rate	Cache warm‑up succeeded; key naming & TTL are correct
test_session_sliding_ttl	Sliding‑TTL logic works under realistic request cadence
test_logout_all_clears_user_sessions	Pattern‑based deletion isolates tenants/users

Run with pytest -q test_redis_cache_and_session.py.
All three tests should pass in < 2 seconds on a modest CI box.

---

6. Validation & Verification Checklist

✅ Item	How to verify
Key namespace isolation	SCAN cache:* returns only cache keys; SCAN session:* returns only session keys
TTL correctness	TTL key matches the policy (fixed vs sliding) for a random sample ≥ 30 keys
Payload schema compliance	Deserialize a sample and run a JSON‑Schema validator (e.g., jsonschema)
Memory footprint	INFO memory after seeding ≤ expected budget (cache ≈ 200 KB, sessions ≈ 50 KB)
Eviction behavior	Fill Redis to maxmemory and confirm LRU evicts cache keys first (configure maxmemory-policy allkeys-lru)
Concurrency safety	Run the test suite in parallel (pytest -n 4) – no ConnectionError or key collisions
Deterministic cleanup	After each module, FLUSHDB leaves zero keys (DBSIZE == 0)
CI reproducibility	Same Docker image, same seed script → identical key count & TTL distribution across runs

Automate the checklist as a post‑seed verification step in your pipeline:

# ci/verify_redis_seed.sh


redis-cli --scan --pattern "cache:*" | wc -l   # expect 200
redis-cli --scan --pattern "session:*" | wc -l # expect 100
redis-cli INFO memory | grep used_memory_human

Fail the build if any count deviates > 5 % from the target.

---

7. Common Failure Modes & Mitigations

Symptom	Root Cause	Mitigation
Flaky TTL assertions	System clock drift between test container and Redis container	Use redis-cli TIME as the authoritative clock; compute expected TTL from that value
OOM in CI	maxmemory not set; test seeds more data than the default 3 GB limit	Add maxmemory 256mb + maxmemory-policy allkeys-lru to redis.conf used by CI
Key collisions across parallel test workers	Shared Redis DB (default 0) with no namespace per worker	Prefix keys with worker{N}: or spin up a Redis instance per worker (Docker --network host + dynamic port)
Stale cache data after DB update	Cache warm‑up runs once, but integration tests mutate the DB	Add a cache:invalidate:{entity}:{id} pub/sub channel; tests publish after DB writes
Session hijacking false positive	Test reuses a session ID across users	Generate session IDs with uuidv4 (128‑bit entropy) – QA3 generator does this automatically
Slow test suite	No pipelining; each SET/GET round‑trips	Always batch writes with pipeline(); batch reads with mget where possible
Data‑generator drift	Schema changes in production but generator not updated	Store generator schemas as code (JSON Schema files) and run a CI job that validates generated samples against the live schema

---

8. Scaling the Approach

Scale	Adjustments
< 10 k keys	Single Redis instance, in‑process seeding, flushdb per module
10 k – 1 M keys	Use Redis Cluster mode; seed per shard via redis-cli --cluster call; keep generator stateless
> 1 M keys	Pre‑generate JSON files, load with redis-cli --pipe (bulk import) → 10× faster than pipelined SET
Multi‑tenant	Add tenant prefix to every key (tenant:{id}:cache:…); seed per tenant in parallel jobs
Geo‑distributed	Seed local Redis replicas; use REPLICAOF to verify replication lag in tests

---

9. Tooling Recap

Need	Recommended tool
Schema‑driven test data