Property-Based Testing for Idempotent Handlers

Example-based tests verify the interleavings you thought of. The interleaving that breaks your handler is, by definition, one you did not. Property-based testing inverts the relationship: you specify the invariant, the framework searches for a schedule that violates it, and when it finds one it shrinks it to something readable. This runbook implements the approach outlined in testing & verification of idempotent systems.

Prerequisites. A containerised store — real Redis or PostgreSQL, never a mock — and a handler whose side effects you can count from outside.

Step 1 — State the invariants precisely

# The whole specification. Everything else is machinery to search for violations.
#
# INV-1  effect_count(key) <= 1, and == 1 whenever no failure was injected
# INV-2  all successful responses for a key are byte-identical to the first
Shrinking is what makes a failure actionable An original failing schedule of twelve waves with nine injected faults is reduced step by step until only two waves and one fault remain, still reproducing the failure. The reduced case is short enough to read, reason about and turn into a permanent regression test. original 12 waves, 9 faults shrinking still fails · still fails · still fails reported 2 waves, 1 fault — a bug report a human can act on

Both need care. <= 1 rather than == 1 because a fail-closed policy under an injected store outage correctly produces zero executions; encoding == 1 would make the test fail on correct behaviour. And “byte-identical” rather than “semantically equal”, because a client comparing JSON with == will notice a reordered key even if a human would not.

The property-test loop and what each stage contributes Four boxes in a cycle. The generator emits a schedule of deliveries, delays, concurrency and injected faults. The runner executes it against a containerised store. The oracle checks effect count at most one and identical responses. On a violation the shrinker reduces the schedule, re-running until it finds the smallest failing case, then reports it. On success the loop returns to the generator for the next example. generator n deliveries, gaps, concurrency, faults seeded, reproducible runner real Redis / Postgres in a container fresh key namespace oracle effects <= 1 responses identical nothing else pass → next example violation shrinker drop deliveries, drop faults, re-run until minimal report the 2-delivery, 1-fault case a human can read

Step 2 — Generate delivery schedules

from hypothesis import given, settings, strategies as st

Delivery = st.fixed_dictionaries({
    "gap_ms":     st.integers(min_value=0, max_value=50),
    "concurrent": st.integers(min_value=1, max_value=8),   # how many at once
    "fault":      st.sampled_from([None, "store_timeout", "handler_5xx",
                                   "kill_before_complete", "network_reset"]),
})

Schedule = st.lists(Delivery, min_size=1, max_size=12)

Each generated element describes one wave: how long to wait, how many duplicates to release simultaneously, and whether to inject a fault. That vocabulary covers all three axes — sequential, concurrent and crash — from a single generator.

Step 3 — The oracle against a real store

@settings(max_examples=300, deadline=None, print_blob=True)
@given(schedule=Schedule, key=st.uuids().map(str))
def test_idempotency_invariants(schedule, key, client, db, faults):
    responses, any_fault = [], False

    for wave in schedule:
        if wave["fault"]:
            any_fault = True
            faults.arm(wave["fault"])
        with ThreadPoolExecutor(wave["concurrent"]) as pool:
            responses.extend(pool.map(
                lambda _: post(client, key, PAYLOAD), range(wave["concurrent"])))
        faults.disarm()
        time.sleep(wave["gap_ms"] / 1000)

    effects = db.scalar(
        "SELECT count(*) FROM charges WHERE idempotency_key = %s", (key,))
    successes = [r for r in responses if 200 <= r.status_code < 300]

    # INV-1
    assert effects <= 1, f"{effects} effects for one key; schedule={schedule}"
    if not any_fault:
        assert effects == 1, "no fault injected but nothing executed"

    # INV-2
    if successes:
        first = successes[0].content
        assert all(r.content == first for r in successes), "responses diverged"
# The fault injector: real failures at real seams, not mocked return values.
class Faults:
    def arm(self, kind: str) -> None:
        match kind:
            case "store_timeout":        self._toxi.add_toxic("timeout", stream="upstream")
            case "handler_5xx":          self._flags.set("charges.force_500", True)
            case "kill_before_complete": self._flags.set("charges.kill_after_effect", True)
            case "network_reset":        self._toxi.add_toxic("reset_peer")

Use a real proxy (Toxiproxy or equivalent) rather than patching the client library. A patched client tests your mock’s behaviour; a proxy tests what the socket actually does, which is where the interesting failures live.

Step 4 — A stateful model for multi-key sequences

Single-key properties miss cross-key interference — a store whose eviction, TTL sweeper or namespace scoping is wrong. Hypothesis’s RuleBasedStateMachine models the whole store.

from hypothesis.stateful import RuleBasedStateMachine, rule, invariant, Bundle

class IdempotencyStateMachine(RuleBasedStateMachine):
    keys = Bundle("keys")

    def __init__(self):
        super().__init__()
        self.model: dict[str, int] = {}      # key -> expected effect count

    @rule(target=keys, key=st.uuids().map(str))
    def new_key(self, key):
        self.model[key] = 0
        return key

    @rule(key=keys)
    def deliver(self, key):
        r = post(self.client, key, PAYLOAD)
        if 200 <= r.status_code < 300 and self.model[key] == 0:
            self.model[key] = 1
        return r

    @rule(key=keys, n=st.integers(min_value=2, max_value=16))
    def deliver_concurrently(self, key, n):
        with ThreadPoolExecutor(n) as pool:
            list(pool.map(lambda _: post(self.client, key, PAYLOAD), range(n)))
        if self.model[key] == 0:
            self.model[key] = 1

    @invariant()
    def effects_match_model(self):
        for key, expected in self.model.items():
            actual = self.db.scalar(
                "SELECT count(*) FROM charges WHERE idempotency_key=%s", (key,))
            assert actual == expected, f"{key}: model {expected}, actual {actual}"

