DEV Community: BAOFUFAN

How Playwright’s Clock API Turned a 4-Hour Memory Decay Test into 10 Minutes

BAOFUFAN — Wed, 22 Jul 2026 01:04:30 +0000

At 2 a.m., my QA colleague spammed the group chat: “The memory decay algorithm you shipped yesterday seems broken again. Similarity search ranking is completely off — the top results are all old data.” I groggily opened the admin dashboard, copied in a few memories, then stared at the screen for an hour waiting for the decay to kick in. After that, I manually compared rankings… and debugged until dawn, only to discover the decay formula was missing a time coefficient. This kind of manual regression testing — waiting for real time and verifying rankings by hand — is painfully dumb. And honestly, this wasn’t the first time.

Are you still manually validating memory time decay?

“Memory storage” for large language models sounds fancy, but when you break it down, it’s essentially two things: semantic similarity ranking via vector search, and time decay that gives recent memories higher weight. Unit tests can guarantee the decay function itself isn’t broken, but end-to-end regression is a nightmare. You need to actually observe that one hour after a memory is written, its similarity score drops to the expected 0.8×, and after 24 hours it falls to 0.2×. If you rely on real time, testing a full decay curve can take days. And it’s not just the algorithm — the UI often has its own bugs, like pagination, highlighting, or timestamp display glitching after you adjust the decay coefficient.

The usual workaround is mocking time: freezegun on the backend, jest.useFakeTimers() on the frontend. But testing them separately never catches “backend and front end clocks out of sync” bugs. We even tried running Selenium for the UI and using backdoors to tweak database timestamps — to keep everything in sync we ended up writing a pile of hacky code that was a nightmare to maintain.

Why we chose Playwright — it gave the browser a “time accelerator”

When evaluating tools, we had a few hard requirements:

Operate both the browser UI and call backend APIs, so one regression run can validate the admin dashboard and create test data via API.
Accelerate the entire browser’s timeline — speed up Date.now(), setTimeout, requestAnimationFrame, and everything time-related together, not just mock a few APIs.
Maintainable scripts that a new teammate can pick up within an hour.

Cypress has cy.clock(), but it’s built on Sinon fake timers and can’t truly drive the whole browser timeline (WebSocket heartbeats, CSS animation timers, etc., go out of sync). Selenium offers almost no primitives for this. Playwright’s clock API, however, intercepts time at the browser process level. After you call page.clock.install(), a subsequent page.clock.fastForward('01:00:00') makes everything inside the browser believe an hour has passed — new Date(), setTimeout, and any third-party time library. That means you can turn a “wait 1 hour” test into a 1-second test without touching a line of frontend code.

The architecture is simple: Playwright script → concurrently calls API (to seed memory data) and drives the dashboard page → backend service → vector DB (Qdrant/Chroma) + decay calculation. The entire end-to-end verification runs inside a single process.

Core implementation: three steps to a time-accelerated regression test

Step 1: Launch a browser with a controlled clock

This snippet shows how to make the browser live on our timeline. The key is installing the fake timer before the page loads — otherwise the page initializes with real time, and subsequent fast-forwards drift.

# test_memory.py
from playwright.sync_api import sync_playwright, expect

def test_memory_time_decay():
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)
        context = browser.new_context()
        page = context.new_page()

        # Install fake timer before page load so all time operations run on simulated clock
        page.clock.install(time="2025-01-01T00:00:00Z")

        # Now navigate, the page's Date.now() starts from 2025-01-01
        page.goto("https://memory-dashboard.example.com")

        # Login and other setup omitted...
        # The dashboard will now display our simulated time

Step 2: Seed a memory via API and verify the initial ranking on the UI

This snippet shows how to mix API and UI operations for a complete decay-validation loop. We deliberately make the new memory’s vector slightly more similar to the query, but the old memory is “fresher” and gets a time-decay boost — so initially the old memory should rank first.

        # Use API to add two memories: one from 1 hour ago, one just created
        api_context = context.request
        # Add older memory (backend writes it with created_at = current simulated time - 1 hour)
        api_context.post("https://api.example.com/memories", json={
            "content": "User likes latte",
            "created_at": "2024-12-31T23:00:00Z",  # simulated time is 2025-01-01T00:00, so 1 hour ago
            "vector": [0.1, 0.9, 0.05]   # vector related to "latte"
        })
        # Add a newer memory with similar content but more recent timestamp
        api_context.post("https://api.example.com/memories", json={
            "content": "User often orders latte coffee",
            "created_at": "2025-01-01T00:00:00Z",
            "vector": [0.12, 0.88, 0.06]
        })

        # Go back to dashboard and search for "coffee preference"
        page.fill('input[placeholder="Search memories"]', "coffee preference")
        page.click('button:has-text("Search")')

        # The older memory still has high time-decay weight, so it should rank first
        first_result = page.locator(".memory-item").nth(0)
        expect(first_result).to_contain_text("User likes latte")

Step 3: Fast-forward time and verify the ranking flips

This is the soul of the whole approach — fastForward makes the decay effective instantly.

        # Fast-forward 1 hour to let decay take effect
        page.clock.fastForward("01:00:00")

        # Search again, now the newer memory should rank higher due to decay
        page.click('button:has-text("Search")')
        first_result = page.locator(".memory-item").nth(0)
        expect(first_result).to_contain_text("User often orders latte coffee")

With this setup, a full decay-curve regression that used to take hours (or even days) now completes in minutes — and you can test edge cases around clock boundaries without any real wait. Playwright’s clock API turned a maintenance headache into a reliable, readable test that any team member can extend. No more 2 a.m. bug hunts over a missing coefficient.

End the Guesswork: Automated AI Memory Testing Cuts Debug Time by 10

BAOFUFAN — Tue, 21 Jul 2026 12:04:31 +0000

At 2 a.m., I was jolted awake by an alert call. Our RAG‑powered AI customer support bot started "hallucinating"—even though the latest return policy was configured in the backend, it kept spitting out an old, deprecated version. I rushed into the database and found the new rule’s doc lying quietly in Chroma, but API queries couldn’t retrieve it at all. After digging deeper, I discovered that fine-tuned embeddings had left old caches uncleared; a vector dimension mismatch caused the upsert to fail silently. Yet the business logs didn’t even show a single WARN.

That’s when it hit me: We’ve connected AI with thousands of pieces of memory, but we never gave that memory an automated “health check.”

Problem Breakdown

In RAG / Agent architectures, the vector database serves as AI’s long-term memory: user profiles, knowledge chunks, tool descriptions—all encoded as vectors. The typical pipeline is: text → embedding → upsert → indexing → query. Every step can go wrong:

Silent write failures: upsert returns success but nothing actually lands on disk (for example, certain Chroma versions silently drop batches when limits are exceeded).
Inaccurate search: query similarity scores suddenly all become 1.0; turns out the normalization layer wasn’t applied.
Missing results: After writing to Milvus, you need flush() + load_collection() before the data becomes searchable. The load call is tucked away on page 8 of the official docs—newcomers miss it nine times out of ten.

