507 lines
18 KiB
Python
507 lines
18 KiB
Python
#!/usr/bin/env python3
|
|
"""test_crdt.py -- property + mutation tests for orcrdt.
|
|
|
|
Run: python3 test_crdt.py (exit 0 iff every test passes)
|
|
python3 test_crdt.py -v
|
|
|
|
Two kinds of evidence:
|
|
1. SIMULATION -- N replicas take random local ops, then gossip their serialized
|
|
snapshots in random order with duplicate deliveries. Every replica must end at
|
|
the exact join of all pre-gossip states (state AND semantic value).
|
|
2. MUTATION TEETH -- the same convergence check is pointed at deliberately broken
|
|
merge implementations; if it does not go red for them, the check is worthless
|
|
and this suite fails.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import random
|
|
import sys
|
|
import unittest
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
from orcrdt import CRDTError, LWWRegister, ORSet, PNCounter # noqa: E402
|
|
|
|
ELEMS = ["alpha", "beta", "gamma", "delta", "epsilon"]
|
|
SEEDS = range(16)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# simulation harness
|
|
# ---------------------------------------------------------------------------
|
|
def _family(kind):
|
|
"""Map a CRDT class (or a mutant subclass of one) to its base family."""
|
|
for base in (ORSet, PNCounter, LWWRegister):
|
|
if issubclass(kind, base):
|
|
return base
|
|
raise TypeError("unknown CRDT family: %r" % (kind,))
|
|
|
|
|
|
def _local_op(kind, rng, rep):
|
|
fam = _family(kind)
|
|
if fam is ORSet:
|
|
elem = rng.choice(ELEMS)
|
|
if rng.random() < 0.55:
|
|
rep.add(elem)
|
|
else:
|
|
rep.remove(elem)
|
|
elif fam is PNCounter:
|
|
d = rng.choice([1, 1, 1, 2, 3, -1, -2])
|
|
if d > 0:
|
|
rep.increment(d)
|
|
else:
|
|
rep.decrement(-d)
|
|
else:
|
|
ts = rep.stamp[0] + 1 + rng.randrange(3)
|
|
rep.set(rng.randrange(50), ts)
|
|
|
|
|
|
def make_replicas(kind, n, ops, seed):
|
|
rng = random.Random(seed)
|
|
reps = [kind("r%d" % i) for i in range(n)]
|
|
for _ in range(ops):
|
|
_local_op(kind, rng, rng.choice(reps))
|
|
return reps
|
|
|
|
|
|
def gossip(kind, reps, seed, rounds=4):
|
|
"""Random pairwise merges with duplicate deliveries, over serialized copies."""
|
|
rng = random.Random(seed * 7919 + 13)
|
|
n = len(reps)
|
|
live = [kind.loads(r.dumps()) for r in reps]
|
|
for _ in range(rounds * n * 3):
|
|
dst, src = rng.randrange(n), rng.randrange(n)
|
|
if dst == src:
|
|
continue
|
|
live[dst] = live[dst].merge(live[src])
|
|
return live
|
|
|
|
|
|
def check_convergence(kind, seed, n=3, ops=80, rounds=4):
|
|
"""True iff gossip converges to the join of the pre-gossip replica states."""
|
|
reps = make_replicas(kind, n, ops, seed)
|
|
snaps = [kind.loads(r.dumps()) for r in reps]
|
|
|
|
full = snaps[0]
|
|
for s in snaps[1:]:
|
|
full = full.merge(s)
|
|
|
|
live = gossip(kind, reps, seed, rounds)
|
|
state_ok = all(x.state_key() == full.state_key() for x in live)
|
|
value_ok = all(x.value_key() == full.value_key() for x in live)
|
|
join_ok = full.state_key() == full.merge(*snaps).state_key()
|
|
return state_ok and value_ok and join_ok
|
|
|
|
|
|
def check_convergence_detail(kind, seed, **kw):
|
|
reps = make_replicas(kind, kw.get("n", 3), kw.get("ops", 80), seed)
|
|
snaps = [kind.loads(r.dumps()) for r in reps]
|
|
full = snaps[0]
|
|
for s in snaps[1:]:
|
|
full = full.merge(s)
|
|
live = gossip(kind, reps, seed, kw.get("rounds", 4))
|
|
return {
|
|
"state_ok": all(x.state_key() == full.state_key() for x in live),
|
|
"value_ok": all(x.value_key() == full.value_key() for x in live),
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# deliberately broken merges (mutation testing of the harness above)
|
|
# ---------------------------------------------------------------------------
|
|
class ORSetDropsForeignTombstones(ORSet):
|
|
"""BUG: merge keeps only self's tombstone set -> removed elements resurrect."""
|
|
|
|
def merge(self, *others):
|
|
out = super().merge(*others)
|
|
out._removed = set(self._removed)
|
|
return out
|
|
|
|
|
|
class PNCounterDropsDecrements(PNCounter):
|
|
"""BUG: merge folds inc by max but drops the others' dec vectors."""
|
|
|
|
def merge(self, *others):
|
|
out = PNCounter(self.replica)
|
|
out.inc, out.dec = dict(self.inc), dict(self.dec)
|
|
for st in others:
|
|
for k, v in st.inc.items():
|
|
if v > out.inc.get(k, 0):
|
|
out.inc[k] = v
|
|
return out
|
|
|
|
|
|
class LWWLastArgumentWins(LWWRegister):
|
|
"""BUG: merge always takes the last argument -> order dependent."""
|
|
|
|
def merge(self, *others):
|
|
out = LWWRegister(self.replica)
|
|
out._value, out._stamp = self._value, self._stamp
|
|
for st in others:
|
|
out._value, out._stamp = st._value, st._stamp
|
|
return out
|
|
|
|
|
|
class ORSetNoOpMerge(ORSet):
|
|
"""BUG: merge ignores its argument (no propagation) -> not commutative."""
|
|
|
|
def merge(self, *others):
|
|
out = ORSet(self.replica)
|
|
out._adds = {e: set(t) for e, t in self._adds.items()}
|
|
out._removed = set(self._removed)
|
|
out._clock = self._clock
|
|
return out
|
|
|
|
|
|
class PNCounterSumMerge(PNCounter):
|
|
"""BUG: merge adds the vectors instead of taking pointwise max.
|
|
|
|
Commutative and associative, so naive gossip tests can miss it -- it fails
|
|
IDEMPOTENCE, and any duplicate delivery then inflates the value.
|
|
"""
|
|
|
|
def merge(self, *others):
|
|
out = PNCounter(self.replica)
|
|
out.inc, out.dec = dict(self.inc), dict(self.dec)
|
|
for st in others:
|
|
for k, v in st.inc.items():
|
|
out.inc[k] = out.inc.get(k, 0) + v
|
|
for k, v in st.dec.items():
|
|
out.dec[k] = out.dec.get(k, 0) + v
|
|
return out
|
|
|
|
|
|
def mutant_detected(mutant_cls, seeds=SEEDS):
|
|
"""True iff the convergence check goes red (or raises) for >=1 seed."""
|
|
hits = []
|
|
for seed in seeds:
|
|
try:
|
|
ok = check_convergence(mutant_cls, seed)
|
|
except CRDTError:
|
|
ok = False
|
|
if not ok:
|
|
hits.append(seed)
|
|
return hits
|
|
|
|
|
|
class RefORSet:
|
|
"""Independent naive add-wins OR-Set, written from the paper semantics.
|
|
|
|
Different layout on purpose: a flat set of ``(elem, replica, counter)`` tags
|
|
plus a tombstone set, so agreeing with ``orcrdt.ORSet`` is evidence and not
|
|
self-confirmation. Test-only; not exported by the library.
|
|
"""
|
|
|
|
def __init__(self, replica):
|
|
self.replica = replica
|
|
self.adds = set()
|
|
self.removed = set()
|
|
self.clock = 0
|
|
|
|
def add(self, elem):
|
|
self.clock += 1
|
|
self.adds.add((elem, self.replica, self.clock))
|
|
|
|
def remove(self, elem):
|
|
for tag in [t for t in self.adds if t[0] == elem]:
|
|
self.removed.add(tag)
|
|
|
|
def elements(self):
|
|
return frozenset(t[0] for t in self.adds if t not in self.removed)
|
|
|
|
def merge(self, *others):
|
|
out = RefORSet(self.replica)
|
|
for o in (self,) + others:
|
|
out.adds |= o.adds
|
|
out.removed |= o.removed
|
|
out.clock = max(out.clock, o.clock)
|
|
return out
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# tests
|
|
# ---------------------------------------------------------------------------
|
|
class TestMergeLaws(unittest.TestCase):
|
|
def _sample(self, kind, seed):
|
|
reps = make_replicas(kind, 3, 40, seed)
|
|
return [kind.loads(r.dumps()) for r in reps]
|
|
|
|
def test_commutative_and_associative(self):
|
|
for kind in (ORSet, PNCounter, LWWRegister):
|
|
for seed in range(8):
|
|
a, b, c = self._sample(kind, seed)
|
|
self.assertEqual(a.merge(b).state_key(), b.merge(a).state_key(),
|
|
"%s not commutative (seed %d)" % (kind.__name__, seed))
|
|
self.assertEqual(a.merge(b).merge(c).state_key(),
|
|
a.merge(b.merge(c)).state_key(),
|
|
"%s not associative (seed %d)" % (kind.__name__, seed))
|
|
|
|
def test_idempotent(self):
|
|
for kind in (ORSet, PNCounter, LWWRegister):
|
|
for seed in range(8):
|
|
for x in self._sample(kind, seed):
|
|
self.assertEqual(x.merge(x).state_key(), x.state_key(),
|
|
"%s not idempotent (seed %d)" % (kind.__name__, seed))
|
|
self.assertEqual(x.merge(x, x, x).state_key(), x.state_key())
|
|
|
|
def test_merge_is_pure(self):
|
|
for kind in (ORSet, PNCounter, LWWRegister):
|
|
a, b, _ = self._sample(kind, 3)
|
|
before = (a.state_key(), b.state_key())
|
|
a.merge(b)
|
|
self.assertEqual((a.state_key(), b.state_key()), before,
|
|
"%s.merge mutated an input" % kind.__name__)
|
|
|
|
def test_duplicate_delivery_harmless(self):
|
|
for kind in (ORSet, PNCounter, LWWRegister):
|
|
a, b, _ = self._sample(kind, 5)
|
|
once = a.merge(b)
|
|
many = a
|
|
for _ in range(5):
|
|
many = many.merge(b)
|
|
self.assertEqual(many.state_key(), once.state_key())
|
|
|
|
def test_type_confusion_refused(self):
|
|
self.assertRaises(CRDTError, ORSet("a").merge, PNCounter("b"))
|
|
self.assertRaises(CRDTError, PNCounter("a").merge, LWWRegister("b"))
|
|
self.assertRaises(CRDTError, LWWRegister("a").merge, ORSet("b"))
|
|
|
|
|
|
class TestORSetSemantics(unittest.TestCase):
|
|
def test_add_wins_on_concurrent_add_remove(self):
|
|
a = ORSet("a")
|
|
a.add("x")
|
|
b = ORSet.loads(a.dumps())
|
|
a.remove("x")
|
|
b.add("x")
|
|
self.assertFalse(a.contains("x"))
|
|
self.assertTrue(b.contains("x"))
|
|
self.assertTrue(a.merge(b).contains("x"), "add-wins violated")
|
|
self.assertTrue(b.merge(a).contains("x"))
|
|
|
|
def test_remove_only_kills_observed_tags(self):
|
|
a = ORSet("a")
|
|
a.add("x")
|
|
b = ORSet.loads(a.dumps())
|
|
b.remove("x")
|
|
a.add("x") # new tag, not observed by b
|
|
self.assertTrue(a.merge(b).contains("x"))
|
|
self.assertFalse(b.merge(b).contains("x"))
|
|
|
|
def test_remove_then_merge_then_still_absent(self):
|
|
a = ORSet("a")
|
|
a.add("x")
|
|
b = a.merge(ORSet("b"))
|
|
b.remove("x")
|
|
self.assertFalse(a.merge(b).contains("x"))
|
|
|
|
def test_elements_and_iteration(self):
|
|
a = ORSet("a")
|
|
for e in ("delta", "alpha", "gamma"):
|
|
a.add(e)
|
|
self.assertEqual(sorted(a), ["alpha", "delta", "gamma"])
|
|
self.assertEqual(len(a), 3)
|
|
self.assertIn("alpha", a)
|
|
a.remove("alpha")
|
|
self.assertEqual(sorted(a), ["delta", "gamma"])
|
|
|
|
def test_rejects_non_scalar_elements(self):
|
|
a = ORSet("a")
|
|
self.assertRaises(CRDTError, a.add, ["list"])
|
|
self.assertRaises(CRDTError, a.add, {"k": 1})
|
|
self.assertRaises(CRDTError, ORSet, "")
|
|
self.assertRaises(CRDTError, a.remove, ["list"])
|
|
|
|
|
|
class TestPNCounter(unittest.TestCase):
|
|
def test_value_and_merge(self):
|
|
a, b = PNCounter("a"), PNCounter("b")
|
|
a.increment(5)
|
|
a.decrement(2)
|
|
b.increment(3)
|
|
self.assertEqual(a.value(), 3)
|
|
self.assertEqual(b.value(), 3)
|
|
m = a.merge(b)
|
|
self.assertEqual(m.value(), 6)
|
|
self.assertEqual(m.state_key(), b.merge(a).state_key())
|
|
|
|
def test_pointwise_max_not_sum(self):
|
|
a = PNCounter("a")
|
|
a.increment(4)
|
|
again = PNCounter.loads(a.dumps())
|
|
self.assertEqual(a.merge(again).value(), 4, "merge double-counted a slot")
|
|
|
|
def test_rejects_negative_delta(self):
|
|
a = PNCounter("a")
|
|
self.assertRaises(CRDTError, a.increment, -1)
|
|
self.assertRaises(CRDTError, a.decrement, -3)
|
|
|
|
|
|
class TestLWWRegister(unittest.TestCase):
|
|
def test_max_stamp_wins_regardless_of_order(self):
|
|
a, b = LWWRegister("a"), LWWRegister("b")
|
|
a.set("old", 1)
|
|
b.set("new", 2)
|
|
self.assertEqual(a.merge(b).get(), "new")
|
|
self.assertEqual(b.merge(a).get(), "new")
|
|
|
|
def test_replica_tiebreak(self):
|
|
a, b = LWWRegister("a"), LWWRegister("b")
|
|
a.set("from-a", 7)
|
|
b.set("from-b", 7)
|
|
self.assertEqual(a.merge(b).get(), "from-b") # ("b" > "a") at equal ts
|
|
self.assertEqual(b.merge(a).get(), "from-b")
|
|
|
|
def test_non_monotone_write_refused(self):
|
|
a = LWWRegister("a")
|
|
a.set("v1", 5)
|
|
self.assertRaises(CRDTError, a.set, "v2", 5)
|
|
self.assertRaises(CRDTError, a.set, "v2", 4)
|
|
|
|
def test_identical_stamp_conflict_refused(self):
|
|
a = LWWRegister("a")
|
|
a.set("v1", 3)
|
|
impostor = LWWRegister.loads(a.dumps())
|
|
impostor._value = "v2"
|
|
self.assertRaises(CRDTError, a.merge, impostor)
|
|
|
|
|
|
class TestSerialization(unittest.TestCase):
|
|
def test_roundtrip_preserves_state(self):
|
|
for kind in (ORSet, PNCounter, LWWRegister):
|
|
for seed in range(6):
|
|
for r in make_replicas(kind, 3, 30, seed):
|
|
self.assertEqual(kind.loads(r.dumps()).state_key(), r.state_key(),
|
|
"%s roundtrip lost state" % kind.__name__)
|
|
|
|
def test_merge_across_serialization_boundary(self):
|
|
for kind in (ORSet, PNCounter, LWWRegister):
|
|
reps = make_replicas(kind, 3, 50, 11)
|
|
direct = reps[0].merge(reps[1])
|
|
over_wire = kind.loads(reps[0].dumps()).merge(kind.loads(reps[1].dumps()))
|
|
self.assertEqual(over_wire.state_key(), direct.state_key())
|
|
self.assertEqual(over_wire.value_key(), direct.value_key())
|
|
|
|
def test_malformed_state_refused(self):
|
|
for kind in (ORSet, PNCounter, LWWRegister):
|
|
self.assertRaises(CRDTError, kind.loads, "not json")
|
|
self.assertRaises(CRDTError, kind.loads, "[1, 2, 3]")
|
|
self.assertRaises(CRDTError, kind.loads, '{"type": "Other"}')
|
|
|
|
|
|
class TestDifferentialReference(unittest.TestCase):
|
|
"""Compare OR-Set against an independently written naive reference model."""
|
|
|
|
def test_orset_matches_independent_reference(self):
|
|
for seed in range(16):
|
|
rng = random.Random(seed * 31 + 5)
|
|
live = [ORSet("r%d" % i) for i in range(3)]
|
|
ref = [RefORSet("r%d" % i) for i in range(3)]
|
|
for _ in range(120):
|
|
i, elem = rng.randrange(3), rng.choice(ELEMS)
|
|
if rng.random() < 0.55:
|
|
live[i].add(elem)
|
|
ref[i].add(elem)
|
|
else:
|
|
live[i].remove(elem)
|
|
ref[i].remove(elem)
|
|
schedule = [(rng.randrange(3), rng.randrange(3)) for _ in range(60)]
|
|
for dst, src in schedule:
|
|
if dst == src:
|
|
continue
|
|
live[dst] = live[dst].merge(live[src])
|
|
ref[dst] = ref[dst].merge(ref[src])
|
|
self.assertEqual(live[0].elements(), ref[0].elements(),
|
|
"differential mismatch at seed %d" % seed)
|
|
for x in live[1:]:
|
|
self.assertEqual(x.elements(), live[0].elements(),
|
|
"replica split at seed %d" % seed)
|
|
|
|
|
|
class TestConvergence(unittest.TestCase):
|
|
def test_random_gossip_converges_all_kinds(self):
|
|
for kind in (ORSet, PNCounter, LWWRegister):
|
|
for seed in SEEDS:
|
|
detail = check_convergence_detail(kind, seed)
|
|
self.assertTrue(
|
|
detail["state_ok"] and detail["value_ok"],
|
|
"%s diverged at seed %d: %r" % (kind.__name__, seed, detail),
|
|
)
|
|
self.assertTrue(check_convergence(kind, seed),
|
|
"%s failed join check at seed %d" % (kind.__name__, seed))
|
|
|
|
def test_larger_ensemble_converges(self):
|
|
for kind in (ORSet, PNCounter, LWWRegister):
|
|
for seed in (0, 1, 2):
|
|
self.assertTrue(
|
|
check_convergence(kind, seed, n=5, ops=150, rounds=6),
|
|
"%s diverged with 5 replicas at seed %d" % (kind.__name__, seed),
|
|
)
|
|
|
|
|
|
class TestMutantsBreakNamedLaws(unittest.TestCase):
|
|
"""Each mutant is tied to the specific law it violates."""
|
|
|
|
def test_sum_merge_is_not_idempotent(self):
|
|
a = PNCounterSumMerge("a")
|
|
a.increment(3)
|
|
self.assertNotEqual(a.merge(a).state_key(), a.state_key())
|
|
self.assertEqual(a.merge(a).value(), 6)
|
|
|
|
def test_noop_merge_is_not_commutative(self):
|
|
a, b = ORSetNoOpMerge("a"), ORSetNoOpMerge("b")
|
|
a.add("x")
|
|
b.add("y")
|
|
self.assertNotEqual(a.merge(b).state_key(), b.merge(a).state_key())
|
|
|
|
def test_last_argument_wins_is_not_commutative(self):
|
|
a, b = LWWLastArgumentWins("a"), LWWLastArgumentWins("b")
|
|
a.set("A", 1)
|
|
b.set("B", 2)
|
|
self.assertNotEqual(a.merge(b).value_key(), b.merge(a).value_key())
|
|
|
|
|
|
class TestMutationTeeth(unittest.TestCase):
|
|
"""If a broken merge survives the check, the check is decoration."""
|
|
|
|
MUTANTS = (ORSetDropsForeignTombstones, PNCounterDropsDecrements,
|
|
LWWLastArgumentWins, ORSetNoOpMerge, PNCounterSumMerge)
|
|
|
|
def test_broken_merges_are_caught(self):
|
|
for mutant in self.MUTANTS:
|
|
hits = mutant_detected(mutant)
|
|
self.assertTrue(hits, "%s survived every seed -- check has no teeth"
|
|
% mutant.__name__)
|
|
for seed in hits:
|
|
self.assertFalse(check_convergence(mutant, seed))
|
|
|
|
def test_real_classes_pass_the_same_seeds(self):
|
|
pairs = ((ORSet, ORSetDropsForeignTombstones),
|
|
(PNCounter, PNCounterDropsDecrements),
|
|
(LWWRegister, LWWLastArgumentWins),
|
|
(ORSet, ORSetNoOpMerge),
|
|
(PNCounter, PNCounterSumMerge))
|
|
for real, mutant in pairs:
|
|
hits = mutant_detected(mutant)
|
|
self.assertTrue(hits)
|
|
for seed in hits:
|
|
self.assertTrue(check_convergence(real, seed),
|
|
"%s failed where its mutant was caught (seed %d)"
|
|
% (real.__name__, seed))
|
|
|
|
def test_orset_mutant_is_semantically_wrong_not_just_state_wrong(self):
|
|
hits = mutant_detected(ORSetDropsForeignTombstones)
|
|
self.assertTrue(hits)
|
|
wrong_value = False
|
|
for seed in hits:
|
|
d = check_convergence_detail(ORSetDropsForeignTombstones, seed)
|
|
if not d["value_ok"]:
|
|
wrong_value = True
|
|
self.assertTrue(wrong_value, "OR-Set mutant only differed in bookkeeping")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2 if "-v" in sys.argv else 1)
|