Last Friday afternoon, I was sipping coffee and getting ready to wrap up the week when a message from QA landed in our group chat: “Users are complaining the AI can’t remember earlier conversations – check if the memory disappeared again.” My heart sank a little. I opened the logs – the Redis key still contained the conversation history, but the order was completely scrambled, with the latest message sitting at the front. What annoyed me even more was that this was already the third time this month we’d been bitten by an “invisible” memory storage bug.
Every time we tried to reproduce the issue, it involved spinning up three terminals, manually sending messages like “Hi, my name is Alice” and “What’s my name?” to the AI, then visually comparing the responses. “Manual validation” sounds generous – it was basically crossing our fingers.
I decided to put an end to the guesswork by building a Pytest + Redis test suite. The goal wasn’t just to check whether memories were saved, but to verify order, TTL expiry, and consistency under concurrent writes. After setting it up, a manual regression cycle that used to take 30 minutes now finishes in 3 minutes for 30+ test cases, and I caught three logical bugs that even code review had missed. This article reproduces the entire journey – all the code is ready to run.
Breaking down the problem: why your AI memory tests have gaps
Here’s the typical setup: an AI chat system stores every user–assistant turn in Redis, probably under a key like chat:memory:{session_id}. The value is a JSON array containing message objects in chronological order. When the user sends a new message, the backend fetches the history from Redis, prepends it to the prompt, and feeds everything to the LLM. Simple on paper, but riddled with traps in practice:
- Ordering issues – under concurrent requests, multiple writes can interleave, messing up the array order. A user asks “What’s the weather?” and then “What’s my name?”; the AI answers the name first and the weather later because in the stored memory the “name” message jumped ahead of the “weather” one.
-
TTL boundaries – the conversation memory is set to expire after 30 minutes, but if a new message is written at 29:59 and the
EXPIREreset happens at the wrong moment, the memory vanishes in the middle of the chat. -
Serialization inconsistencies – you save with
json.dumpsand load withjson.loads, but when multiple processes or services are involved,datetimeobjects can serialize differently (one process includes timezone info, another doesn’t). This causes memory comparisons to fail or even breaks the conversation flow with an error.
Manual testing typically looks like this: open a Redis client, insert a record by hand, call the API with curl, then reconnect to check if the data is still “correct”. This approach has two fatal flaws: when there are 50 turns in a conversation, no human can reliably verify the order and content of all messages. Moreover, TTL-dependent test cases require waiting 30 minutes – nobody actually waits that long in a test environment, so that logic is almost never exercised. And concurrency scenarios? Nearly impossible to simulate by hand.
What we really need is a test setup that can generate data, assert correctness automatically, fast‑forward time, and run in parallel.
Design decisions: why we test against a real Redis instance instead of mocking it
When testing external dependencies, people usually pick one of two extremes: mock everything (replace all Redis calls with fake objects) or connect directly to a remote test Redis. I tried mocking and hit several walls:
- Advanced features of
redis-pylike pipelines, Lua scripts, and blocking commands are extremely hard to fake realistically. For example, the timeout behavior ofbrpopin a mock is nothing like the real thing. - The very things we need to test are Redis’s real behaviors: TTL expiry, type conversions, and reconnection after a disconnect. Mocking would turn the tests into “testing my own mock code”, which is pointless.
Using a shared remote Redis instance also has downsides: parallel tests pollute each other’s keys, requiring constant cleanup, and network latency can cause flakiness. The sweet spot is testcontainers – we spin up a temporary Redis container that behaves exactly like a production instance but is fully isolated, and it’s destroyed automatically when the tests finish. Combined with Pytest fixtures and freezegun to freeze time, we can compress hours of time‑dependent tests into a few seconds.
The test strategy is layered into three levels:
- Unit tests – verify the CRUD operations and serialization logic of the memory handler.
- Integration tests – exercise the full request path (FastAPI + Redis).
- Consistency tests – validate memory order and completeness under concurrent writes.
Core implementation: solving a real pain point at every step
Let’s build the test suite step by step. All the code includes imports so you can drop it into a test_memory.py and run it.
1. Redis container fixture – solving environmental pollution
This snippet eliminates leftover data and instability issues. Each test class gets its own Redis container.
# conftest.py or top of test_memory.py
import pytest
from testcontainers.redis import RedisContainer
@pytest.fixture(scope="session")
def redis_container():
# session-scoped to boost speed – starts once
with RedisContainer("redis:7-alpine") as container:
container.start()
yield container
@pytest.fixture
def redis_client(redis_container):
import redis
client = redis.Redis(
host=redis_container.get_container_host_ip(),
port=redis_container.get_exposed_port(6379),
decode_responses=True # auto-decode for easy assertions
)
yield client
client.flushall() # clean after each test to avoid cross-contamination
2. Storing and reading memory – verifying basic correctness
We need to confirm that after saving conversation history, the retrieved list has the exact same length, content, and order, without any weird escaping.
import json
from datetime import datetime, timezone
class MemoryService:
"""实际业务代码(简化版),你项目里可能更复杂"""
def __init__(self, redis_client, ttl=1800):
self.redis = redis_client
self.ttl = ttl
def append_message(self, session_id: str, role: str, content: str):
key = f"chat:memory:{session_id}"
# 取出历史
raw = self.red
Top comments (0)