How did we verify in the past? Open a Jupyter notebook, manually insert a few rows, then handcraft a curl query and eyeball the id and score. That ritual took at least 30 minutes each time and only covered the one or two cases I happened to think of. CI/CD never stood a chance.

Solution Design

To automate this, we need to use the vector database like a real consumer, not just call the SDK’s health_check. The core idea: sink the full write‑and‑query pipeline into pytest cases and rerun the regression on every commit.

I set three constraints for the implementation:

Isolation: Each test creates an independent collection / namespace and destroys it afterwards to prevent cross‑test pollution.
Determinism: Test vectors must not rely on randomness or external models. Use fixed‑dimension float arrays so you can assert exact score values.
Portability: Abstract a VectorDB protocol. Switching between Chroma, Milvus, or Qdrant only requires implementing insert and search—test cases stay the same.

Why not use a production‑database copy? Because you never know when a manual GC will suddenly spike latency and give you a flaky test. Why not mock the database? That would skip serialization, network transport, index building—exactly the stages where mistakes hide. It’s lying to yourself.

Core Implementation

1. Turn the vector database into a testable fixture

This snippet solves a critical problem: giving pytest a namespace that is automatically created and destroyed for every test function. We use Chroma as an example, but any object that implements search / insert can be plugged in.

# conftest.py
import uuid
import pytest
from chromadb import Client, Settings

class MemoryTester:
    """封装向量库的写入与查询，隐藏具体实现"""
    def __init__(self, collection_name: str):
        self.client = Client(Settings(anonymized_telemetry=False))
        self.collection = self.client.create_collection(
            name=collection_name,
            metadata={"hnsw:space": "cosine"}   # 距离度量固定为余弦距离
        )

    def insert(self, ids, docs, embeddings):
        self.collection.add(ids=ids, documents=docs, embeddings=embeddings)

    def search(self, query_embedding, top_k=3):
        return self.collection.query(
            query_embeddings=[query_embedding],
            n_results=top_k,
            include=["documents", "distances"]
        )

@pytest.fixture
def memory():
    """每个测试函数拿到的 memory 都是隔离的"""
    collection_name = f"test_{uuid.uuid4().hex[:8]}"
    tester = MemoryTester(collection_name)
    yield tester
    tester.client.delete_collection(collection_name)

Two tricky details: create_collection must explicitly set the distance metric—Chroma defaults to l2, while your production may use cosine, so the score distributions won’t match unless you specify it. Also, appending a random suffix to the collection name prevents parallel test runs from stepping on each other’s toes.

2. Write a “write‑then‑read” core test case

This case covers 80% of memory‑storage failures: insert three knowledge snippets, immediately query with the embedding of one snippet, and assert that the top-1 id is an exact match.

# test_memory_consistency.py
import numpy as np

def test_write_then_read_exact_match(memory):
    """插入三条记录后，用同一条 embedding 查询，应该原样返回自身"""
    docs = [
        "用户偏好：喜欢安静的环境",
        "用户偏好：对花粉过敏",
        "历史订单：2024-09-12 坚果礼盒"
    ]
    dim = 128
    # 固定向量，方便复现
    embeddings = np.eye(len(docs), dim, dtype=np.float32)
    ids = [f"user_pref_{i}" for i in range(len(docs))]

    memory.insert(ids=ids, docs=docs, embeddings=embeddings.tolist())

    # 用第一条记录的 embedding 检索自己
    result = memory.

From 40% to 0: Automating LLM Memory Regression with Playwright+VCR

BAOFUFAN — Tue, 21 Jul 2026 01:05:54 +0000

At 1 a.m., I was jolted awake by a barrage of DingTalk messages. Our product manager shouted in the group chat: “Users are complaining that our AI assistant completely forgot the ‘emergency contact’ they set up last week, even though it was mentioned in a previous conversation!” Groggily, I opened the logs and scrolled through the chat history. The model’s latest response indeed did not match the historical memory—a classic memory hallucination. But the previous version had worked fine. Who had changed the memory storage logic? No one admitted it. That night I manually compared 30 conversation threads until my eyes glazed over, then made a gut decision: rollback. Later I found myself wondering, why, in 2025, are we still relying on “manual eyeballing” to verify large model memory? That question directly led to our automated regression approach using Playwright + VCR.

The Problem: Why Testing Large Model Memory Is So Hard

Our product has a “long‑term memory” feature: the model remembers a user’s personalised information (name, preferences, recently discussed projects) across multi‑turn conversations and recalls it later. The memory storage layer is built on a vector database and a summarisation engine. Every time we upgrade the memory extraction logic, swap the embedding model, or tweak a prompt, we risk a regression—either missing memory entries or generating completely wrong new memories (hallucination). The manual regression flow was: open the chat window, play the user role by following a scripted conversation, then manually verify that the model remembered what it should and did not fabricate anything. This flow had three fatal pain points:

Time‑consuming and unreliable: Running a 20‑turn script took a skilled tester 15 minutes. By the fourth repetition fatigue kicked in, and the false‑positive rate shot up when comparing similar dialogues. Our product manager once marked a hallucinated answer as “fine” because it “looked reasonable enough.”
Narrow coverage: We could only spot‑check typical paths. Changing a single line of memory deduplication code left us completely unsure whether edge‑case long conversations were affected.
Environment differences: Something that worked perfectly locally might behave differently in staging due to different model versions or API latency. Manual testing could never cover that gap.

What about conventional test approaches? Unit tests that mock the LLM API only verify the storage layer’s code logic—they completely miss the model’s real unpredictability, the timing of frontend interactions, and the exact moment streaming triggers memory extraction. Selenium scripts can control the browser, but asserting memory content requires a pile of sleep calls and fragile selectors; the maintenance cost ends up higher than the system under test. What we needed was a method that worked like a video replay: freeze the entire user–AI interaction as a baseline, then automatically compare every future run—any deviation is a bug.

Solution Design: Why Playwright + VCR Mode

The selection logic was straightforward: use Playwright to drive a real browser and simulate user input, clicks, and waiting for streaming responses. At the same time, borrow the VCR (Video Cassette Recorder) metaphor: record the “conversation sequence + AI response text + key DOM snapshots” of each test run. The first run is set to record mode, producing a baseline. Subsequent regression runs automatically enter replay mode and perform a turn‑by‑turn comparison with the baseline. Any text difference, missing memory, or hallucination contamination gets caught immediately.

Why not the alternatives?

Cypress: Its support for multi‑tab and WebSocket streaming updates isn’t as native as Playwright’s, and our chat interface uses SSE streaming.
pytest-vcr (HTTP mocking): It records and replays LLM API calls, which saves tokens, but also means you are forever testing against the same stale model responses—you’d never catch new hallucinations introduced by a model upgrade or a memory prompt change.
Pure screenshot comparison (visual regression): AI‑generated text length varies wildly; a single scroll offset can turn the entire pixel grid red and drive your false‑positive rate through the roof. We therefore compare only structured text and the existence of certain CSS classes.

Our final architecture: pytest + Playwright synchronous API + a lightweight custom VCR tool we named ChatVCR. It serialises each conversational turn (user input, expected memory assertion, actual AI response) into JSON. Baselines are stored under tests/cassettes/, with paths generated automatically from test function names. A --vcr-record CLI flag controls recording vs. replaying.

