"""Stdlib-only tests (no pytest): plain asserts in functions called from a __main__ block. Run: `python3 test_crdt.py`. These pin the numbers the commentary quotes: the seed-0 trace and both of its flips, the flat LWW column and the 16.8x OR-Set improvement under gossip, the closed form, the load-bearing `covers` test, and both boundary conditions. The heavier sweeps run at the same 200 seeds as the demo, so the whole file takes about half a minute. """ import io import contextlib import random from crdt import (POLICIES, Replica, PureLWW, fmt, gen_trace, present, run, run_gossip, sequential) SEED, N_REP, N_ELEM, N_OPS, SKEW = 0, 3, 5, 16, 9 SEEDS = 200 SW_REP, SW_ELEM, SW_OPS = 3, 6, 120 SKEW_CHOICES = [0, 3, 7, 25, 100] def gossip_sweep(rate, policy, n_rep=SW_REP, seeds=SEEDS): """Mean elements wrong, plus how many traces were exactly right.""" wrong, exact, absent, allconv = 0, 0, 0, True for seed in range(seeds): rng = random.Random(10000 + seed) skews = {r: rng.choice(SKEW_CHOICES) for r in range(n_rep)} value, conv, ground = run_gossip(seed, n_rep, policy, rate, skews, n_ops=SW_OPS, n_elems=SW_ELEM) wrong += len(value ^ ground) exact += value == ground absent += len(ground - value) allconv = allconv and conv return wrong / seeds, exact, absent / seeds, allconv def schedule_sweep(every, policy, skew=1000, seeds=SEEDS): wrong, exact = 0, 0 for seed in range(seeds): trace = gen_trace(seed, SW_OPS, SW_REP, SW_ELEM, every) value, conv, _ = run(trace, policy, SW_REP, skew={0: skew}) assert conv, "replicas diverged at sync_every=%d" % every truth = sequential(trace) wrong += len(value ^ truth) exact += value == truth return wrong / seeds, exact # --- the headline --------------------------------------------------------- def test_gossip_moves_the_orset_and_not_the_lww_store(): """60x the gossip: 1.960 -> 0.130 for the OR-Set, 1.280 flat for LWW.""" orset = {rate: gossip_sweep(rate, "add-wins")[0] for rate in (0.0, 0.05, 0.5, 3.0)} assert orset == {0.0: 2.185, 0.05: 1.960, 0.5: 0.880, 3.0: 0.130}, orset assert round(orset[0.0] / orset[3.0], 1) == 16.8 assert round(orset[0.05] / orset[3.0], 1) == 15.1 lww = {} for rate in (0.0, 0.05, 0.5, 3.0): wrong, exact, _, conv = gossip_sweep(rate, "pure-lww") assert conv lww[rate] = (wrong, exact) assert set(lww.values()) == {(1.280, 56)}, lww def test_every_run_converges_including_the_ones_that_are_wrong(): """Convergence is free: it holds at every gossip rate, both structures.""" for rate in (0.0, 0.5, 3.0): for policy in ("add-wins", "pure-lww"): assert gossip_sweep(rate, policy)[3] is True def test_the_orset_never_loses_an_add(): """All of the OR-Set's error is resurrection; LWW's is half deletion.""" for rate in (0.0, 0.5, 3.0): assert gossip_sweep(rate, "add-wins")[2] == 0.0 assert gossip_sweep(rate, "pure-lww")[2] == 0.615 # --- the seed-0 trace ----------------------------------------------------- def test_seed_zero_trace_is_the_one_the_page_prints(): trace = gen_trace(SEED, N_OPS, N_REP, N_ELEM, 1) ops = [(e[1], e[3], e[2]) for e in trace if e[0] == "op"] assert ops[:4] == [(1, "add", "e3"), (2, "add", "e3"), (1, "add", "e3"), (0, "add", "e4")] assert ops[-3:] == [(2, "add", "e0"), (1, "rm", "e0"), (1, "rm", "e1")] assert fmt(sequential(trace)) == "{e2}" def test_seed_zero_three_schedules(): truth = sequential(gen_trace(SEED, N_OPS, N_REP, N_ELEM, 1)) expected = {1: ("{e2}", "{e2, e4}"), 4: ("{e0, e2}", "{e2, e4}"), 16: ("{e0, e2, e3, e4}", "{e2, e4}")} for every, (want_orset, want_lww) in expected.items(): trace = gen_trace(SEED, N_OPS, N_REP, N_ELEM, every) orset, c1, _ = run(trace, "add-wins", N_REP, skew={0: SKEW}) lww, c2, _ = run(trace, "pure-lww", N_REP, skew={0: SKEW}) assert c1 and c2, every assert fmt(orset) == want_orset, (every, fmt(orset)) assert fmt(lww) == want_lww, (every, fmt(lww)) # The LWW answer is identical at all three; the OR-Set's is not. assert len({run(gen_trace(SEED, N_OPS, N_REP, N_ELEM, k), "pure-lww", N_REP, skew={0: SKEW})[0] for k in (1, 4, 16)}) == 1 assert len(truth ^ frozenset({"e2", "e4"})) == 1 def test_e4_flips_at_skew_three(): """The add is at t=3 and the delete at t=5, so it needs +3 to overtake.""" trace = gen_trace(SEED, N_OPS, N_REP, N_ELEM, 1) got = {} for skew in range(6): value, conv, _ = run(trace, "pure-lww", N_REP, skew={0: skew}) assert conv got[skew] = "e4" in value assert got == {0: False, 1: False, 2: False, 3: True, 4: True, 5: True}, got def test_e0_flips_between_sync_every_2_and_4(): """No clock is involved: the delete either saw the add or it didn't.""" got = {} for every in (1, 2, 4, 8, 16): trace = gen_trace(SEED, N_OPS, N_REP, N_ELEM, every) value, conv, _ = run(trace, "add-wins", N_REP, skew={0: SKEW}) assert conv got[every] = "e0" in value assert got == {1: False, 2: False, 4: True, 8: True, 16: True}, got # --- the invariance and its closed form ----------------------------------- def test_lww_value_is_identical_across_nine_sync_schedules(): schedules = (1, 2, 3, 4, 8, 16, 32, 64, 120) changed = {"add-wins": 0, "pure-lww": 0} for seed in range(SEEDS): for policy in ("add-wins", "pure-lww"): values = set() for every in schedules: trace = gen_trace(seed, SW_OPS, SW_REP, SW_ELEM, every) value, conv, _ = run(trace, policy, SW_REP, skew={0: 1000}) assert conv values.add(value) changed[policy] += len(values) != 1 assert changed == {"add-wins": 186, "pure-lww": 0}, changed def test_closed_form_predicts_the_lww_column_exactly(): predicted = SW_ELEM * (2 / 3) * (2 * 0.55 * 0.45) assert round(predicted, 4) == 1.9800 missed = 0 for seed in range(SEEDS): trace = gen_trace(seed, SW_OPS, SW_REP, SW_ELEM, 8) last_r0, last_any = {}, {} for event in trace: if event[0] != "op": continue _, rid, elem, label = event last_any[elem] = label if rid == 0: last_r0[elem] = label prediction = frozenset(e for e in last_any if last_r0.get(e, last_any[e]) == "add") value, _, _ = run(trace, "pure-lww", SW_REP, skew={0: 1000}) missed += value != prediction assert missed == 0, missed assert schedule_sweep(8, "pure-lww")[0] == 1.980 # --- the load-bearing line ------------------------------------------------ def test_covers_is_load_bearing_at_sync_every_one_and_inert_at_120(): baseline = {k: schedule_sweep(k, "add-wins", skew=0)[0] for k in (1, 8, 120)} assert baseline == {1: 0.000, 8: 0.650, 120: 2.185}, baseline union = {} for every in (1, 8, 120): wrong = 0 for seed in range(SEEDS): trace = gen_trace(seed, SW_OPS, SW_REP, SW_ELEM, every) value, conv, _ = run(trace, "add-wins", SW_REP, causal_merge=False) assert conv, "a plain-union merge still converges" wrong += len(value ^ sequential(trace)) union[every] = wrong / SEEDS assert union == {1: 2.710, 8: 2.685, 120: 2.185}, union def test_without_covers_the_set_can_never_delete(): """At sync_every=1 the union merge's value is every element ever touched.""" for seed in range(SEEDS): trace = gen_trace(seed, SW_OPS, SW_REP, SW_ELEM, 1) value, _, _ = run(trace, "add-wins", SW_REP, causal_merge=False) assert value == frozenset(e[2] for e in trace if e[0] == "op") def test_supersede_is_what_bounds_the_state(): """Turning it off breaks the answer at every schedule, not just wide ones.""" got = {} for every in (1, 8, 120): wrong = 0 for seed in range(SEEDS): trace = gen_trace(seed, SW_OPS, SW_REP, SW_ELEM, every) value, _, _ = run(trace, "add-wins", SW_REP, supersede=False) wrong += len(value ^ sequential(trace)) got[every] = wrong / SEEDS assert got == {1: 2.710, 8: 2.710, 120: 2.710}, got # --- the boundaries ------------------------------------------------------- def test_lww_is_exact_when_the_clock_is_good_enough(): got = {skew: schedule_sweep(8, "pure-lww", skew=skew) for skew in range(5)} assert got[0] == (0.000, 200) and got[1] == (0.000, 200), got assert got[2] == (0.135, 173) and got[3] == (0.200, 161), got assert got[4] == (0.310, 144), got # With a perfect clock, the OR-Set is the worse structure everywhere. for rate in (0.0, 0.5, 3.0): lww, orset = 0, 0 for seed in range(SEEDS): v, _, ground = run_gossip(seed, SW_REP, "pure-lww", rate, {}, n_ops=SW_OPS, n_elems=SW_ELEM) lww += len(v ^ ground) v, _, ground = run_gossip(seed, SW_REP, "add-wins", rate, {}, n_ops=SW_OPS, n_elems=SW_ELEM) orset += len(v ^ ground) assert lww == 0 and orset > 0, (rate, lww, orset) def test_the_policy_is_unobservable_without_concurrency(): got = {} for every in (1, 12): differ = 0 for seed in range(SEEDS): values = set() for policy in POLICIES: trace = gen_trace(seed, SW_OPS, SW_REP, SW_ELEM, every) value, conv, _ = run(trace, policy, SW_REP, skew={0: 7}) assert conv values.add(value) differ += len(values) > 1 got[every] = differ assert got == {1: 0, 12: 185}, got # --- unit checks on the mechanism ----------------------------------------- def test_concurrent_add_and_remove_survive_as_two_assertions(): a, b = Replica(0), Replica(1) a.add("x", 0) b.merge(a) # b has seen a's add b.remove("x", 1) # ... so this supersedes it a.add("x", 2) # concurrent with b's remove: a never saw it a.merge(b) b.merge(a) assert len(a.el["x"]) == 2, a.el["x"] assert a.value() == b.value() == frozenset({"x"}) assert present(a.el["x"], "add-wins") is True assert present(a.el["x"], "rm-wins") is False assert present(a.el["x"], "lww") is True # a's add is stamped 2 > 1 def test_a_causally_prior_assertion_is_dropped_not_kept(): a, b = Replica(0), Replica(1) a.add("x", 0) b.merge(a) b.remove("x", 1) a.merge(b) assert len(a.el["x"]) == 1 assert a.value() == frozenset() assert a.covers((0, 1)) and b.covers((0, 1)) def test_merge_is_idempotent_and_commutative(): a, b = Replica(0), Replica(1) a.add("x", 0) b.add("y", 0) b.remove("x", 1) before = None for _ in range(4): a.merge(b) b.merge(a) assert before is None or a.value() == before before = a.value() assert a.value() == b.value() == frozenset({"x", "y"}) def test_pure_lww_ignores_arrival_order_entirely(): a, b = PureLWW(0), PureLWW(1) a.add("x", 10) b.remove("x", 3) a.merge(b) b.merge(a) assert a.value() == b.value() == frozenset({"x"}) # The delete happened later in real time and still loses. def test_demo_output_is_byte_identical_across_runs(): """demo.py prints on import, so importing and reloading it runs it twice.""" import importlib buf = io.StringIO() with contextlib.redirect_stdout(buf): demo = importlib.import_module("demo") first = buf.getvalue() buf = io.StringIO() with contextlib.redirect_stdout(buf): importlib.reload(demo) assert first == buf.getvalue() assert "60x the gossip moves one column and not the other." in first TESTS = [ test_gossip_moves_the_orset_and_not_the_lww_store, test_every_run_converges_including_the_ones_that_are_wrong, test_the_orset_never_loses_an_add, test_seed_zero_trace_is_the_one_the_page_prints, test_seed_zero_three_schedules, test_e4_flips_at_skew_three, test_e0_flips_between_sync_every_2_and_4, test_lww_value_is_identical_across_nine_sync_schedules, test_closed_form_predicts_the_lww_column_exactly, test_covers_is_load_bearing_at_sync_every_one_and_inert_at_120, test_without_covers_the_set_can_never_delete, test_supersede_is_what_bounds_the_state, test_lww_is_exact_when_the_clock_is_good_enough, test_the_policy_is_unobservable_without_concurrency, test_concurrent_add_and_remove_survive_as_two_assertions, test_a_causally_prior_assertion_is_dropped_not_kept, test_merge_is_idempotent_and_commutative, test_pure_lww_ignores_arrival_order_entirely, test_demo_output_is_byte_identical_across_runs, ] if __name__ == "__main__": failed = 0 for test in TESTS: try: test() except AssertionError as exc: failed += 1 print("FAIL %s: %s" % (test.__name__, exc)) else: print("PASS %s" % test.__name__) if failed: print("\n%d of %d tests FAILED" % (failed, len(TESTS))) raise SystemExit(1) print("\nAll %d tests PASSED" % len(TESTS))