A cache can pass a test for each operation and still break when you combine
them. Take an LRU cache, short for least recently used. It holds at most
capacity entries. When it is full and a new key arrives, it evicts the entry
you used longest ago. With capacity two, this test must pass:
def test_get_refreshes_recency() -> None:
cache = LruCache(capacity=2)
cache.put("a", 1)
cache.put("b", 2)
cache.get("a") # "a" is now the most recently used
cache.put("c", 3) # the cache is full, so "b" goes
assert cache.get("b") is None
assert cache.get("a") == 1
If get does not mark "a" as used, the cache evicts "a" instead. You wrote
this test because you thought of this sequence. Hypothesis can generate
sequences like it for you.
Compare the cache with a model
A model is a second, much simpler version of the same behavior, one you can
trust by reading it. For an LRU cache, a plain dict works. A dict remembers
insertion order, so if you delete a key and insert it again, it moves to the
end. The first key is then always the least recently used.
Hypothesis's RuleBasedStateMachine runs the cache and the model side by side.
Each method marked @rule is a rule, an operation Hypothesis may pick at
each step, with a strategy for each argument. Hypothesis runs random sequences
of rules, and each rule checks the cache against the model.
Save this as test_lru.py, and add the test above to the same file. It
includes a small cache built on OrderedDict. In your project, import your
real cache instead.
from collections import OrderedDict
from hypothesis import strategies as st
from hypothesis.stateful import RuleBasedStateMachine, invariant, rule
CAPACITY = 3
KEYS = st.sampled_from("abcd")
class LruCache:
def __init__(self, capacity: int) -> None:
self.capacity = capacity
self.entries: OrderedDict[str, int] = OrderedDict()
def put(self, key: str, value: int) -> None:
self.entries[key] = value
self.entries.move_to_end(key)
if len(self.entries) > self.capacity:
self.entries.popitem(last=False)
def get(self, key: str) -> int | None:
if key not in self.entries:
return None
self.entries.move_to_end(key)
return self.entries[key]
def keys_in_lru_order(self) -> list[str]:
return list(self.entries)
class LruCacheMachine(RuleBasedStateMachine):
def __init__(self) -> None:
super().__init__()
self.cache = LruCache(CAPACITY)
self.model: dict[str, int] = {}
@rule(key=KEYS, value=st.integers(0, 5))
def put(self, key: str, value: int) -> None:
self.cache.put(key, value)
self.model.pop(key, None)
self.model[key] = value
if len(self.model) > CAPACITY:
del self.model[next(iter(self.model))]
@rule(key=KEYS)
def get(self, key: str) -> None:
assert self.cache.get(key) == self.model.get(key)
if key in self.model:
self.model[key] = self.model.pop(key)
@invariant()
def same_order(self) -> None:
assert self.cache.keys_in_lru_order() == list(self.model)
TestLruCache = LruCacheMachine.TestCase
Run pytest test_lru.py. The last line turns the machine into a test case
that pytest finds. Each sequence starts from a new machine.
@invariant() marks an invariant, a check that runs after every step. A
rule checks the result of its own operation, as get does with the returned
value. An invariant checks what must be true after any step, such as the order
of the keys.
Read the shortest failing sequence
Break the cache on purpose by deleting self.entries.move_to_end(key) from
get. Hypothesis finds a failing sequence and then shrinks it. It removes
steps and simplifies arguments for as long as the test still fails. The report
starts with a Failing test case: line, then lists the steps as code you can
paste into a regular test:
Failing test case: <state machine>
state = LruCacheMachine()
state.same_order()
state.put(key='a', value=0)
state.same_order()
state.put(key='b', value=0)
state.same_order()
state.get(key='a')
state.same_order()
state.teardown()
Three steps are enough. Reading "a" should move it to the end, and the
invariant fails because the cache did not move it. Each state.same_order()
line is the invariant check. Hypothesis runs it once at the start and again
after every step. On some runs the report uses the two keys the other way
round.
The exercise after this article builds a small, exact version of this search. It tries every sequence of length one, then two, and so on, so the first failure it finds is a shortest one.
Use a bundle for values you cannot generate
You choose the cache keys, so a strategy can generate them. Some values exist
only after your code creates them, such as a session token from log_in().
A bundle is a named pool of such values. A rule with target= adds its
return value to the bundle, and a later rule can take an argument from it:
import secrets
from hypothesis.stateful import Bundle, RuleBasedStateMachine, rule
class SessionStore:
def __init__(self) -> None:
self._open: set[str] = set()
def log_in(self) -> str:
token = secrets.token_hex(8)
self._open.add(token)
return token
def log_out(self, token: str) -> None:
self._open.discard(token)
def is_valid(self, token: str) -> bool:
return token in self._open
class SessionMachine(RuleBasedStateMachine):
tokens = Bundle("tokens")
def __init__(self) -> None:
super().__init__()
self.store = SessionStore()
self.model: set[str] = set()
@rule(target=tokens)
def log_in(self) -> str:
token = self.store.log_in()
self.model.add(token)
return token
@rule(token=tokens)
def log_out(self, token: str) -> None:
self.store.log_out(token)
self.model.discard(token)
@rule(token=tokens)
def check(self, token: str) -> None:
assert self.store.is_valid(token) == (token in self.model)
TestSessions = SessionMachine.TestCase
The bundle keeps every token a rule added to it, including ones you have
logged out. So check asks the model whether a token should still be valid.
Common mistakes
A model that copies the code. If the model repeats the cache's logic, it also repeats its bugs, and the two always agree. Write the model from the rules the cache must follow, with the simplest data structure that works.
A precondition that hides a promise. @precondition(lambda self: self.model)
above a @rule runs that rule only when the model is not empty. Use it for
operations that make no sense in some states. If pop() on an empty queue must
raise IndexError, write a rule that checks for that error instead of skipping
the case.
Missing operations. The machine tests only the operations it has rules for.
If your cache also has delete() or resize(), give each one a rule.
Check your understanding
Your machine draws keys from "ab", the capacity is three, and 1,000 runs pass. Is eviction tested?
Show answer
Hide answer
No. With two keys the cache never holds more than two entries, so it never fills up and never evicts. The eviction code never runs, so it cannot fail. Use at least one more key than the capacity, as the example does with four keys and capacity three.
Going further
A state machine runs one step at a time, so it does not test what happens when two threads use the cache at once.