Core Implementation: A Truly Usable ChatVCR Test Framework

The code below solves one core problem: how to let a Playwright test script automatically record conversation baselines, then compare them precisely during regression while avoiding interference from random text (timestamps, IDs).

1. The ChatVCR base class – responsible for recording and comparison

This snippet defines the ChatVCR class, wrapping conversation save/load and smart comparison logic. The key is that during comparison a strip_dynamic_text function uses regular expressions to clean dynamic content that changes across runs—otherwise the baseline would never match.

# chat_vcr.py
import json
import re
from pathlib import Path
from typing import Any, Dict, List

class ChatVCR:
    def __init__(self, cassette_path: str, record_mode: bool):
        self.path = Path(cassette_path)
        self.record_mode = record_mode
        self.dialogues: List[Dict[str, Any]] = []

    def add_turn(self, user_msg: str, assistant_response: str, expected_memory: str = ""):
        """录制一轮对话"""
        self.dialogues.append({
            "user": user_msg,
            "assistant": self._clean(assistant_response),
            "expected_memory": expected_memory
        })

    def save(self):
        if self.record_mode:
            self.path.parent.mkdir(parents=True, exist_ok=True)
            self.path.write_text(json.dumps(self.dialogues, ensure_ascii=False, indent=2))

    def load_expected(self) -> List[Dict[str, Any]]:
        if not self.path.exists():
            raise FileNotFoundError(f"Baseline not found at {self.path}, run with --vcr-record first")
        return json.loads(self.path.read_text())

    de

How We Cut Memory Hallucination Debugging from 6 Hours to 30 Minutes with Playwright + pytest

BAOFUFAN — Mon, 20 Jul 2026 12:04:12 +0000

At 1 AM, the product manager dropped a screenshot into the group chat. A user had asked, "How’s that project I mentioned going?" and the AI answered, "You haven’t told me about any project." The user fired back with a chat history screenshot that clearly showed a project description from three days earlier. The PM said: "You call this memory? This is amnesia."

I knew instantly — another hallucination regression. The memory storage module had been touched, but no one had verified it. And running a full regression manually across the entire conversation flow would take at least half a day. That night, after fixing the bug, I committed a Playwright + pytest end-to-end verification script and swore: if anyone ever made me test memory manually again, I’d slam the test report on their desk.

Why memory storage regression is so painful

Our AI application’s killer feature is cross-session memory — everything a user mentions in any conversation gets stored in a persistent memory store (pgvector + vector index) and is automatically recalled in future chats. Sounds great, but the "cross-session" part is exactly where it hurts. To verify that memory really works, you must simulate full multi-turn, multi-session interactions and check that the AI’s responses contain the expected memory fragments.

Our old regression workflow looked like this:

Manually call the /chat endpoint via Swagger and paste a message containing personal info (e.g., "I’m working on a project called Phoenix").
Wait for the backend to asynchronously chunk the memory, generate embeddings, and write to pgvector.
Open a new session and ask, "What was the project I mentioned earlier?"
Visually inspect the response for the word "Phoenix."

Every time we touched the memory module — even just adjusting the chunk size — we had to repeat that cycle N times, covering edge cases (Chinese, English, special characters, very long texts, etc.). A complete regression took at least 6 hours, and human eyes easily missed hallucinations (the AI could sound right but substitute the wrong entity). Worse, memory writes are asynchronous, so waiting relied on sleeps; during manual testing we constantly ran into timeouts or incomplete writes, leading to a sky‑high false‑positive rate.

Why Playwright + pytest, and not something else

We needed an end‑to‑end framework that could realistically simulate user interactions in the browser, wait for async results, and assert on natural language fragments. A few options were on the table:

Pure API tests (requests + pytest): You can hit endpoints directly, but you can’t carry the frontend‑side context (headers, cookies, session IDs in localStorage). Our app streams responses over SSE, which is painful to simulate with plain requests.
Cypress: Great browser automation, but our backend stack is Python, the team isn’t comfortable with JS/TS, and Cypress’s parallelism and fixture management are less flexible than pytest.
Playwright + pytest: Playwright ships an official pytest-playwright plugin. You write tests in Python, reuse all of pytest’s fixtures, parametrization, conftest, etc. You control Chromium, grab page DOM or network responses, and can even call backend APIs from inside tests for data setup/teardown. Most importantly, page.wait_for_response lets you wait for async requests to finish naturally, eliminating hard‑coded sleeps.

Architecturally, we split things into three layers:

Page interaction layer: Playwright opens the frontend, types into the chat input, clicks send, and reads the AI reply from the streaming output.
Verification layer: Extract the AI’s reply from the DOM, assert it contains a known memory word (like "Phoenix"), and use negative assertions to ensure no hallucination (e.g., "Project X" should not appear).
Data helper layer: Before a test, seeds known memories directly via the API or cleans the database to guarantee a clean environment.

Core implementation: setting up memory E2E verification from scratch

1. Defining test fixtures to isolate environments

What this solves: each test case needs its own browser context, a logged‑in state, and a cleanup step that wipes memories afterward so tests don’t pollute each other.

# conftest.py
import pytest
from playwright.sync_api import Page, BrowserContext

@pytest.fixture(scope="session")
def browser_context(browser: BrowserContext):
    # 设置全局上下文，如视口大小、忽略HTTPS错误
    context = browser.new_context(
        viewport={"width": 1440, "height": 900},
        ignore_https_errors=True
    )
    yield context
    context.close()

@pytest.fixture(scope="function")
def page(browser_context: BrowserContext) -> Page:
    page = browser_context.new_page()
    # 先登录，获取token存入localStorage
    page.goto("https://ai-app.example.com/login")
    page.fill('input[name="email"]', "test@example.com")
    page.fill('input[name="password"]', "testpass")
    page.click('button[type="submit"]')
    # 等待登录完成，跳转到聊天页
    page.wait_for_url("**/chat")
    yield page
    # 清理：调用后端API删除该测试用户的所有记忆，防止数据残留
    page.request.delete("https://api.ai-app.example.com/memories/test@example.com")
    page.close()

2. Wrapping conversation and memory verification into a reusable helper

What this solves: abstracting “send a message → get the reply → assert memory” into a reusable function that supports streaming waits and memory‑fragment checks.

# test_helpers.py
from playwright.sync_api import Page
import re

def send_and_get_reply(page: Page, message: str, timeout: int = 15000) -> str:
    """发送消息并等待AI完整回复（SSE流式结束后提取文本）"""
    page.fill('textarea[data-testid="chat-input"]', message)
    page.click('button[data-testid="send"]')
    # 关键点：等待消息列表中最后一条AI消息的出现，且内部文本不再变化
    # 因为SSE流式更新，我们等待消息气泡出现且包含内容长度稳定
    last_msg = page.locator('[data-testid="message-bubble"]:last-child')
    last_msg.wait_for(state="visible", timeout=timeout)
    # 等待文本长度稳定（即流式结束，简单实现：连续两次长度一致）
    previous_len = -1
    for _ in range(10):
        current_len = len(last_msg.inner_text())
        if current_len == previous_len and current_len > 0:
            break
        previous_len = current_len
        page.wait_for_timeout(500)
    return last_msg.inner_text()

