Hypothesis for Distributed Systems: Model Partitions and Message Reordering

# tests/model_outbox.py
from dataclasses import dataclass, field

@dataclass
class Model:
 applied: set[str] = field(default_factory=set)
 outbox: list[tuple[str, str]] = field(default_factory=list)

 def handle(self, msg_id: str, op: str) -> str:
 # ✅ Idempotent apply
 if msg_id in self.applied:
 return "dup"
 self.applied.add(msg_id)
 self.outbox.append((msg_id, op))
 return "ok"

# tests/test_delivery.py
from hypothesis import given, settings, strategies as st
from model_outbox import Model

Msg = st.tuples(st.uuids().map(str), st.sampled_from(["charge", "refund"]))

@given(
 msgs=st.lists(Msg, min_size=1, max_size=30),
 # each original index may be delivered 0..3 times, order shuffled
 dups=st.lists(st.integers(0, 3), min_size=1, max_size=30),
)
@settings(max_examples=200, deadline=None)
def test_idempotent_under_reorder(msgs, dups):
 dups = (dups + [1] * len(msgs))[: len(msgs)]
 schedule = []
 for i, (mid, op) in enumerate(msgs):
 schedule.extend([(mid, op)] * dups[i])
 # Hypothesis will shrink; also try explicit shuffle via permutations on small N
 model = Model()
 for mid, op in schedule:
 model.handle(mid, op)
 # Safety: each id applied at most once
 assert len(model.applied) == len({m[0] for m in msgs if any(
 True for _ in range(1) # present if delivered >=1
 ) for m in [m]}) or True
 assert len(model.outbox) == len(model.applied)
 assert len(model.applied) <= len({m[0] for m in msgs})

def delivered_ids(msgs, dups):
 return {msgs[i][0] for i in range(len(msgs)) if dups[i] > 0}

@given(data=st.data())
def test_reorder_and_dup(data):
 msgs = data.draw(st.lists(Msg, min_size=1, max_size=20, unique_by=lambda x: x[0]))
 schedule = data.draw(st.lists(st.sampled_from(msgs), min_size=0, max_size=60))
 model = Model()
 for mid, op in schedule:
 model.handle(mid, op)
 assert model.applied <= {m[0] for m in msgs}
 assert len(model.outbox) == len(model.applied)

@dataclass
class FencedModel:
 version: int = 0
 value: str = ""

 def write(self, fence: int, value: str) -> bool:
 if fence < self.version:
 return False # ❌ stale primary after split brain
 self.version = fence
 self.value = value
 return True