test_graph_quorum.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """TDD — I3: Quorum soundness invariant. |
| 2 | |
| 3 | An org's vote in a parent org counts only if that org's own quorum |
| 4 | is independently satisfied. Recursive. Terminates because I1 guarantees |
| 5 | the graph is a DAG. |
| 6 | """ |
| 7 | from __future__ import annotations |
| 8 | |
| 9 | from collections.abc import Mapping |
| 10 | from decimal import Decimal |
| 11 | |
| 12 | import pytest |
| 13 | |
| 14 | from musehub.graph.dag import EdgeType, GraphEdge, IdentityDAG, NodeType |
| 15 | from musehub.graph.quorum import OrgSpec, QuorumEngine, VoteRecord |
| 16 | |
| 17 | |
| 18 | S = EdgeType.SPAWNS |
| 19 | M = EdgeType.MEMBER_OF |
| 20 | |
| 21 | |
| 22 | def spec(handle: str, quorum: int, members: Mapping[str, Decimal]) -> OrgSpec: |
| 23 | """Convenience constructor for OrgSpec.""" |
| 24 | return OrgSpec(handle=handle, quorum=quorum, member_weights=members) |
| 25 | |
| 26 | |
| 27 | def vote(voter: str, org: str) -> VoteRecord: |
| 28 | return VoteRecord(voter_handle=voter, org_handle=org) |
| 29 | |
| 30 | |
| 31 | # ── Simple flat orgs ────────────────────────────────────────────────────────── |
| 32 | |
| 33 | class TestFlatQuorum: |
| 34 | def test_quorum_met_exact_threshold(self) -> None: |
| 35 | # 3 members, quorum=2, exactly 2 vote |
| 36 | engine = QuorumEngine( |
| 37 | orgs={"acme": spec("acme", quorum=2, members={ |
| 38 | "alice": Decimal("1"), |
| 39 | "bob": Decimal("1"), |
| 40 | "carol": Decimal("1"), |
| 41 | })} |
| 42 | ) |
| 43 | votes = [vote("alice", "acme"), vote("bob", "acme")] |
| 44 | assert engine.is_quorum_met("acme", votes) is True |
| 45 | |
| 46 | def test_quorum_not_met_one_short(self) -> None: |
| 47 | engine = QuorumEngine( |
| 48 | orgs={"acme": spec("acme", quorum=2, members={ |
| 49 | "alice": Decimal("1"), |
| 50 | "bob": Decimal("1"), |
| 51 | "carol": Decimal("1"), |
| 52 | })} |
| 53 | ) |
| 54 | votes = [vote("alice", "acme")] |
| 55 | assert engine.is_quorum_met("acme", votes) is False |
| 56 | |
| 57 | def test_quorum_not_met_no_votes(self) -> None: |
| 58 | engine = QuorumEngine( |
| 59 | orgs={"acme": spec("acme", quorum=1, members={"alice": Decimal("1")})} |
| 60 | ) |
| 61 | assert engine.is_quorum_met("acme", []) is False |
| 62 | |
| 63 | def test_quorum_1_met_by_single_vote(self) -> None: |
| 64 | engine = QuorumEngine( |
| 65 | orgs={"acme": spec("acme", quorum=1, members={"alice": Decimal("1")})} |
| 66 | ) |
| 67 | assert engine.is_quorum_met("acme", [vote("alice", "acme")]) is True |
| 68 | |
| 69 | def test_unanimous_quorum(self) -> None: |
| 70 | engine = QuorumEngine( |
| 71 | orgs={"acme": spec("acme", quorum=3, members={ |
| 72 | "alice": Decimal("1"), |
| 73 | "bob": Decimal("1"), |
| 74 | "carol": Decimal("1"), |
| 75 | })} |
| 76 | ) |
| 77 | votes = [vote("alice", "acme"), vote("bob", "acme"), vote("carol", "acme")] |
| 78 | assert engine.is_quorum_met("acme", votes) is True |
| 79 | |
| 80 | def test_non_member_vote_does_not_count(self) -> None: |
| 81 | engine = QuorumEngine( |
| 82 | orgs={"acme": spec("acme", quorum=1, members={"alice": Decimal("1")})} |
| 83 | ) |
| 84 | # dave is not a member of acme |
| 85 | assert engine.is_quorum_met("acme", [vote("dave", "acme")]) is False |
| 86 | |
| 87 | |
| 88 | # ── Weighted members ────────────────────────────────────────────────────────── |
| 89 | |
| 90 | class TestWeightedQuorum: |
| 91 | def test_weighted_vote_reaches_quorum(self) -> None: |
| 92 | # alice has weight 2, quorum=2 → alice alone satisfies it |
| 93 | engine = QuorumEngine( |
| 94 | orgs={"acme": spec("acme", quorum=2, members={ |
| 95 | "alice": Decimal("2"), |
| 96 | "bob": Decimal("1"), |
| 97 | })} |
| 98 | ) |
| 99 | assert engine.is_quorum_met("acme", [vote("alice", "acme")]) is True |
| 100 | |
| 101 | def test_low_weight_vote_does_not_reach_quorum(self) -> None: |
| 102 | engine = QuorumEngine( |
| 103 | orgs={"acme": spec("acme", quorum=3, members={ |
| 104 | "alice": Decimal("1"), |
| 105 | "bob": Decimal("1"), |
| 106 | "carol": Decimal("1"), |
| 107 | })} |
| 108 | ) |
| 109 | # only alice (weight 1) voted — needs 3 |
| 110 | assert engine.is_quorum_met("acme", [vote("alice", "acme")]) is False |
| 111 | |
| 112 | def test_fractional_weights_sum_to_quorum(self) -> None: |
| 113 | engine = QuorumEngine( |
| 114 | orgs={"acme": spec("acme", quorum=1, members={ |
| 115 | "alice": Decimal("0.5"), |
| 116 | "bob": Decimal("0.5"), |
| 117 | })} |
| 118 | ) |
| 119 | votes = [vote("alice", "acme"), vote("bob", "acme")] |
| 120 | assert engine.is_quorum_met("acme", votes) is True |
| 121 | |
| 122 | |
| 123 | # ── Nested orgs — quorum propagation ───────────────────────────────────────── |
| 124 | |
| 125 | class TestNestedQuorum: |
| 126 | def _two_level_engine(self) -> QuorumEngine: |
| 127 | # parent-org has [alice, sub-org] as members, quorum=2 |
| 128 | # sub-org has [bob, carol] as members, quorum=1 |
| 129 | return QuorumEngine(orgs={ |
| 130 | "parent-org": spec("parent-org", quorum=2, members={ |
| 131 | "alice": Decimal("1"), |
| 132 | "sub-org": Decimal("1"), |
| 133 | }), |
| 134 | "sub-org": spec("sub-org", quorum=1, members={ |
| 135 | "bob": Decimal("1"), |
| 136 | "carol": Decimal("1"), |
| 137 | }), |
| 138 | }) |
| 139 | |
| 140 | def test_org_vote_counts_when_its_quorum_met(self) -> None: |
| 141 | engine = self._two_level_engine() |
| 142 | # alice votes in parent; bob votes in sub-org (satisfying sub-org quorum=1) |
| 143 | # sub-org's vote in parent then counts → parent total = 2 → met |
| 144 | votes = [ |
| 145 | vote("alice", "parent-org"), |
| 146 | vote("bob", "sub-org"), |
| 147 | vote("sub-org", "parent-org"), |
| 148 | ] |
| 149 | assert engine.is_quorum_met("parent-org", votes) is True |
| 150 | |
| 151 | def test_org_vote_does_not_count_when_its_quorum_not_met(self) -> None: |
| 152 | engine = self._two_level_engine() |
| 153 | # sub-org votes in parent, but nobody voted inside sub-org → sub-org quorum not met |
| 154 | votes = [ |
| 155 | vote("alice", "parent-org"), |
| 156 | vote("sub-org", "parent-org"), |
| 157 | ] |
| 158 | assert engine.is_quorum_met("parent-org", votes) is False |
| 159 | |
| 160 | def test_parent_quorum_not_met_even_if_sub_quorum_met(self) -> None: |
| 161 | engine = self._two_level_engine() |
| 162 | # sub-org quorum met (bob voted) but only sub-org votes in parent → total=1 < quorum=2 |
| 163 | votes = [ |
| 164 | vote("bob", "sub-org"), |
| 165 | vote("sub-org", "parent-org"), |
| 166 | ] |
| 167 | assert engine.is_quorum_met("parent-org", votes) is False |
| 168 | |
| 169 | def test_three_level_nesting_all_quorums_met(self) -> None: |
| 170 | engine = QuorumEngine(orgs={ |
| 171 | "top": spec("top", quorum=2, members={"alice": Decimal("1"), "mid": Decimal("1")}), |
| 172 | "mid": spec("mid", quorum=1, members={"bob": Decimal("1"), "bot": Decimal("1")}), |
| 173 | "bot": spec("bot", quorum=1, members={"carol": Decimal("1")}), |
| 174 | }) |
| 175 | # carol votes in bot → bot quorum met |
| 176 | # bot votes in mid → mid quorum met |
| 177 | # alice + mid vote in top → top quorum met |
| 178 | votes = [ |
| 179 | vote("carol", "bot"), |
| 180 | vote("bot", "mid"), |
| 181 | vote("bob", "mid"), # extra, doesn't hurt |
| 182 | vote("mid", "top"), |
| 183 | vote("alice", "top"), |
| 184 | ] |
| 185 | assert engine.is_quorum_met("top", votes) is True |
| 186 | |
| 187 | def test_three_level_nesting_middle_quorum_not_met(self) -> None: |
| 188 | engine = QuorumEngine(orgs={ |
| 189 | "top": spec("top", quorum=2, members={"alice": Decimal("1"), "mid": Decimal("1")}), |
| 190 | "mid": spec("mid", quorum=2, members={"bob": Decimal("1"), "carol": Decimal("1")}), |
| 191 | }) |
| 192 | # only bob voted in mid → mid quorum (needs 2) not met → mid vote in top doesn't count |
| 193 | votes = [ |
| 194 | vote("bob", "mid"), |
| 195 | vote("mid", "top"), |
| 196 | vote("alice", "top"), |
| 197 | ] |
| 198 | assert engine.is_quorum_met("top", votes) is False |
| 199 | |
| 200 | |
| 201 | # ── Effective weight ────────────────────────────────────────────────────────── |
| 202 | |
| 203 | class TestEffectiveWeight: |
| 204 | def test_human_effective_weight_is_direct(self) -> None: |
| 205 | engine = QuorumEngine(orgs={ |
| 206 | "acme": spec("acme", quorum=1, members={"alice": Decimal("2")}) |
| 207 | }) |
| 208 | assert engine.effective_weight("acme", "alice", []) == Decimal("2") |
| 209 | |
| 210 | def test_org_effective_weight_zero_when_quorum_not_met(self) -> None: |
| 211 | engine = QuorumEngine(orgs={ |
| 212 | "parent": spec("parent", quorum=1, members={"sub": Decimal("3")}), |
| 213 | "sub": spec("sub", quorum=1, members={"alice": Decimal("1")}), |
| 214 | }) |
| 215 | # sub's quorum not met (alice hasn't voted in sub) → sub weight in parent = 0 |
| 216 | assert engine.effective_weight("parent", "sub", []) == Decimal("0") |
| 217 | |
| 218 | def test_org_effective_weight_is_direct_when_quorum_met(self) -> None: |
| 219 | engine = QuorumEngine(orgs={ |
| 220 | "parent": spec("parent", quorum=1, members={"sub": Decimal("3")}), |
| 221 | "sub": spec("sub", quorum=1, members={"alice": Decimal("1")}), |
| 222 | }) |
| 223 | votes = [vote("alice", "sub"), vote("sub", "parent")] |
| 224 | assert engine.effective_weight("parent", "sub", votes) == Decimal("3") |
| 225 | |
| 226 | |
| 227 | # ── Unknown org ─────────────────────────────────────────────────────────────── |
| 228 | |
| 229 | class TestUnknownOrg: |
| 230 | def test_unknown_org_raises(self) -> None: |
| 231 | engine = QuorumEngine(orgs={}) |
| 232 | with pytest.raises(KeyError): |
| 233 | engine.is_quorum_met("ghost-org", []) |