russell@unturf.com
0a580b313d
Authors: russell@unturf.com · brackishbert@gmail.com · foxhop.net · TimeHexOn.com Patches, unit tests, benchmarks, whitepaper, and outreach briefs. Public domain — no copyright claimed. Use freely.
198 lines
7.3 KiB
Python
198 lines
7.3 KiB
Python
"""
|
|
Unit test for luigi-0001: dfs_paths path-set rebuild CWE-407.
|
|
|
|
Defect: dfs_paths rebuilds set(path) — O(depth) — on every recursive call.
|
|
Total work: O(D²) where D = dependency chain depth.
|
|
|
|
Fix: carry a parallel path_set so each call pays O(1) set difference.
|
|
|
|
Measurement: instrument the defective path rebuild directly, counting the
|
|
total elements iterated during each set(path) construction.
|
|
"""
|
|
|
|
import unittest
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Minimal stub tasks — no real Luigi needed
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class FakeTask:
|
|
def __init__(self, name, requires=None):
|
|
self.task_family = name
|
|
self._requires = requires or []
|
|
|
|
def __hash__(self): return hash(self.task_family)
|
|
def __eq__(self, o): return isinstance(o, FakeTask) and self.task_family == o.task_family
|
|
def __repr__(self): return f"Task({self.task_family})"
|
|
|
|
|
|
def get_task_requires(task):
|
|
return frozenset(task._requires) # frozenset avoids set() call measurement noise
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DEFECTIVE — instruments the set(path) rebuild cost explicitly
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def dfs_paths_defective(start_task, goal_task_family, path=None, _work=None):
|
|
if path is None:
|
|
path = [start_task]
|
|
if _work is None:
|
|
_work = [0]
|
|
if start_task.task_family == goal_task_family or goal_task_family is None:
|
|
for item in path:
|
|
yield item
|
|
# Measure the O(depth) rebuild: len(path) elements iterated to build set
|
|
_work[0] += len(path)
|
|
for nxt in get_task_requires(start_task) - set(path):
|
|
for t in dfs_paths_defective(nxt, goal_task_family, path + [nxt], _work):
|
|
yield t
|
|
|
|
|
|
def run_defective(root, goal):
|
|
"""Returns (results, total_rebuild_work)."""
|
|
work = [0]
|
|
results = list(dfs_paths_defective(root, goal, _work=work))
|
|
return results, work[0]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# FIXED — path_set maintained alongside path, no set(path) rebuild
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def dfs_paths_fixed(start_task, goal_task_family, path=None, path_set=None, _work=None):
|
|
if path is None:
|
|
path = [start_task]
|
|
path_set = {start_task}
|
|
if _work is None:
|
|
_work = [0]
|
|
if start_task.task_family == goal_task_family or goal_task_family is None:
|
|
for item in path:
|
|
yield item
|
|
# O(1) set difference — cost is 1 regardless of depth
|
|
_work[0] += 1
|
|
for nxt in get_task_requires(start_task) - path_set:
|
|
for t in dfs_paths_fixed(nxt, goal_task_family,
|
|
path + [nxt], path_set | {nxt}, _work):
|
|
yield t
|
|
|
|
|
|
def run_fixed(root, goal):
|
|
work = [0]
|
|
results = list(dfs_paths_fixed(root, goal, _work=work))
|
|
return results, work[0]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helper: build a linear chain T0 → T1 → T2 → ... → T(D-1)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def linear_chain(depth):
|
|
tasks = [FakeTask(f"T{i}") for i in range(depth)]
|
|
for i in range(depth - 1):
|
|
tasks[i]._requires = [tasks[i + 1]]
|
|
return tasks[0], tasks[-1]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestLuigi0001DfsPaths(unittest.TestCase):
|
|
|
|
def test_correctness_both_match(self):
|
|
root, _ = linear_chain(8)
|
|
res_def, _ = run_defective(root, None)
|
|
res_fix, _ = run_fixed(root, None)
|
|
self.assertEqual(
|
|
{t.task_family for t in res_def},
|
|
{t.task_family for t in res_fix},
|
|
)
|
|
print("PASS test_correctness_both_match")
|
|
|
|
def test_correctness_goal_family(self):
|
|
root, leaf = linear_chain(6)
|
|
res_fix, _ = run_fixed(root, leaf.task_family)
|
|
families = {t.task_family for t in res_fix}
|
|
self.assertIn(leaf.task_family, families)
|
|
self.assertIn(root.task_family, families)
|
|
print("PASS test_correctness_goal_family")
|
|
|
|
def test_defective_work_grows_quadratically(self):
|
|
"""
|
|
Rebuild cost for defective = sum(1..D) = D*(D+1)/2 — O(D²).
|
|
When depth doubles, work should grow ~4x.
|
|
"""
|
|
depths = [8, 16, 32]
|
|
works = []
|
|
for d in depths:
|
|
root, _ = linear_chain(d)
|
|
_, w = run_defective(root, None)
|
|
works.append(w)
|
|
|
|
ratio_0_1 = works[1] / works[0]
|
|
ratio_1_2 = works[2] / works[1]
|
|
self.assertGreater(ratio_0_1, 3.0,
|
|
f"defective work should grow >3x when depth doubles; got {ratio_0_1:.2f}")
|
|
self.assertGreater(ratio_1_2, 3.0,
|
|
f"defective work should grow >3x when depth doubles; got {ratio_1_2:.2f}")
|
|
print(f"PASS test_defective_work_grows_quadratically "
|
|
f"works={works} ratios=({ratio_0_1:.2f}, {ratio_1_2:.2f})")
|
|
|
|
def test_fixed_work_grows_linearly(self):
|
|
"""
|
|
Fixed cost = 1 per node = D — O(D).
|
|
When depth doubles, work should grow ~2x.
|
|
"""
|
|
depths = [8, 16, 32]
|
|
works = []
|
|
for d in depths:
|
|
root, _ = linear_chain(d)
|
|
_, w = run_fixed(root, None)
|
|
works.append(w)
|
|
|
|
ratio_0_1 = works[1] / works[0]
|
|
ratio_1_2 = works[2] / works[1]
|
|
self.assertAlmostEqual(ratio_0_1, 2.0, delta=0.1,
|
|
msg=f"fixed work should grow ~2x when depth doubles; got {ratio_0_1:.2f}")
|
|
self.assertAlmostEqual(ratio_1_2, 2.0, delta=0.1,
|
|
msg=f"fixed work should grow ~2x when depth doubles; got {ratio_1_2:.2f}")
|
|
print(f"PASS test_fixed_work_grows_linearly works={works}")
|
|
|
|
def test_ratio_at_depth_32(self):
|
|
"""
|
|
At depth=32, defective does 32*33/2 = 528 units of work.
|
|
Fixed does 32. Ratio ≈ 16.5.
|
|
"""
|
|
root, _ = linear_chain(32)
|
|
_, def_w = run_defective(root, None)
|
|
_, fix_w = run_fixed(root, None)
|
|
ratio = def_w / fix_w
|
|
self.assertGreater(ratio, 10,
|
|
f"at depth=32, defective should be >10x more work; ratio={ratio:.1f}")
|
|
print(f"PASS test_ratio_at_depth_32 "
|
|
f"defective={def_w}, fixed={fix_w}, ratio={ratio:.1f}x")
|
|
|
|
def test_exact_defective_work(self):
|
|
"""Defective work = D*(D+1)/2 exactly (sum of 1+2+...+D)."""
|
|
for d in [4, 8, 16]:
|
|
root, _ = linear_chain(d)
|
|
_, w = run_defective(root, None)
|
|
expected = d * (d + 1) // 2
|
|
self.assertEqual(w, expected,
|
|
f"depth={d}: expected {expected} units of work, got {w}")
|
|
print("PASS test_exact_defective_work")
|
|
|
|
def test_exact_fixed_work(self):
|
|
"""Fixed work = D exactly (one O(1) op per node)."""
|
|
for d in [4, 8, 16]:
|
|
root, _ = linear_chain(d)
|
|
_, w = run_fixed(root, None)
|
|
self.assertEqual(w, d,
|
|
f"depth={d}: expected {d} units of work, got {w}")
|
|
print("PASS test_exact_fixed_work")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|