java-topology/defects/substrate/unit/test_substrate_0003_npos_node_root.py

"""
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<NodeRef>

    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<NodeRef<A>> - 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<NodeId<A>> - 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)