It was 1 a.m. when my testing colleague blew up my phone on DingTalk: “Not again! I taught it my preferences in the previous round, and now it’s forgotten everything – Redis still has the old values!” I shuddered as I scrolled through the chat history. This was already the third “memory loss” incident this month. Our AI application stores user context in Redis, and every model session relies on this memory. The problem? Redis write and read behavior combined with our business logic hid far more pitfalls than we imagined. Manual validation couldn’t cover them all, and we kept blaming network hiccups. That night, I decided to build a fully automated consistency test suite with pytest and Redis, determined to dig out every single memory storage trap.
Breaking down the problem
We maintain a multi-turn conversational AI service where each session’s “memory” – user preferences, past decisions, linked entities – lives as a JSON string in Redis, either as a plain String or within a Hash. Keys follow the pattern mem:{session_id}:{slot}. The write paths are diverse: real-time writes during conversations, asynchronous summary aggregation and backfill, expiration-based refreshes, and more. Reads happen in bulk via MGET before every model call.
The symptom was classic: users would complain that “the setting I just changed went back to default,” but an exported Redis snapshot showed the latest value. The root cause was rarely Redis itself; it was almost always in the application layer – write atomicity, serialization version conflicts, TTL logic, and multiple paths racing for the same key. To reproduce such bugs, you need stateful scenarios: consecutive dialogue turns, concurrent requests, key expiration and reload, and then assert that the data content, TTL, and structure are perfectly consistent. Manually preparing data for one round took over 20 minutes. Running 200 regression cases was impossible, and human eyes miss subtle JSON diffs.
Conventional unit tests (e.g., mocking Redis) are nearly useless for consistency scenarios. A mock won’t expire, won’t handle concurrent clients, and won’t expose serialisation bugs. What actually caused “memory loss” in production was real Redis behaviour combined with application race conditions. We had to replay those scenarios against a real Redis instance with automated regression.
Design
No surprises in the tool choice: pytest as the test framework, combined with redis-py connecting directly to a dedicated test Redis instance. To match production, we spin up Redis 7 via Docker Compose and automatically FLUSHDB before each test run. Why not use Redis’s mock mode or fakeredis? Fakeredis doesn’t support Streams, and its expiration semantics for certain commands differ subtly from real Redis. We’ve been burnt: a piece of code using EXAT with a precise Unix timestamp passed on fakeredis but caused memory to expire early in production due to a seconds-level discrepancy. A real instance is non-negotiable.
The architectural principle is simple: organize cases around a “memory contract.” For each memory slot, define the expected state transitions: creation (no TTL), update (value changes, TTL may reset), soft expiration (passive deletion then reload), and concurrent updates (multiple writers racing). Each test case verifies one contract. Pytest fixtures prepare a clean session, and the test makes strict assertions on value, TTL, and even the Redis type.
Core implementation
1. Test environment fixture: clean, isolated Redis connection
This fixture ensures environmental consistency for every single test case: connect to Redis, run FLUSHDB, and generate a fresh session ID – zero cross-test pollution.
import pytest
import redis
import uuid
from typing import Generator
@pytest.fixture(scope="function")
def redis_client() -> Generator[redis.Redis, None, None]:
"""创建连接，清库，确保每个用例独立的 Redis 环境"""
r = redis.Redis(
host="localhost",
port=6379,
db=0,
decode_responses=True, # 自动解码为字符串，方便断言
socket_connect_timeout=2
)
r.flushdb() # 关键：每次测试从零开始，避免脏数据
yield r
r.close()
@pytest.fixture
def session_id() -> str:
return f"sess:{uuid.uuid4().hex[:12]}"
2. Core test case: consistency across write, update, and read in a multi-turn conversation
This test simulates a common scenario: create memory, update it, then read it back. It verifies that the JSON payload, key existence, and TTL all match expectations across the lifecycle. It directly validates the write path most prone to “memory reversion.”
import json
from typing import Dict, Any
def put_memory(r: redis.Redis, sess: str, slot: str, payload: Dict[str, Any], ttl: int = None):
key = f"mem:{sess}:{slot}"
r.set(key, json.dumps(payload))
if ttl:
r.expire(key, ttl)
def get_memory(r: redis.Redis, sess: str, slot: str) -> Dict[str, Any]:
key = f"mem:{sess}:{slot}"
raw = r.get(key)
return json.loads(raw) if raw else {}
def test_memory_update_consistency(redis_client, session_id):
"""验证记忆更新后读取的一致性，以及 TTL 是否正确重置"""
r = redis_client
slot = "user_prefs"
payload_v1 = {"theme": "dark", "lang": "en"}
payload_v2 = {"theme": "light", "lang": "zh", "font_size": 14}
# 1. 初次写入，给一个较短的 TTL
put_memory(r, session_id, slot, payload_v1, ttl=30)
mem = get_memory(r, session_id, slot)
assert mem == payload_v1, "首次写入的内容应完全一致"
assert r.ttl(f"mem:{session_id}:{slot}") <= 30
# 2. 模拟业务更新：重新 PUT 新值，重置 TTL 到 60 秒
(The original article continues; this translation covers the beginning portion. A full translation would follow this structure and voice throughout.)
Top comments (0)