Redis — Testing Setup & Isolation¶

uv add redis
uv add --dev pytest pytest-xdist pytest-asyncio "fakeredis[lua]" "testcontainers[redis]" time-machine

# tests/conftest.py
import os
from urllib.parse import urlparse

import fakeredis
import pytest
import redis


# --- fast: in-process fake ------------------------------------------------
@pytest.fixture
def fake_r():
    r = fakeredis.FakeRedis(decode_responses=True)   # new empty server for every test
    yield r
    r.close()


# --- real Redis: shared server from REDIS_URL, or one container per session --
@pytest.fixture(scope="session")
def redis_base_url():
    if url := os.getenv("REDIS_URL"):                # CI service / docker compose
        yield url
        return
    from testcontainers.community.redis import RedisContainer

    with RedisContainer("redis:8") as container:
        host = container.get_container_host_ip()
        port = container.get_exposed_port(6379)
        yield f"redis://{host}:{port}"


@pytest.fixture(scope="session")
def redis_db() -> int:
    # xdist sets PYTEST_XDIST_WORKER=gw0, gw1, ...; db 0 stays for manual work
    worker = os.getenv("PYTEST_XDIST_WORKER", "gw0")
    return int(worker.removeprefix("gw")) + 1


@pytest.fixture
def real_r(redis_base_url, redis_db):
    host = urlparse(redis_base_url).hostname
    if host not in {"localhost", "127.0.0.1", "::1"} and os.getenv("REDIS_TEST_ALLOW_FLUSH") != "1":
        pytest.fail(f"refusing to FLUSHDB on {host}; set REDIS_TEST_ALLOW_FLUSH=1 for a dedicated test server")
    r = redis.Redis.from_url(redis_base_url, db=redis_db, decode_responses=True)
    r.flushdb()                                      # this db only, never FLUSHALL
    yield r
    r.flushdb()
    r.close()


@pytest.fixture(params=["fake", pytest.param("real", marks=pytest.mark.integration)])
def r(request):
    """Run a test against both backends: pytest -m 'not integration' for the fast set."""
    return request.getfixturevalue(f"{request.param}_r")

# pyproject.toml
[tool.pytest.ini_options]
pythonpath = ["."]
markers = ["integration: needs a real Redis (Docker or REDIS_URL)"]

uv run pytest -m "not integration"                  # fakeredis only, no Docker
uv run pytest                                       # both; starts redis:8 via testcontainers
REDIS_URL=redis://localhost:6379 uv run pytest      # use an already running Redis

# app/main.py
from functools import lru_cache
from typing import Annotated

import redis
from fastapi import Depends, FastAPI


@lru_cache
def get_redis() -> redis.Redis:
    return redis.Redis.from_url("redis://localhost:6379/0", decode_responses=True)


app = FastAPI()


@app.post("/login-attempts/{user}")
def login_attempt(user: str, r: Annotated[redis.Redis, Depends(get_redis)]) -> dict:
    attempts = r.incr(f"myapp:login:{user}")
    r.expire(f"myapp:login:{user}", 900, nx=True)
    return {"attempts": attempts, "locked": attempts > 5}

# tests/test_api.py
import fakeredis
import pytest
from fastapi.testclient import TestClient

from app.main import app, get_redis


@pytest.fixture
def client():
    fake = fakeredis.FakeRedis(decode_responses=True)
    app.dependency_overrides[get_redis] = lambda: fake
    with TestClient(app) as c:
        yield c
    app.dependency_overrides.clear()


def test_sixth_attempt_locks_the_account(client):
    for _ in range(5):
        assert client.post("/login-attempts/ann").json()["locked"] is False
    assert client.post("/login-attempts/ann").json() == {"attempts": 6, "locked": True}

# tests/test_prefix.py
import os
import uuid

import pytest
import redis


@pytest.fixture(scope="session")
def shared_r():
    url = os.getenv("SHARED_REDIS_URL") or pytest.skip("SHARED_REDIS_URL not set")
    r = redis.Redis.from_url(url, decode_responses=True)   # no FLUSHDB on this server
    yield r
    r.close()


@pytest.fixture
def key_prefix(shared_r):
    prefix = f"test:{uuid.uuid4().hex[:8]}:"
    yield prefix
    for key in shared_r.scan_iter(match=f"{prefix}*", count=1000):   # delete only what this test created
        shared_r.unlink(key)


def test_prefix_cleanup(shared_r, key_prefix):
    shared_r.set(f"{key_prefix}myapp:user:1", "x", ex=3600)   # TTL: a safety net if teardown never runs
    assert shared_r.exists(f"{key_prefix}myapp:user:1") == 1

uv run pytest -n auto                                    # each worker: its own testcontainers Redis
REDIS_URL=redis://localhost:6379 uv run pytest -n 4      # one server: gw0 → db 1 … gw3 → db 4

# tests/test_async.py
import fakeredis
import pytest
import pytest_asyncio
import redis.asyncio as aioredis


async def get_or_set(r: aioredis.Redis, key: str, value: str) -> str:
    return await r.set(key, value, nx=True, get=True) or value   # SET NX GET: Redis 7.0+


@pytest_asyncio.fixture
async def async_r():
    r = fakeredis.FakeAsyncRedis(decode_responses=True)
    yield r
    await r.aclose()


@pytest.mark.asyncio
async def test_first_writer_wins(async_r):
    assert await get_or_set(async_r, "myapp:k", "a") == "a"
    assert await get_or_set(async_r, "myapp:k", "b") == "a"

# .github/workflows/tests.yml (fragment)
jobs:
  tests:
    runs-on: ubuntu-latest
    services:
      redis:
        image: redis:8.10                       # same major/minor as production
        ports: ["6379:6379"]
        options: >-
          --health-cmd "redis-cli ping"
          --health-interval 5s --health-timeout 3s --health-retries 10
    env:
      REDIS_URL: redis://localhost:6379
    steps:
      - uses: actions/checkout@v7
      - uses: astral-sh/setup-uv@v10
      - run: uv run pytest -n 4 --junitxml=report.xml