def assert_memory_recall(page: Page, memory_word: str, question: str):
    """发送问题并断言回复中包含记忆词"""
    reply = send_and_get_reply(page, question)
    assert memory_word.lower() in reply.lower(), \
        f"Expected memory word '{memory_word}' not found in reply: {reply[:200]}"

def assert_no_hallucination(page: Page, prohibited_word: str, question: str):
    """断言回复中不包含特定幻觉词"""
    reply = send_and_get_reply(page, question)
    assert prohibited_word.lower() not in reply.lower(), \
        f"Hallucination detected: '{prohibited_word}' found in reply: {reply[:200]}"

From Manual to Automated: 10x Faster LLM Memory Bug Detection (with 3 Pitfalls)

BAOFUFAN — Mon, 20 Jul 2026 01:04:37 +0000

At 1 a.m., a colleague dropped a screenshot into our group chat: the production chatbot had completely forgotten a user’s context from three days earlier and was responding with total nonsense. My first reaction was “that’s impossible, Redis still has the data.” I logged in — the keys were almost all gone, with only a few newly written records from today. That was the moment I knew our old way of manually testing memory (“send a message, scroll up, see if context is still there”) belonged in the trash.

Breaking down the problem

Storing conversation memory for large language models seems simple at first: one session ID maps to one list of messages, stuffed into a Redis List or ZSET. Read, write, done. But real-world usage introduces a lot of ugly edge cases:

Context window trimming: Models have token limits, so the oldest messages must be dropped automatically. If that logic is slightly off, the bot effectively develops amnesia in production.
Expiration and eviction policies: Memory gets a TTL, but what happens when Redis hits max memory? How do you guarantee critical sessions aren’t silently kicked out?
Concurrent writes: Multiple messages pouring into the same session simultaneously must preserve order without overwrites.
Environment differences: Locally everything hums along using fakeredis, but live Redis behaves differently — sometimes radically so.

Manual testing only ever covers the happy path of “write one, read one.” The edge cases above barely get touched. What we really needed was an automated test suite that validates memory storage correctness, runs in CI, and exposes environment discrepancies.

Designing the approach

The technology choices were straightforward: Pytest + real Redis (not fakeredis).

Why not fakeredis?

Fakeredis is a pure-Python Redis mock. It’s great because you don’t need to install Redis, but its handling of expiration, eviction policies, and Lua scripts subtly differs from the real thing. Once your memory logic relies on TTL or LRANGE + LTRIM, a passing test means very little. The decision was clear: tests must use a real Redis instance, spun up via Docker in CI.

Why not unittest?

Pytest fixtures are a natural fit for managing Redis connections and session isolation, and parametrized tests let you cover multiple model configurations (GPT‑4, Claude, etc.) in one sweep.

Overall, we define a MemoryStore abstraction that encapsulates appending, querying, trimming, and expiration. Pytest fixtures inject a Redis client and generate a unique namespace (key prefix) per test function to avoid parallel collision. Then pytest.mark.parametrize feeds in different window sizes and message lengths to validate overall correctness in batch.

Core implementation

The class below encapsulates Redis List operations and implements context window trimming.

import time
import json
from typing import List, Dict, Optional
import redis

class RedisMemoryStore:
    """Redis-List-based conversation memory with automatic window trimming"""

    def __init__(self, client: redis.Redis, max_tokens: int = 2000):
        self.client = client
        self.max_tokens = max_tokens   # using char count as a rough token limit

    def _session_key(self, session_id: str) -> str:
        return f"mem:{session_id}"

    def append(self, session_id: str, role: str, content: str, ttl: int = 3600):
        """Append a message, automatically trim old ones that exceed the window"""
        key = self._session_key(session_id)
        msg = json.dumps({"role": role, "content": content, "ts": time.time()})
        pipe = self.client.pipeline()
        pipe.rpush(key, msg)
        # Trimming logic: delete from the oldest until total chars are within limit
        pipe.lrange(key, 0, -1)
        pipe.expire(key, ttl)
        results = pipe.execute()
        all_msgs = [json.loads(m) for m in results[1]]
        total = sum(len(m["content"]) for m in all_msgs)
        # Note: ltrim removes from the head (oldest first)
        while total > self.max_tokens and all_msgs:
            total -= len(all_msgs[0]["content"])
            all_msgs.pop(0)
        # Keep only the trimmed portion
        if len(all_msgs) < len(results[1]):
            self.client.ltrim(key, -len(all_msgs), -1)

    def get(self, session_id: str, n: Optional[int] = None) -> List[Dict]:
        """Fetch the last n messages, or all if n is None"""
        key = self._session_key(session_id)
        end = -1 if n is None else -(n+1)
        msgs = self.client.lrange(key, -n, -1) if n else self.client.lrange(key, 0, -1)
        return [json.loads(m) for m in msgs]

The following Pytest fixture provides each test function with an isolated namespace to avoid parallel-test interference.

# conftest.py
import pytest
import redis
import uuid

@pytest.fixture(scope="function")
def redis_client():
    """Each test function gets a dedicated Redis DB and key prefix"""
    r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)
    # Isolated namespace
    namespace

Agent Memory Testing with SQLite Snapshots: 10x Debugging Efficiency

BAOFUFAN — Sun, 19 Jul 2026 12:03:38 +0000

At 1 a.m., a coworker dropped a screenshot into the team chat: after 20 turns of continuous conversation, our client’s AI customer-support agent had “forgotten” an order number that was confirmed earlier. Worse still, this was the third memory-loss incident this month. I stared at the pile of assert agent.memory.get("xxx") == "yyy" in the code and it hit me: we aren’t bad at writing tests; we just never figured out how to verify “persisted memory.”

Are you really testing that the memory hit the disk?

Most teams test agent memory the same way: spin up the agent, feed it a few messages, look it up in memory, call save() or trigger auto-persistence, then read it back and compare. The problem? How do you know the data actually came from disk and isn’t still hanging around as a Python object in RAM?

The most absurd case I’ve seen was a LangChain app where ConversationBufferMemory worked perfectly in tests—all green. In production, as soon as a user switched session IDs, every memory vanished. The reason was simple: the dev tests never broke the object reference on chat_history. They always directly asserted assert memory.buffer, and that attribute was the very same in-memory list. The tests were lying.

True persistence verification has to cross processes—or even separate runs. Write the data, kill the process, restart, and then confirm the data is still there. Writing such tests by hand is painful: you have to manage temporary files, control lifecycle, handle serialization formats, and guarantee isolation between test cases. The conventional approach isn’t impossible—it’s just too slow, too brittle, and never turns into a reusable pattern.

Why not mock, why not Redis?

Initially we thought about mocking the filesystem, but we quickly dropped it. You mock open(), the test passes, but what if the real environment has a write‑permission issue? Or if the serialization uses pickle and a version bump breaks everything? A mock can only prove “the code called a function”—it can’t prove “the data can actually be read back.”

