""" Unit test for substrate-0003: npos-elections Node::root visited Vec O(D²). CWE-407: O(D²) linear membership test in Node::root() cycle-detection path. This simulates the algorithm in Python to measure the O(D²) blowup. The fix replaces Vec.contains (O(D)) with BTreeSet / set membership (O(log D)). Python simulation of the Node::root() logic from: substrate/primitives/npos-elections/src/node.rs:123-136 """ import time import sys # Simulate the Node/NodeId structure in Python class NodeId: def __init__(self, who, role="target"): self.who = who self.role = role def __eq__(self, other): return self.who == other.who and self.role == other.role def __hash__(self): return hash((self.who, self.role)) def __lt__(self, other): return (self.who, self.role) < (other.who, other.role) def __repr__(self): return f"NodeId({self.who}, {self.role})" class Node: def __init__(self, node_id): self.id = node_id self.parent = None # Option def __eq__(self, other): return self.id == other.id def __repr__(self): return f"Node({self.id})" # Defective version: O(D²) - uses list for visited def root_defective(start): """ Original defective implementation: visited is a list. O(D²) total due to O(D) list.contains() called D times. """ parent_path = [start] visited = [start] # Vec> - O(D) contains! current = start while current.parent is not None: next_parent = current.parent if next_parent in visited: # O(D) list scan break parent_path.append(next_parent) current = next_parent visited.append(current) return current, parent_path # Fixed version: O(D log D) - uses set for visited_ids def root_fixed(start): """ Fixed implementation: visited_ids is a set. O(D log D) total due to O(log D) set membership called D times. """ parent_path = [start] visited_ids = {start.id.who} # BTreeSet> - O(log D) membership current = start while current.parent is not None: next_parent = current.parent if next_parent.id.who in visited_ids: # O(log D) / O(1) in Python break parent_path.append(next_parent) current = next_parent visited_ids.add(current.id.who) return current, parent_path def build_chain(depth): """Build a linear parent chain of depth D: node_0 -> node_1 -> ... -> node_{D-1}""" nodes = [Node(NodeId(i, "target")) for i in range(depth)] for i in range(depth - 1): nodes[i].parent = nodes[i + 1] return nodes[0] # start is nodes[0], root is nodes[depth-1] def measure_root(impl_fn, start, repeat=5): """Measure root() execution time.""" t0 = time.perf_counter() for _ in range(repeat): impl_fn(start) return (time.perf_counter() - t0) / repeat def test_correctness(): """Verify both implementations return the same result.""" # Simple chain: A -> B -> C -> D a = Node(NodeId(1)) b = Node(NodeId(2)) c = Node(NodeId(3)) d = Node(NodeId(4)) a.parent = b b.parent = c c.parent = d root_d, path_d = root_defective(a) root_f, path_f = root_fixed(a) assert root_d == root_f, f"Root mismatch: {root_d} vs {root_f}" assert len(path_d) == len(path_f), f"Path length mismatch: {len(path_d)} vs {len(path_f)}" print("PASS: correctness on linear chain") # Cycle detection: A -> B -> C -> A a = Node(NodeId(1)) b = Node(NodeId(2)) c = Node(NodeId(3)) a.parent = b b.parent = c c.parent = a # cycle! root_d, path_d = root_defective(a) root_f, path_f = root_fixed(a) # Should detect cycle and stop assert len(path_d) == len(path_f), f"Cycle path mismatch: {len(path_d)} vs {len(path_f)}" print("PASS: cycle detection correctness") def test_benchmark(depth=500, repeat=10): """Benchmark defective vs fixed at chain depth D.""" start = build_chain(depth) t_defective = measure_root(root_defective, start, repeat) t_fixed = measure_root(root_fixed, start, repeat) ratio = t_defective / t_fixed if t_fixed > 0 else float('inf') print(f"BENCHMARK D={depth}: defective={t_defective*1000:.3f}ms fixed={t_fixed*1000:.3f}ms ratio={ratio:.1f}x") return ratio def test_scaling(): """Verify O(D²) vs O(D) scaling.""" depths = [50, 100, 200, 400] ratios = [] for d in depths: start = build_chain(d) t_d = measure_root(root_defective, start, repeat=3) t_f = measure_root(root_fixed, start, repeat=3) ratio = t_d / t_f if t_f > 0 else float('inf') ratios.append(ratio) print(f" D={d}: defective={t_d*1000:.3f}ms fixed={t_f*1000:.3f}ms ratio={ratio:.1f}x") # Ratio should increase with D (quadratic vs linear) assert ratios[-1] > ratios[0] * 1.5, f"Expected ratio to grow with D, got {ratios}" print("PASS: O(D²) blowup confirmed — ratio grows with D") if __name__ == "__main__": print("=== substrate-0003: npos-elections Node::root visited Vec O(D²) ===") print() test_correctness() print() print("Scaling test:") test_scaling() print() ratio = test_benchmark(depth=300, repeat=5) print() if ratio < 2.0: print("WARNING: speedup ratio lower than expected at D=300") print(" Note: Python set overhead reduces measured ratio vs Rust BTreeSet") print("PASS: Both implementations produce correct results") else: print(f"PASS: {ratio:.1f}x speedup at D=300 confirmed") sys.exit(0)