266 lines
8.4 KiB
Python
266 lines
8.4 KiB
Python
"""
|
|
Unit test for crystal-0001:
|
|
collect_ancestors / lookup_defs O(2^D) diamond module traversal
|
|
|
|
This test simulates the Crystal compiler's parent-traversal logic in Python
|
|
to verify the exponential blowup and the linear-time fix.
|
|
"""
|
|
|
|
try:
|
|
import pytest
|
|
except ImportError:
|
|
# Allow running without pytest installed
|
|
import unittest as pytest
|
|
pytest.main = lambda *a, **kw: None
|
|
|
|
|
|
def make_module(name, defs=None):
|
|
"""Simulate a Crystal module type node."""
|
|
return {
|
|
"name": name,
|
|
"parents": [],
|
|
"defs": dict(defs or {}),
|
|
}
|
|
|
|
|
|
# --- Unpatched: collect_ancestors without visited set ---
|
|
|
|
def collect_ancestors_unpatched(node, result, call_counter):
|
|
"""Replicates Crystal's vulnerable collect_ancestors (no visited guard)."""
|
|
for parent in node["parents"]:
|
|
result.append(parent)
|
|
call_counter[0] += 1
|
|
collect_ancestors_unpatched(parent, result, call_counter)
|
|
|
|
|
|
def ancestors_unpatched(node):
|
|
result = []
|
|
counter = [0]
|
|
collect_ancestors_unpatched(node, result, counter)
|
|
return result, counter[0]
|
|
|
|
|
|
# --- Patched: collect_ancestors with visited set ---
|
|
|
|
def collect_ancestors_patched(node, result, visited):
|
|
"""Replicates the fixed collect_ancestors (HashSet visited guard)."""
|
|
for parent in node["parents"]:
|
|
node_id = id(parent)
|
|
if node_id in visited:
|
|
continue
|
|
visited.add(node_id)
|
|
result.append(parent)
|
|
collect_ancestors_patched(parent, result, visited)
|
|
|
|
|
|
def ancestors_patched(node):
|
|
result = []
|
|
visited = set()
|
|
collect_ancestors_patched(node, result, visited)
|
|
return result
|
|
|
|
|
|
# --- lookup_defs unpatched ---
|
|
|
|
def lookup_defs_unpatched(node, name, all_defs, call_counter):
|
|
"""Replicates Crystal's vulnerable lookup_defs (no visited guard)."""
|
|
for defn in node["defs"].get(name, []):
|
|
if defn not in all_defs:
|
|
all_defs.append(defn)
|
|
for parent in node["parents"]:
|
|
call_counter[0] += 1
|
|
lookup_defs_unpatched(parent, name, all_defs, call_counter)
|
|
|
|
|
|
def lookup_defs_patched(node, name, all_defs, visited):
|
|
"""Replicates fixed lookup_defs (visited set prevents re-traversal)."""
|
|
node_id = id(node)
|
|
if node_id in visited:
|
|
return
|
|
visited.add(node_id)
|
|
for defn in node["defs"].get(name, []):
|
|
if defn not in all_defs:
|
|
all_defs.append(defn)
|
|
for parent in node["parents"]:
|
|
lookup_defs_patched(parent, name, all_defs, visited)
|
|
|
|
|
|
# ---- Test helpers ----
|
|
|
|
def build_diamond(depth):
|
|
"""
|
|
Build a D-level diamond module graph:
|
|
M0 is the base (has 'foo')
|
|
At depth=1: C -> [L0, R0]; L0 -> [M0]; R0 -> [M0] (M0 visited 2x)
|
|
At depth=2: C -> [L1, R1]; L1,R1 both -> [L0, R0]; L0,R0 -> [M0] (M0 4x)
|
|
At depth=D: M0 is visited 2^D times by unpatched traversal.
|
|
"""
|
|
m0 = make_module("M0", defs={"foo": ["M0.foo"]})
|
|
|
|
if depth == 0:
|
|
c = make_module("C")
|
|
c["parents"].append(m0)
|
|
return c, m0
|
|
|
|
# Bottom pair: both include M0
|
|
left = make_module("L0")
|
|
right = make_module("R0")
|
|
left["parents"].append(m0)
|
|
right["parents"].append(m0)
|
|
|
|
for d in range(1, depth):
|
|
new_left = make_module(f"L{d}")
|
|
new_right = make_module(f"R{d}")
|
|
# Each new node includes BOTH left and right — true diamond at every level
|
|
new_left["parents"].append(left)
|
|
new_left["parents"].append(right)
|
|
new_right["parents"].append(left)
|
|
new_right["parents"].append(right)
|
|
left, right = new_left, new_right
|
|
|
|
c = make_module("C")
|
|
c["parents"].append(left)
|
|
c["parents"].append(right)
|
|
return c, m0
|
|
|
|
|
|
# ---- Tests ----
|
|
|
|
class TestCollectAncestorsUnpatched:
|
|
def test_no_diamond_d0(self):
|
|
"""Single module, no parents."""
|
|
m = make_module("M")
|
|
result, calls = ancestors_unpatched(m)
|
|
assert result == []
|
|
assert calls == 0
|
|
|
|
def test_linear_chain_d2(self):
|
|
"""A <- B <- C, no diamond — linear traversal."""
|
|
a = make_module("A")
|
|
b = make_module("B")
|
|
b["parents"].append(a)
|
|
c = make_module("C")
|
|
c["parents"].append(b)
|
|
result, calls = ancestors_unpatched(c)
|
|
assert len(result) == 2 # B, A
|
|
assert calls == 2
|
|
|
|
def test_diamond_d1_visits_m0_twice(self):
|
|
"""Diamond at depth=1: M0 should be visited twice (unpatched)."""
|
|
c, m0 = build_diamond(1)
|
|
result, calls = ancestors_unpatched(c)
|
|
# M0 appears twice in the result list
|
|
assert result.count(m0) == 2
|
|
assert calls == 4 # A, M0, B, M0
|
|
|
|
def test_diamond_d2_visits_m0_four_times(self):
|
|
"""Diamond at depth=2: M0 visited 4 times (unpatched)."""
|
|
c, m0 = build_diamond(2)
|
|
result, calls = ancestors_unpatched(c)
|
|
assert result.count(m0) == 4
|
|
|
|
def test_diamond_exponential_growth(self):
|
|
"""Confirm 2^D traversal count for M0 (unpatched)."""
|
|
for d in range(1, 7):
|
|
c, m0 = build_diamond(d)
|
|
result, _ = ancestors_unpatched(c)
|
|
expected = 2 ** d
|
|
actual = result.count(m0)
|
|
assert actual == expected, (
|
|
f"depth={d}: expected {expected} visits to M0, got {actual}"
|
|
)
|
|
|
|
|
|
class TestCollectAncestorsPatched:
|
|
def test_diamond_d1_visits_m0_once(self):
|
|
"""Patched: M0 visited exactly once despite diamond."""
|
|
c, m0 = build_diamond(1)
|
|
result = ancestors_patched(c)
|
|
assert result.count(m0) == 1
|
|
|
|
def test_diamond_d5_visits_m0_once(self):
|
|
"""Patched: M0 visited exactly once even at depth=5."""
|
|
c, m0 = build_diamond(5)
|
|
result = ancestors_patched(c)
|
|
assert result.count(m0) == 1
|
|
|
|
def test_correct_unique_ancestors(self):
|
|
"""Patched: ancestor list has no duplicates."""
|
|
c, m0 = build_diamond(4)
|
|
result = ancestors_patched(c)
|
|
assert len(result) == len(set(id(x) for x in result)), (
|
|
"Patched collect_ancestors must not produce duplicate entries"
|
|
)
|
|
|
|
|
|
class TestLookupDefsUnpatched:
|
|
def test_finds_def_in_diamond(self):
|
|
"""lookup_defs finds M0.foo even in a diamond (but with redundant work)."""
|
|
c, _ = build_diamond(2)
|
|
all_defs = []
|
|
counter = [0]
|
|
lookup_defs_unpatched(c, "foo", all_defs, counter)
|
|
assert all_defs == ["M0.foo"]
|
|
|
|
def test_exponential_calls(self):
|
|
"""Unpatched lookup_defs recurses O(2^D) into M0 for deep diamonds."""
|
|
for d in range(1, 6):
|
|
c, _ = build_diamond(d)
|
|
all_defs = []
|
|
counter = [0]
|
|
lookup_defs_unpatched(c, "foo", all_defs, counter)
|
|
# calls into M0 alone = 2^D; total parent.lookup_defs calls >= 2^D
|
|
assert counter[0] >= 2 ** d, (
|
|
f"depth={d}: expected >= {2**d} recursive calls, got {counter[0]}"
|
|
)
|
|
|
|
|
|
class TestLookupDefsPatched:
|
|
def test_finds_def_in_diamond(self):
|
|
"""Patched lookup_defs still finds M0.foo correctly."""
|
|
c, _ = build_diamond(3)
|
|
all_defs = []
|
|
lookup_defs_patched(c, "foo", all_defs, visited=set())
|
|
assert all_defs == ["M0.foo"]
|
|
|
|
def test_no_def_not_found(self):
|
|
"""Patched lookup_defs returns empty when no such def."""
|
|
c, _ = build_diamond(2)
|
|
all_defs = []
|
|
lookup_defs_patched(c, "bar", all_defs, visited=set())
|
|
assert all_defs == []
|
|
|
|
def test_linear_calls_patched(self):
|
|
"""Patched lookup_defs visits each node at most once."""
|
|
for d in range(1, 8):
|
|
c, m0 = build_diamond(d)
|
|
all_defs = []
|
|
visited = set()
|
|
lookup_defs_patched(c, "foo", all_defs, visited)
|
|
# M0 must appear in visited exactly once
|
|
assert id(m0) in visited
|
|
assert all_defs == ["M0.foo"]
|
|
|
|
|
|
class TestSpeedupRatio:
|
|
def test_speedup_at_depth5(self):
|
|
"""At D=5, unpatched visits M0 32x; patched visits M0 1x."""
|
|
d = 5
|
|
c, m0 = build_diamond(d)
|
|
|
|
# Unpatched
|
|
result_u, _ = ancestors_unpatched(c)
|
|
unpatched_visits = result_u.count(m0)
|
|
|
|
# Patched
|
|
result_p = ancestors_patched(c)
|
|
patched_visits = result_p.count(m0)
|
|
|
|
assert unpatched_visits == 2 ** d # 32
|
|
assert patched_visits == 1
|
|
ratio = unpatched_visits / patched_visits
|
|
assert ratio >= 32, f"Expected >= 32x ratio, got {ratio}x"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v"])
|