generated from coulomb/repo-seed
tegwick
24108b65aa
Append-only, totally-ordered-per-space decision log (in-process append authority; git+lease later) with event types overlay/binding/alias/merge/fork, and a derived fold to CoordinationState (aliases LWW, transitively-merged equivalence groups, open overlays). derived = f(log); read-your-writes. 6 tests green. (blueprint §8.1) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
52 lines
2.1 KiB
Python
52 lines
2.1 KiB
Python
"""Tests for the event-sourced DecisionLog (SHARD-WP-0007 T5)."""
|
|
|
|
from shard_wiki.coordination import DecisionLog, EventType
|
|
|
|
|
|
def test_append_is_totally_ordered_per_space():
|
|
log = DecisionLog()
|
|
log.append("spaceA", EventType.ALIAS_SET, {"alias": "Home", "target": "shardA:Index"})
|
|
log.append("spaceA", EventType.ALIAS_SET, {"alias": "Start", "target": "shardA:Index"})
|
|
log.append("spaceB", EventType.ALIAS_SET, {"alias": "X", "target": "shardB:Y"})
|
|
seqs = [e.seq for e in log.events("spaceA")]
|
|
assert seqs == [0, 1] # per-space monotonic
|
|
assert [e.seq for e in log.events("spaceB")] == [0] # independent ordering
|
|
|
|
|
|
def test_read_your_writes():
|
|
log = DecisionLog()
|
|
ev = log.append("s", EventType.ALIAS_SET, {"alias": "a", "target": "shardA:b"})
|
|
assert log.events("s")[-1] is ev
|
|
|
|
|
|
def test_fold_reproduces_current_state():
|
|
log = DecisionLog()
|
|
log.append("s", EventType.ALIAS_SET, {"alias": "Home", "target": "shardA:Index"})
|
|
log.append("s", EventType.ALIAS_SET, {"alias": "Home", "target": "shardB:Main"}) # LWW
|
|
state = log.fold("s")
|
|
assert state.resolve_alias("Home") == "shardB:Main"
|
|
assert state.resolve_alias("missing") is None
|
|
|
|
|
|
def test_fold_is_pure_function_of_log():
|
|
log = DecisionLog()
|
|
log.append("s", EventType.BINDING_MADE, {"members": ["shardA:Home", "shardB:Home"]})
|
|
first = log.fold("s")
|
|
second = log.fold("s")
|
|
assert first.equivalence_groups == second.equivalence_groups # derived = f(log)
|
|
|
|
|
|
def test_equivalence_groups_merge_transitively():
|
|
log = DecisionLog()
|
|
log.append("s", EventType.BINDING_MADE, {"members": ["a", "b"]})
|
|
log.append("s", EventType.BINDING_MADE, {"members": ["b", "c"]})
|
|
state = log.fold("s")
|
|
assert state.equivalent_to("a") == frozenset({"a", "b", "c"})
|
|
assert state.equivalent_to("lonely") == frozenset({"lonely"})
|
|
|
|
|
|
def test_page_fork_creates_equivalence():
|
|
log = DecisionLog()
|
|
log.append("s", EventType.PAGE_FORKED, {"source": "shardA:Doc", "fork": "shardB:Doc"})
|
|
assert log.fold("s").equivalent_to("shardA:Doc") == frozenset({"shardA:Doc", "shardB:Doc"})
|