TestIdempotencyStore = IdempotencyStateMachine.TestCase

The model is deliberately trivial — a dictionary of expected counts. Its value is that Hypothesis will interleave new_key, deliver and deliver_concurrently in orders nobody would write by hand, and the invariant catches any divergence.

Step 5 — Capture and replay counterexamples

# print_blob=True emits a reproducible blob on failure. Commit it as a regression test.
from hypothesis import reproduce_failure

@reproduce_failure("6.98.0", b"AXicY2BkYGBgZGRkYGBiYGBgZGBgYGBgYGBgAAAoAAO+")
def test_regression_two_concurrent_after_store_timeout(client, db, faults):
    """Shrunk from a 12-wave schedule: 1 timeout, then 2 concurrent deliveries."""
    ...
# Nightly: a large run that files an issue with the shrunk case attached.
HYPOTHESIS_PROFILE=nightly pytest tests/test_properties.py -q \
  --hypothesis-seed=random 2>&1 | tee property_run.log
if grep -q 'Falsifying example' property_run.log; then
  gh issue create --title "Idempotency property violation $(date -I)" \
    --body "$(sed -n '/Falsifying example/,/^$/p' property_run.log)" \
    --label bug,idempotency
fi
# Two profiles: fast on every commit, exhaustive nightly.
from hypothesis import settings, Phase
settings.register_profile("ci",      max_examples=50,   deadline=None)
settings.register_profile("nightly", max_examples=2000, deadline=None,
                          phases=[Phase.generate, Phase.target, Phase.shrink])

Verification and testing

Confirm the harness can actually detect a bug by breaking the implementation deliberately.

# Mutate the registration to a non-atomic check-then-write, then run the suite.
git stash && sed -i 's/SET NX PX/GET; SET PX/' src/store.py
pytest tests/test_properties.py -q
# expect: FAILED with a falsifying example showing 2 effects for one key
git checkout src/store.py && git stash pop

A property suite that passes against a knowingly broken implementation is not testing anything. Run this mutation check whenever the harness changes.

Confirm the store is real, not a fake.

pytest tests/test_properties.py -q -s 2>&1 | grep -i 'testcontainers\|redis:7\|postgres:16'
# expect: a container image line — its absence means a mock crept back in

Confirm shrinking is enabled. A failing run should report a schedule of one or two waves, not the original twelve.

Failure scenarios and debugging

Failure Scenario Remediation Steps Observability Hooks
Oracle asserts effects == 1 unconditionally, so correct fail-closed behaviour under an injected outage reports as a failure Assert effects <= 1 always and == 1 only when no fault was injected. Track the injected-fault flag through the schedule. Falsifying examples that all contain a store_timeout fault — a signature of an over-strict oracle
Property tests run against an in-memory fake, so they pass while the real store’s atomicity is broken Use Testcontainers and assert the image name in the run log. Delete the fake so it cannot be reintroduced. The container-image grep above, run as a CI step rather than by hand
Shrinking disabled by a deadline setting, so failures report unreadable twelve-wave schedules Set deadline=None for property tests; a wall-clock deadline aborts shrinking. Enable print_blob=True. Falsifying examples longer than three waves consistently means shrinking is not running
Keys reused across examples, so state from one example causes a spurious failure in the next Generate a fresh UUID key per example and namespace every store key by test name. Never flush the store between examples — that hides real leakage. Failures that disappear on re-run with the same seed indicate cross-example leakage
A 2,000-example run added to the pull-request pipeline, and someone marks it skip within a month Split into a 50-example CI profile and a 2,000-example nightly profile that files issues automatically. Suite duration in CI; a property job above two minutes will be disabled eventually
Prove the harness can fail The atomic registration is deliberately mutated into a non-atomic read-then-write, the property suite is run, and it must report a falsifying example. A suite that stays green against a knowingly broken implementation is testing nothing, so this check belongs in the pipeline whenever the harness itself changes. mutate the registration SET NX → GET; SET run the suite it MUST report a failure green here means the suite is decorative

SRE / observability checklist

  1. property_falsifying_examples_total — Counter from the nightly job. Any value is a real defect; there is no such thing as a flaky property failure with a fixed seed.
  2. property_examples_executed — Gauge per run. A drop means examples are being rejected by filters and coverage has silently narrowed.
  3. property_shrink_steps — Gauge of shrinking iterations. Zero on a failure means shrinking is disabled and the report is unusable.
  4. property_suite_duration_seconds — Histogram. Above 120 s in the commit path predicts the suite being disabled.
  5. mutation_detection_rate — Share of deliberately injected mutations the suite catches. Below 100% on the atomicity mutation means the harness has a blind spot.
  6. Seed and blob recorded per run — So every failure is replayable exactly, which is what turns a property failure into a fixable bug rather than an argument.