You have a CRUD REST service and a pile of example tests that each exercise one endpoint, yet production still throws 404s on resources that were just created, or returns a deleted record from a list endpoint. The defect lives in the interaction between requests — create, update, delete, re-read — not in any single handler. Hypothesis stateful testing generates those request sequences for you and shrinks any failure to a minimal reproduction. This guide maps REST verbs onto a RuleBasedStateMachine: endpoints become rules, created resources flow through Bundles, and business rules become invariants.
Prerequisites
- Python 3.9+,
hypothesis >= 6.0,pytest >= 7.0 - A web app exposing CRUD endpoints. The example uses FastAPI with
starlette.testclient.TestClient(pip install "fastapi>=0.110" httpx), but the pattern applies to any client. - Familiarity with the primitives in Stateful and Model-Based Testing —
@rule,@initialize,Bundle,@invariant,@precondition.
Solution
The mapping is mechanical once internalized: each verb is a transition, the set of live resource ids is a Bundle, and the server's contract is a set of invariants checked against an in-memory model.
# test_items_api_state.py
from fastapi import FastAPI, HTTPException
from fastapi.testclient import TestClient
from hypothesis import strategies as st, settings, HealthCheck
from hypothesis.stateful import (
RuleBasedStateMachine, Bundle, rule, initialize, invariant,
precondition, consumes,
)
# --- A minimal system under test -------------------------------------------
app = FastAPI()
_DB: dict[int, dict] = {}
_NEXT = {"id": 1}
@app.post("/items")
def create_item(body: dict):
item_id = _NEXT["id"]; _NEXT["id"] += 1
_DB[item_id] = {"id": item_id, "name": body["name"]}
return _DB[item_id]
@app.get("/items/{item_id}")
def read_item(item_id: int):
if item_id not in _DB:
raise HTTPException(status_code=404)
return _DB[item_id]
@app.put("/items/{item_id}")
def update_item(item_id: int, body: dict):
if item_id not in _DB:
raise HTTPException(status_code=404)
_DB[item_id]["name"] = body["name"]
return _DB[item_id]
@app.delete("/items/{item_id}")
def delete_item(item_id: int):
if item_id not in _DB:
raise HTTPException(status_code=404)
del _DB[item_id]
return {"deleted": item_id}
@app.get("/items")
def list_items():
return list(_DB.values())
# --- The state machine -----------------------------------------------------
names = st.text(min_size=1, max_size=12)
class ItemsAPIMachine(RuleBasedStateMachine):
items = Bundle("items") # server-issued ids of live resources
@initialize()
def setup(self):
_DB.clear(); _NEXT["id"] = 1 # reset server state per sequence
self.client = TestClient(app)
self.model: dict[int, str] = {} # id -> expected name
@rule(target=items, name=names)
def create(self, name: str):
resp = self.client.post("/items", json={"name": name})
assert resp.status_code == 200
item_id = resp.json()["id"] # use the REAL server id
self.model[item_id] = name
return item_id # push id into the `items` Bundle
@rule(item_id=items, name=names)
def update(self, item_id: int, name: str):
resp = self.client.put(f"/items/{item_id}", json={"name": name})
assert resp.status_code == 200
self.model[item_id] = name
@rule(item_id=items)
def read(self, item_id: int):
resp = self.client.get(f"/items/{item_id}")
assert resp.status_code == 200
assert resp.json()["name"] == self.model[item_id]
@rule(item_id=consumes(items)) # consumes => id never drawn again
def delete(self, item_id: int):
resp = self.client.delete(f"/items/{item_id}")
assert resp.status_code == 200
del self.model[item_id]
# Contract: a deleted resource must now 404.
assert self.client.get(f"/items/{item_id}").status_code == 404
@precondition(lambda self: True)
@invariant()
def list_count_matches_model(self):
# Business rule: the collection endpoint lists exactly the live items.
listed = self.client.get("/items").json()
assert len(listed) == len(self.model)
assert {i["id"] for i in listed} == set(self.model)
TestItemsAPI = ItemsAPIMachine.TestCase
TestItemsAPI.settings = settings(
max_examples=100,
stateful_step_count=40,
suppress_health_check=[HealthCheck.too_slow],
)
Why this works
The Bundle is what makes generated request sequences realistic: a read or delete only ever fires against an id the server actually issued from a prior create, so Hypothesis never wastes steps probing random 404s and instead explores the meaningful state space. Using consumes(items) on delete removes the id from the queue, so the deleted-then-read contract is enforced by construction rather than by chance. The in-memory model dict is the oracle — cheap, obviously-correct code that the real handlers must agree with after every transition, which the @invariant checks.
Edge cases and failure modes
- Shared server state leaks across sequences. Module-level
_DBmust be reset in@initialize; otherwise resources from a previous example pollute the next and invariants fail spuriously. Prefer a fresh app/database fixture per machine instance for real services. consumesvs plain Bundle reference. Using the bare Bundle (item_id=items) indeletewould leave the deleted id drawable, and a laterreadwould assert 200 against a 404 — a false failure. Alwaysconsumes()on terminal operations.- Non-deterministic ids. If the server issues UUIDs or relies on wall-clock ordering, capture the id from the response body (as above) rather than predicting it; never hardcode
item_id=1. - Slow real I/O. Hitting a live database for 40 steps across 100 examples is thousands of requests. Suppress
HealthCheck.too_slow, lowerstateful_step_count, or use an in-processTestClientas shown. See reducing Hypothesis test execution time. - Auth and pagination. Endpoints requiring tokens or returning paged lists need the invariant to follow pagination; comparing only the first page against a growing model will fail past the page size.
Frequently Asked Questions
How do I represent a POST that creates a resource in a Hypothesis state machine?
Model the POST as an @rule whose target is a Bundle, returning the created resource id from the response. Hypothesis pushes that id into the Bundle so later GET, PUT, and DELETE rules draw a real, server-issued id rather than fabricating one.
How do I model DELETE so the same id is not reused after removal?
Consume the id from the Bundle with consumes(resources) in the DELETE rule. consumes removes the value from the queue, so no subsequent rule will draw a deleted id, matching the server's contract that a deleted resource returns 404.
Where do business rules go in a REST state machine?
Encode them as @invariant methods that query the live API and compare against the in-memory model — for example asserting the list endpoint returns exactly the set of created-and-not-deleted ids after every step.
← Back to Stateful and Model-Based Testing