We also considered Redis as a memory backend, connecting to a local Redis instance in tests. That does verify persistence, but the downsides are obvious: every developer machine needs Redis running, and CI requires a service container. A config mismatch leads to the eerie “green locally, red in CI” mystery. Plus Redis’s key expiry policies and persistence mechanisms (RDB/AOF) add another layer of complexity—you just want to test the Agent’s logic, but you end up having to understand the full Redis ops story.

In the end we picked SQLite. Three reasons:

Zero dependencies, zero configuration — the Python standard library ships with sqlite3, no external services needed, works straight out of the box in CI.
Real file-based persistence — data goes into a .db file. Kill the process, switch Python interpreters, it’s still readable.
Native “snapshot” semantics — treat the entire SQLite file as an immutable snapshot. The test framework only has to do one thing: assert that the current snapshot matches the expected snapshot.

That lets us use pytest’s snapshot testing pattern, validating the persistence behavior of agent memory just like frontend folks test UI components.

Core implementation: flattening memory into a SQLite snapshot

The overall idea has three steps: use a pytest fixture to create a temporary SQLite database and inject it into the agent’s memory backend; after the test runs, dump the entire database content into a comparable text snapshot; then use a custom assert_snapshot function to compare the current snapshot against the previously recorded “golden snapshot.”

1. Creating an injectable SQLite memory backend

This code solves a single problem: provide a lightweight memory backend that can replace production Redis/Postgres and fully adheres to the agent’s interface. Here I’ll use LangChain’s BaseChatMessageHistory as an example and implement a custom SQLiteChatHistory.


python
import sqlite3
import json
from typing import List
from langchain.memory.chat_message_histories.in_memory import BaseChatMessageHistory
from langchain.schema import BaseMessage, messages_from_dict, messages_to_dict

class SQLiteChatHistory(BaseChatMessageHistory):
    """基于 SQLite 的对话历史，每条 session 一张表，方便快照导出"""

    def __init__(self, db_path: str, session_id: str):
        self.db_path = db_path
        self.session_id = session_id
        self._init_table()

    def _init_table(self):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute(f"""
                CREATE TABLE IF NOT EXISTS {self._table_name()} (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    message_json TEXT NOT NULL,
                    created_at DATETIME DEFAULT CURRENT_TIMESTAMP
                )
            """)

    def _table_name(self) -> str:
        # SQLite 表名不能以数字开头，加个前缀
        return f"session_{self.session_id.replace('-', '_')}"

    @property
    def messages(self) -> List[BaseMessage]:
        with sqlite3.connect(self.db_path) as conn:
            rows = conn.execute(
                f"SELECT message_json FROM {self._table_name()} ORDER BY id"
            ).fetchall()
        items = [json.loads(row[0]) for row in rows]
        return messages_from_dict(items)

    def add_message(self, message: BaseMessage) -> None:
        msg_dict = messages_to_dict([message])[0]

One Missed Test Case Cost Me 8 Hours — How I Built a Zero-Regression Memory Test Suite with Pytest + Docker

BAOFUFAN — Sun, 19 Jul 2026 01:05:26 +0000

I got woken up by an alert call at 2 a.m. – the production chatbot had suddenly developed “amnesia.” It was completely unable to recall order information we had just discussed that morning. I scrambled to investigate and discovered the root cause: we had migrated the memory storage from Redis to PostgreSQL + pgvector. The migration script looked fine, but one historical memory entry’s vector index hadn’t been properly built, causing the similarity recall to return empty results. Before going live, I had manually spot-checked 20 conversations as a regression check – and you guessed it, the one broken entry was the 21st I missed. From that moment on, I resolved that regression testing for LLM memory storage must be automated, and every change must replay historical memories. This post shares how I built that automated regression suite with Pytest + Docker, along with two near-fatal pitfalls along the way.

Problem Breakdown: Why Memory Storage Breaks Silently

“LLM memory storage” is the mechanism that stores chat histories, user preferences, summary snippets, etc., so the bot can retrieve them later via vector similarity or keyword search and inject them into prompts as context. Common implementations include Redis (with RediSearch/Vector modules), Milvus, PostgreSQL + pgvector, or even local FAISS files.

The biggest engineering pain point isn’t “can we store it?” but rather “does the system still remember after we upgrade?” Memory storage has many churn points: switching the underlying database, changing the embedding model, tweaking similarity thresholds, altering expiration policies, or even bumping a minor pgvector version. Any of these can cause missing memories, score deviations, or empty result lists – but traditional manual tests only cover a few happy paths and simply can’t cover weeks of accumulated historical data.

The conventional approaches aren’t great either: full end-to-end load testing with manual comparison is slow and irreproducible; mocking the storage layer to test business logic bypasses the exact vector retrieval logic that’s most fragile. What you really need is an environment that can quickly spin up a real storage instance, replay real historical data, and assert that every memory is correctly recalled. That’s exactly where Pytest + Docker shine.

Design Decisions: Why Not an Off-the-Shelf Integration Test Platform?

I considered several options:

A permanent test database: requires dedicated infrastructure, easily polluted by test data, and breaks when multiple people run tests in parallel.
Pure unit tests with mocks: never exercises real vector retrieval behavior, so the false-negative rate is dangerously high.
Pytest + Docker combo: each test session pulls a clean container and destroys it afterwards – perfect isolation. Plus, you can parameterize tests against multiple storage backends easily.

The core idea is to use testcontainers-python to dynamically manage storage containers. During regression tests, you load a snapshot of historical data (sampled from production or generated golden files), then call the memory storage service’s query APIs and assert that the recalled IDs, scores, and ordering match expectations. If someone changes storage logic or upgrades a model, running this regression suite will tell you within 10 minutes whether old memories are broken.

Core Implementation: From Container Management to Regression Assertions

1. Managing Real Storage Environments with testcontainers

This code solves “each test needs a fresh, disposable database.” I chose PostgreSQL + pgvector because it’s the most common and hides the most surprises.

# conftest.py
import pytest
from testcontainers.postgres import PostgresContainer
from testcontainers.redis import RedisContainer
import psycopg2
import redis

# Custom pgvector container: the official postgres image doesn't include pgvector
class PostgresWithVector(PostgresContainer):
    def __init__(self, image="pgvector/pgvector:pg16"):
        super().__init__(image=image)
        self.with_env("POSTGRES_USER", "test")
        self.with_env("POSTGRES_PASSWORD", "test")
        self.with_env("POSTGRES_DB", "memory_test")

@pytest.fixture(scope="session")
def pg_vector_store():
    """Start a pgvector container and return usable connection params"""
    postgres = PostgresWithVector()
    postgres.start()
    # Get the dynamically mapped host port
    host = postgres.get_container_host_ip()
    port = postgres.get_exposed_port(5432)
    dsn = f"postgresql://test:test@{host}:{port}/memory_test"
    # Initialize schema and vector extension
    conn = psycopg2.connect(dsn)
    with conn.cursor() as cur:
        cur.execute("CREATE EXTENSION IF NOT EXISTS vector;")
        cur.execute("""
            CREATE TABLE IF NOT EXISTS memory_blocks (
                id TEXT PRIMARY KEY,
                user_id TEXT,
                content TEXT,
                embedding vector(1536),
                created_at TIMESTAMP DEFAULT NOW()
            );
            CREATE INDEX ON memory_blocks USING ivfflat (embedding vector_cosine_ops);
        """)
    conn.close()
    yield {"dsn": dsn, "host": host, "port": port}
    postgres.stop()

