At 2 a.m., PagerDuty woke me up not for a production outage, but because the CI pipeline had turned red again—the test stage hit a hard timeout of 600 seconds. That moment felt even more frustrating than fixing a midnight bug: the code itself was fine, but the tests were just too slow. Every single run recreated data, re-called external APIs, and recomputed expensive fixtures from scratch.
The root cause was blunt: our tests had no memory. The same fixture that ran yesterday would be recomputed from scratch today on a different machine or branch. Some team members used .pytest_cache for local caching, but that can’t be shared across distributed CI nodes—a fresh machine still suffered the same painful first run. Even worse, we occasionally hit consistency failures where stale cache results didn’t reflect fresh data, and when a popular key expired, dozens of test workers simultaneously hammered the data source—classic cache stampede and penetration issues, all present.
Finally, I spent half a day embedding Redis directly into the Pytest fixture machinery, building a caching layer with memory storage, anti-penetration, and anti-stampede safeguards. Our test pipeline dropped from 200s to 15s—a 13x improvement—and never timed out in CI again. Here’s the full breakdown of the approach.
Problem Dissection: It’s Not Slow Code, It’s Wheel Reinvention
Our test suite had two typical characteristics:
- Many fixtures depend on expensive operations: complex database queries, model inference, external API calls (e.g., fetching tokens).
- The test environment was dynamically scaling: 5 concurrent nodes today, maybe 10 tomorrow. Each new runner started as a blank slate.
The usual approach was @pytest.fixture(scope="session") for in-process reuse, but that only works within a single node’s session. What truly slowed us down was cross-node, cross-session repeated computation.
Why not rely on .pytest_cache or local files? Because distributed CI doesn’t share file systems, and local caches can’t control expiration properly, leading to consistency pitfalls like “the code changed but the cache still returns yesterday’s result”. More critically:
- Cache stampede: When a hot key (e.g., a large query result) just expired, dozens of test workers simultaneously hit the source, overwhelming the database or external API.
- Cache penetration: Test code passes nonexistent parameters (e.g., a wrong user ID) that constantly bypass the cache and hit the backend, causing a flood of empty queries.
- Expiry consistency: After the source data updates, the cache still holds the old value, making assertions falsely pass while production breaks.
To fix this fundamentally, we needed a distributed, penetration-proof, and actively expirable centralized memory store. Redis fit that role perfectly.
Solution Design: Turning Redis into the Tests’ “Shared Memory”
My reasoning wasn’t “I want Redis.” It was “I need a cache that can be shared quickly across distributed test nodes and natively supports atomic operations.”
Candidate approaches:
- Local files + NFS: passable sharing, but no atomic ops—building custom locks and expiration is too painful.
- Memcached: pure cache with good expiry strategies, but weak data structures; implementing Bloom filters or null markers is awkward.
- Redis: strings, hashes, SETNX locks, Lua scripts, expiration—everything we needed from a mature ecosystem. No reason not to pick it.
Architecturally, I designed three layers of protection:
- Use SETNX-based mutual exclusion so that only the first process fetches the data for a given key while others wait—solving stampedes.
- For non-existent data, store a null marker with a short TTL to prevent penetration.
- When generating cache entries, embed a version or fixed TTL and provide explicit invalidation commands—developers or CI scripts can purge entries when the source data changes, solving consistency.
The final form: a lightweight Pytest-plugin-style conftest.py where developers simply wrap a fixture’s return value with a function called cached_fixture to gain memory storage.
Core Implementation: A Memory Layer from Zero to Usable
Here’s the runnable code step by step. You can drop it directly into your conftest.py.
Step 1: Redis Connection and Lock Utilities
This block handles safe Redis communication and provides basic locking. We create one Redis connection per Pytest session and release it automatically when the session ends.
# conftest.py
import pytest
import redis
import time
import hashlib
import json
from contextlib import contextmanager
REDIS_URL = "redis://localhost:6379/0"
@pytest.fixture(scope="session")
def redis_client():
"""Session‑level Redis connection, reused for the entire test lifecycle"""
client = redis.Redis.from_url(REDIS_URL, decode_responses=True)
yield client
client.close()
@contextmanager
def redis_lock(client: redis.Redis, key: str, timeout: int = 10):
"""
Simple mutual exclusion lock based on SETNX.
Serializes "fetch from source" operations on the same key to prevent stampedes.
"""
lock_key = f"lock:{key}"
lock_acquired = client.setnx(lock_key, "1")
if lock_acquired:
client.expire(lock_key, timeout)
try:
yield lock_acquired
finally:
if lock_acquired:
client.delete(lock_key)
Step 2: Core Memory‑Store Function
This section solves “how to store and retrieve fixture return values while handling penetration and expiration.” The function first checks the cache; on a miss, it applies null-marker protection, then uses the lock to prevent a stampede when fetching the real data.
python
NULL_MARKER = "__NULL__" # Marker for null values, distinct from None
DEFAULT_TTL = 3600 * 24 * 7 # Normal cache lives 7 days
NULL_TTL = 60 # Null marker only lives 60 seconds to allow quick correction
def _build_cache_key(func_name: str, args, kwargs) -> str:
"""Generate a unique cache key from function name and parameters, ensuring same inputs hit the same cache."""
params = json.dumps({"args": args, "kwargs": kwargs}, sort_keys=True)
raw = f"{func_name}:{params}"
return hashlib.md5(ra
Top comments (0)