Redis Protocol Reference

Interface declaration:

cache = service("redis",
    interface("main", "redis", 6379),
    image = "redis:7",
    healthcheck = ready(timeout = "30s"),
)

A password-protected server declares it in env= as REDIS_PASSWORD (plus REDIS_USER for a Redis 6+ ACL user); steps and the healthcheck then send AUTH before anything else. Added in v0.16.1 — before that, every command against a server started with --requirepass came back NOAUTH, and since that lands in the step result rather than as a connection error, only a spec that checks r.ok would have seen it.

Methods

get(key="")

Get the value of a key.

resp = cache.main.get(key="user:1")
# resp.data = {"value": "alice"}   (or {"value": None} if key doesn't exist)

set(key="", value="")

Set a key-value pair.

cache.main.set(key="user:1", value="alice")
cache.main.set(key="config:timeout", value="5000")

del(key="")

Delete a key.

cache.main.del(key="user:1")

keys(pattern="*")

List keys matching a glob pattern.

resp = cache.main.keys(pattern="user:*")
# resp.data = ["user:1", "user:2", "user:3"]

resp = cache.main.keys(pattern="*")
# resp.data = ["user:1", "config:timeout", ...]

ping()

Check connectivity.

resp = cache.main.ping()
# resp.data = {"value": "PONG"}

incr(key="")

Increment an integer key.

cache.main.set(key="counter", value="0")
cache.main.incr(key="counter")
resp = cache.main.get(key="counter")
# resp.data = {"value": "1"}

lpush(key="", value="") / rpush(key="", value="")

Push to the left/right of a list.

cache.main.lpush(key="queue", value="first")
cache.main.rpush(key="queue", value="last")

lrange(key="", start="0", stop="-1")

Get a range of elements from a list.

resp = cache.main.lrange(key="queue", start="0", stop="-1")
# resp.data = ["first", "last"]

command(cmd="", args=[])

Execute any Redis command not covered by the methods above.

cache.main.command(cmd="EXPIRE", args=["user:1", "3600"])
cache.main.command(cmd="FLUSHALL")
cache.main.command(cmd="HSET", args=["user:1", "name", "alice", "email", "alice@example.com"])
resp = cache.main.command(cmd="HGETALL", args=["user:1"])

Authentication

Every method also accepts password= and user=, which override what the service declared:

With no password configured, no AUTH is sent and an unprotected server is untouched.

Response Object

A failing command sets .ok = False rather than raising, so an unchecked result is indistinguishable from a successful one:

r = cache.main.set(key="k", value="v")
assert_true(r.ok, "SET failed: %s" % r.error)

Fault Rules

error(command=, key=, message=)

Return a Redis error for matching commands.

set_fails = fault_assumption("set_fails",
    target = cache.main,
    rules = [error(command="SET", message="READONLY")],
)

key_error = fault_assumption("key_error",
    target = cache.main,
    rules = [error(key="order:*", message="WRONGTYPE")],
)

delay(command=, key=, delay=)

slow_reads = fault_assumption("slow_reads",
    target = cache.main,
    rules = [delay(command="GET", delay="500ms")],
)

drop(command=, key=)

drop_writes = fault_assumption("drop_writes",
    target = cache.main,
    rules = [drop(command="SET")],
)

Seed / Reset Patterns

def seed_redis():
    cache.main.command(cmd="FLUSHALL")
    cache.main.set(key="config:max_retries", value="3")
    cache.main.set(key="config:timeout_ms", value="5000")

cache = service("redis",
    interface("main", "redis", 6379),
    image = "redis:7",
    healthcheck = ready(timeout = "30s"),
    reuse = True,
    seed = seed_redis,  # FLUSHALL + seed on every reset
)

Tip: For Redis, seed and reset are often the same function — FLUSHALL clears everything, then you re-seed.

Data Integrity Patterns

No stale cache after failure

fault_scenario("no_stale_cache",
    scenario = create_order,
    faults = redis_down,
    expect = lambda r: (
        assert_true(r.status >= 500),
        assert_eq(len(cache.main.keys(pattern="order:*").data), 0,
            "no cached order data after Redis failure"),
    ),
)

Cache-DB consistency

fault_scenario("cache_consistent",
    scenario = update_user,
    faults = slow_cache,
    expect = lambda r: (
        assert_eq(r.status, 200),
        # DB has the updated value
        assert_eq(db.main.query(sql="SELECT name FROM users WHERE id=1").data[0]["name"], "updated"),
        # Cache also has the updated value (or is empty)
        assert_true(
            cache.main.get(key="user:1").data.get("value") in [None, "updated"],
            "cache must be consistent or empty"),
    ),
)