@pytest.fixture(scope="session")
def redis_memory_store():
    """Similarly, use Redis as an alternative storage for comparison tests"""
    redis_container = RedisContainer("redis/redis-stack-server:latest")
    redis_container.start()
    host = redis_container.get_container_host_ip

Taking Over LLM Memory Store Testing with Pytest: 90% Fewer State Inconsistencies

BAOFUFAN — Sat, 18 Jul 2026 12:05:08 +0000

At 2:17 a.m., a Slack message exploded with user feedback: “Your AI assistant acts like it has amnesia—it just stored meeting minutes, but when I ask again, it says nothing was saved.” I dug into the logs and found that the Redis write was acknowledged, but the entry in the vector store was empty. This “intermittent amnesia” had already pissed off the user for the third time. Previously, we’d just manually curl a few conversation turns and call that a test pass—clearly that approach was broken. I was forced to set a rule the next morning: All state‑consistency logic for memory storage must be automatically verified by Pytest, or it’s not allowed to land on main.

Problem Breakdown

An LLM’s “memory store” typically isn’t a single backend: short‑term memory lives in a Redis List or an in‑memory ring buffer, long‑term memory gets embedded and pushed to a vector database (Qdrant/Milvus), and there’s often a raw text copy in PostgreSQL. During a single save_memory call, if you just write to Redis and PG synchronously and then push to the vector store asynchronously, there’s a window where Redis has already returned but the vector store hasn’t persisted—and a read request during that window can get incomplete history.

Even nastier is “memory compaction.” When the conversation length exceeds a threshold, you trigger summary generation while also cleaning up original messages and updating summary vectors. If any step fails, you end up with both “compacted” and leftover original messages coexisting, leading to duplicate or contradictory content.

The usual remedy is adding a manual verify step or running reconciliation scripts in production to sweep up dirty data. This has three fatal flaws:

Bugs during async windows are nearly impossible to reproduce manually; eyeballing logs is searching for a needle in a haystack.
Reconciliation scripts are post‑mortem bandaids, not development‑stage safety nets.
New team members have no idea which scenarios will break; the illusion of “it works” shatters at the slightest nudge.

So I decided to bake state‑consistency verification directly into a Pytest test suite and let CI do the heavy lifting.

Solution Design

In terms of tooling, the test framework was non‑negotiable: Pytest. The reasons are simple—fixtures elegantly manage the lifecycle of multiple storage backends, and parametrize can cover dozens of failure combinations in one go.

Why not the seemingly obvious unit tests?

Unit tests mock a single layer and can’t verify cross‑storage concurrency races. Mocked return values always follow your script; they never reveal a real window.
Integration tests with Docker Compose spin up real Redis and vector stores, but managing them directly in Pytest means writing a ton of boilerplate. We use pytest-docker or simply testcontainers-python so each test case gets isolated, fresh storage instances that are automatically destroyed after the run.

The architectural idea is straightforward: each test case follows a pattern of concurrent writes + multiple reads + eventual assertions. We simulate two coroutines simultaneously writing messages for the same user, then trigger a compaction signal, and finally poll all backends with an assert_eventually style until they are consistent. No guessing with time.sleep(3); we rely on tenacity’s retry or asyncio.wait_for for timed conditional waits.

I also explicitly modeled the memory store interface as a Protocol, so the test suite doesn’t just cover the current implementation—if we swap out the vector database or caching layer later, the exact same test cases can run with zero changes.

Core Implementation

1. Defining a Testable Memory Store Abstraction

This code answers the question “what interface do we actually test?” by abstracting the production‑level memory storage into a MemoryStoreProtocol. All subsequent tests depend only on this protocol.

from typing import Protocol, List, Dict, Any, runtime_checkable

@runtime_checkable
class MemoryStoreProtocol(Protocol):
    """大模型记忆存储必须实现的接口"""
    async def append_message(self, user_id: str, role: str, content: str) -> None:
        """追加一条对话消息到短期记忆"""
        ...

    async def get_history(self, user_id: str, limit: int = 10) -> List[Dict[str, Any]]:
        """按时间顺序返回最近的消息列表"""
        ...

    async def compact(self, user_id: str) -> None:
        """触发记忆压缩：生成摘要并清理原始短期消息"""
        ...

    async def consistent_state(self, user_id: str) -> bool:
        """检查所有后端数据是否一致（用于测试断言）"""
        ...

2. A Simplified Implementation with Race Conditions

To make the tests meaningful, I deliberately wrote a “buggy” implementation called FragileMemoryStore—it uses a Python list in place of Redis, a dict in place of the vector store, and randomly await asyncio.sleep(0) to simulate async delays. This code is the target we’re shooting at.

import asyncio
import random
from dataclasses import dataclass, field

@dataclass
class FragileMemoryStore:
    # 模拟三层存储
    short_term: Dict[str, List[dict]] = field(default_factory=dict)   # Redis List
    vector: Dict[str, List[str]] = field(default_factory=dict)       # 向量库
    raw_pg: Dict[str, List[dict]] = field(default_factory=dict)      # PG 原始表

    async def append_message(self, user_id: str, role: str, content: str) -> None:
        msg = {"role": role, "content": content}
        # 1. 先写短期缓存（同步写）
        self.short_term.setdefault(user_id, []).append(msg)
        # 2. 写 PG 原始表（同步写）
        self.raw_pg.setdefault(user_id, []).append(msg)
        # 3. 推向量库是异步的，模拟延迟
        await asyncio.sleep(random.uniform(0, 0.02))  # 随机窗口
        self.vector.setdefault(user_id,

LLM Memory Consistency Testing: 3 Pitfalls with Playwright + pytest (And 8 Hours of Debugging)

BAOFUFAN — Sat, 18 Jul 2026 01:04:27 +0000

At 1 AM, our test team's chat (Feishu) exploded: "The user just said they're vegetarian, and the AI immediately recommended braised pork—where is the memory stored? This is a critical production bug!" Staring at the flood of messages, I realized end-to-end memory validation was still a manual wasteland: before every release, QA had to simulate conversations by hand. One wrong click meant starting over, production data was off-limits, and the results were never truly trustworthy. That night I decided: I had to build a fully automated, closed-loop test for memory storage consistency using Playwright + pytest. After stepping into three deep pitfalls, the entire regression test went from 20 minutes down to 3 minutes, and no one ever had to stare at a chat window hunting for bugs again.

Why memory storage deceives so easily

Memory in large models typically relies on a backend memory service combined with front-end session caching. A typical scenario:

User inputs "My name is Wang Xiaoming" → the backend calls the memory storage API, writing user_name: Wang Xiaoming
User asks "What’s my name?" → the chat API retrieves from the memory service and returns "Wang Xiaoming"

Problems fall into three categories:

Write loss: The API was called but the memory didn’t actually persist in the database;
Retrieval error: The data exists but the retrieval picks up the wrong key or gets overwritten;
Front-end session cache interference: localStorage / sessionStorage retains an old memory and doesn't read from the backend.

Manual testing can only sample one or two conversations and can't precisely verify cross-session persistence. Regular API unit tests can't cover front-end cache paths either. What we need is a browser-based end-to-end closed loop: simulate a real conversation to write memory, clear the session, then recreate a new session to query—fully automated and assertable.

Why Playwright + pytest instead of Selenium

When choosing tools, I had three paths:

Pure API testing (requests / httpx): Cannot test front-end caching, and many memory APIs require complex authentication. Bypassing the front-end loses session context.
Selenium + pytest: It works, but element wait mechanisms are primitive. LLM responses are often streamed and rendered progressively; Selenium's explicit waits can easily be defeated by dynamic DOM.
Playwright + pytest: Playwright has native support for auto-waiting, network interception, and multi-context isolation. pytest’s fixture ecosystem can manage browser, page, and context directly, making it a natural fit for the "create new session → write → destroy → new session → query" workflow.

The architecture is simple: each test class represents a memory dimension (user name, preferences, dietary restrictions, etc.), and test methods follow a four-step choreography: write_memory → teardown_session → new_session → verify_memory. pytest fixtures create and destroy browser contexts, ensuring complete isolation. Meanwhile, conversation templates are data-driven; a single JSON file can add or remove test cases.

Core implementation

1. Browser management and session isolation (fixture layer)

This piece solves the problem of "a clean browser context per test to prevent memory cross-contamination."

# conftest.py
import pytest
from playwright.sync_api import sync_playwright

@pytest.fixture(scope="function")   # scope="function" 保证测试间绝对隔离
def browser_context():
    with sync_playwright() as p:
        browser = p.chromium.launch(headless=True)   # CI 环境必须 headless
        context = browser.new_context(
            storage_state=None,      # 无任何缓存
            locale="zh-CN"
        )
        page = context.new_page()
        yield page, context
        context.close()
        browser.close()

Why it matters: If you use module or session scope, the localStorage and cookies written by the previous test case will bleed into the next one, causing memory assertions to falsely pass—they’re actually using data from the old session. This cost me a full 3 hours to track down.

2. Chat helper utility (encapsulating reusable interaction logic)

This part solves the problem of "how to determine a complete response under streaming output."

# chat_helpers.py
from playwright.sync_api import Page, TimeoutError as PlaywrightTimeout

def send_message(page: Page, text: str):
    """向聊天输入框发送消息并回车"""
    input_el = page.locator("textarea[data-testid='chat-input']")
    input_el.fill(text)
    page.locator("button[data-testid='send-btn']").click()

def wait_for_reply_contains(page: Page, keyword: str, timeout: int = 15000) -> bool:
    """
    大模型回复是流式渲染，不能简单 wait_for_selector，
    这里轮询最新的 assistant 气泡文本，直到包含 keyword 或超时。
    """
    try:
        page.wait_for_function(
            f"""
            () => {{
                const bubbles = document.querySelectorAll('[data-role="assistant"]');
                if (!bubbles.length) return false;
                const latest = bubbles[bubbles.length - 1].innerText;
                return latest.includes("{keyword}");
            }}
            """,
            timeout=timeout
        )
        return True
    except PlaywrightTimeout:
        return False

Pitfall alert: If you use page.wait_for_selector("text=keyword") directly, streaming output can easily lead to partial matches and flaky test failures. Always poll the latest bubble content until the expected keyword appears.

Debugging LLM Conversation Memory: A 3-Day Race Condition Nightmare

BAOFUFAN — Fri, 17 Jul 2026 12:04:37 +0000

Here’s what happened: We delivered an AI chat app with “memory” to a client. Every conversation turn was stored so users could pick up right where they left off. The product manager confidently said “just like ChatGPT”, and QA manually tested hundreds of rounds without issues. Then, in the second week after launch, a user complained, “The AI remembered my girlfriend’s name as my ex’s” — conversation memories were getting mixed up.

My first instinct was a database query bug. I combed through the code for an entire day — nothing. It wasn’t until I wrote an automated script with Playwright and ran a stress test in the middle of the night that I finally caught the root cause: there was an inconsistency window between the frontend’s optimistic update and the backend’s asynchronous persistence. When multiple tabs were chatting at the same time, the order of memory storage got overwritten. And manual testing simply couldn’t reproduce it reliably — human hands aren’t fast enough.

Below I’ll share the whole debugging journey, the automated verification strategy, and how we eventually fixed it. If you’re building LLM-powered apps, you’ll hit this pitfall sooner or later.

Breaking Down the Problem

Our “conversation memory” was designed like this: for every user message, the backend appends the message to a conversation history list, calls the LLM to generate a reply, and then serializes the full conversation history into Redis (key = user:{uid}:session:{sid}:messages). On the frontend, after sending a message, the new message is instantly inserted into the on‑page chat list (optimistic update), and the list is refreshed again once the backend responds.

Theoretically clear, right? But two things went wrong:

The Redis write was not transactional — We used SET key value to overwrite the entire JSON list. If a user sends two messages in quick succession, the LLM call for the second message might return first and SET first; then when the response for the first message comes back and SETs, it overwrites the second one.
Multiple tabs shared the same session_id — When a user opens two tabs, both use the same session ID. Both pages send messages, and the backend can’t distinguish which request came from which tab. This jumbles the conversation order, and the memory gets “cross‑wired”.

Conventional locking schemes are extremely awkward in this LLM scenario: LLM calls take 2–5 seconds, and you can’t just make users stare at a distributed lock spinner. What’s worse, the frontend optimistic update hides the problem even more — the UI looks perfectly normal, but as soon as you refresh you see the memory is scrambled. Manual testing could never catch this, because no one is fast enough to race-typing on two computers simultaneously.

I decided to write a deterministic reproduction script with Playwright — simulating concurrent multi‑tab chats and automatically verifying whether the memory stored in Redis matches what the page actually shows.

Designing the Verification Strategy

Why not just write backend unit tests? Because the bug is a cross‑frontend‑backend‑cache timing issue. Mocking the browser would mask the real race‑condition window. We needed to actually launch Chromium, let two pages operate on the same session concurrently, and then inspect Redis’s actual storage after each step.

Tech stack:

Playwright (Python): Simulates the browser, supports concurrent contexts (multi‑page) natively, and allows precise timing control.
pytest: Orchestrates the test cases, with the pytest-playwright plugin managing the browser lifecycle.
redis-py: Reads Redis directly inside the test to compare the stored conversation against the expected order.
Not Selenium: Playwright’s page.wait_for_selector and network waiting mechanisms are more reliable, and its built‑in concurrent contexts make simulating multiple tabs natural.

Architecture: the test script launches two independent browser contexts (simulating two tabs) logged in with the same user and session. Then “Tab A” sends message m1, and without waiting for the reply, “Tab B” immediately sends message m2. At four critical time points — after sending m1, after m1’s reply arrives, after sending m2, and after m2’s reply arrives — we read the memory snapshot from Redis. The final assertion: the message list in Redis must neither lose any message nor scramble the order (m1 must appear before m2).

Core Implementation

1. Simulating concurrent chats and capturing snapshots

This snippet demonstrates how to manufacture a race condition with deterministic timing using Playwright while grabbing Redis snapshots. We use two async tasks to simulate the two tabs, precisely coordinating the sending moments via asyncio.Event.

import asyncio
import json
import redis
from playwright.async_api import async_playwright

REDIS_CLIENT = redis.Redis(host='localhost', port=6379, decode_responses=True)

def get_memory_snapshot(user_id: str, session_id: str) -> list:
    """从 Redis 读出当前存储的对话记忆（已解析的消息列表）"""
    key = f"user:{user_id}:session:{session_id}:messages"
    raw = REDIS_CLIENT.get(key)
    return json.loads(raw) if raw else []

async def send_message(page, text: str):
    """发送一条消息并等待 AI 回复出现在页面上"""
    await page.fill('textarea[placeholder="输入消息"]', text)
    await page.click('button[type="submit"]')
    # 等待 AI 回复的容器出现且包含文本，避免拿到空壳
    await page.wait_for_selector('.message.assistant:last-of-type :text-is("")', state='detached', timeout=15000)
    await page.wait_for_selector('.message.assistant:last-of-type', state='visible')

async def concurrent_chat_test():
    async with async_playwright() as p:
        browser = await p.chromium.launch(headless=True)
        # 创建两个独立 context，模拟两个标签页，共享同一 session
        context = await browser.new_context()
        page_a = await context.new_page()
        page_b = await context.new_page()

        USER_ID = "test_user_01"
        SESSION_ID = "sess_concurrent_01"
        # 省略登录逻辑，直接注入 localStorage/jwt
        await page_a.goto(f"https://chat.example.com/chat?session=

3 Years with Redis — I Never Once Verified Cache Consistency

BAOFUFAN — Fri, 17 Jul 2026 01:04:45 +0000

At 2 a.m., an alert call yanked me out of a dream — users were complaining that they were seeing other people’s orders on their own homepage. My first instinct was an authentication failure, but the logs showed auth was fine. The real cause gave me chills: the user data in Redis hadn’t been deleted after an update, and the frontend was rendering stale cache left behind by a previous request. After fixing the bug, I stared at the screen for ten minutes, thinking: this logic has been in production for nearly three years. Why was it never caught by automated tests? Because I had never verified whether the cache and database were actually consistent.

If you also think “cache inconsistency” is some dark art that only appears in other teams’ postmortems, I urge you to spend five minutes on this article. What we’ll do is straightforward: build a reusable Redis cache consistency validation test suite with Pytest, turning “consistency as I assume it” into “consistency as proved by the machine.”

What does “consistent” even mean?

Let’s break it down. Most of us follow a similar caching pattern: on reads, hit Redis first; if it’s a miss, query the database and write back to the cache. On writes, go straight to the database and then delete the corresponding key (the Cache-Aside pattern). The pattern itself is sound, but a single oversight in the code can cause disaster:

Failed cache deletion: After updating the DB, Redis gets disconnected due to a network hiccup. The delete throws an exception that gets swallowed, and old data stays in the cache forever.
Concurrent read/write: Thread A experiences a cache miss, fetches the old value from the DB, but before it can write it back to Redis, Thread B has already completed an update + cache deletion. Thread A then leisurely writes the old value with set, and the cache is now dirty.
Cache structure mismatch: A field type changes in the DB, but the cache retains the old format due to serialization quirks. When you read and parse it, things explode.

These scenarios can’t be tested manually — you can’t coordinate two hands to trigger a cache miss exactly before a deletion. Unit tests usually mock Redis; coverage looks great, but those mocked return values bear no resemblance to real Redis network behavior, serialization, or connection pool state. They don’t validate the “cache itself.”

So we need an automated test suite that works with a real Redis (or a high-fidelity simulation), locks down these abnormal timing sequences, and runs on every push. That’s exactly what we’ll do with Pytest.

Tooling: why fakeredis, not a real instance

To test Redis behavior, we have two paths:

Real Redis: spin up a local redis-server, or connect to a testing environment instance.
fakeredis: a pure Python in-memory implementation that adapts the redis-py API, with high command coverage.

I chose fakeredis not to save resources, but for test isolation and determinism. With a real Redis and parallel test execution (e.g., pytest-xdist), you must rigorously ensure unique key names, database index isolation, and flushing after each test. Miss one step and you’ll get flaky false positives. fakeredis gives you function-scoped in-memory instances — one instance per test. It can even run alongside Python 3.8 through 3.12 in CI with zero overhead. The trade-off is that it doesn’t support a few commands (like Redis 6 ACLs or some Stream commands), but the get/set/delete trio for Cache-Aside is more than enough.

If your team insists on the “real thing,” you can create a conditional fixture that switches between the two implementations using pytest.mark.parametrize or custom markers — I’ll provide a template for that later.

Core implementation: caging the concurrent timing

Let’s first design an extremely simplified but realistic service: a UserCacheService that reads user info by user_id and caches it in Redis. We’ll fake the database with an in-memory dict, so we can freely mutate data in tests. To reproduce the failure, I’ll deliberately make the cache deletion silently fail (simulating a network timeout) without raising an exception, so our tests can catch the bug.

1. The code under test: a buggy cache service

This code is the “suspect” from our incident — after updating the DB, it calls _delete_cache, but that call can silently fail, and update_user doesn’t check the return value at all.

# user_service.py
import json
from typing import Optional, Dict
from redis import Redis

class UserCacheService:
    def __init__(self, redis_client: Redis, db: dict):
        self.redis = redis_client
        self.db = db          # 假装是数据库连接
        self.ttl = 300

    def get_user(self, user_id: str) -> Optional[Dict]:
        cache_key = f"user:{user_id}"
        cached = self.redis.get(cache_key)
        if cached:
            return json.loads(cached)

        # 缓存未命中，穿透到 DB
        data = self.db.get(user_id)
        if data:
            # 回写缓存，序列化一下
            self.redis.setex(cache_key, self.ttl, json.dumps(data))
        return data

    def update_user(self, user_id: str, data: Dict) -> None:
        # 先更新 DB
        self.db[user_id] = data
        # 然后删缓存——注意这里没有做异常处理！
        try:
            self._delete_cache(user_id)
        except Exception:
            # 这里是 bug 的根源：吞掉异常，认为删缓存成功
            pass

    def _delete_cache(self, user_id: str) -> None:
        # 模拟可能失败的操作，真实场景可能是网络超时
        self.redis.delete(f"user:{user_id}")

2. Pytest fixture: a clean world for every test

We’ll build fixtures with fakeredis to make Redis and the DB function-scoped, ensuring each test starts from scratch without cross-contamination.

# conftest.py (或直接写在测试文件里)
import pytest
from fakeredis import FakeRedis
from u

Automating AI Memory Consistency Testing with pytest and Redis: How I Stopped Daily Manual Regression

BAOFUFAN — Thu, 16 Jul 2026 12:04:26 +0000

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.)