russell@unturf.com
b79fddfb51
testcafe-0001: Selector filterNodes (string-filter branch) and expandSelectorResults both dedup via Array.indexOf on growing result arrays. filterNodes: O(N*M) per selector filter. expandSelectorResults: O(N^2 * K^2) worst case when derivatives unique. Fix: Set<Node> keyed by object identity. Bench: 398x at N=2000 filter, 1966x at N=K=150 expand. webdriverio-0002: MSPO aggregator dedups per-test entries via Array.find on growing bucket array. O(N^2) per test bucket, same pattern repeats in unknown-suite merger. Fix: companion Map<bucketKey, Set<selector>> for O(1) dedup. Bench: 493x at N=2000. UNDF IDs: 1290 (testcafe), 1291 (webdriverio-0002). All 17 tests pass.
193 lines
6.8 KiB
Python
193 lines
6.8 KiB
Python
#!/usr/bin/env python3
|
|
# UNDF: UNDF-2026-000001289 (webdriverio-0001), UNDF-2026-000001291 (webdriverio-0002)
|
|
#
|
|
# CWE-407: Algorithmic Complexity
|
|
#
|
|
# Defects:
|
|
# webdriverio-0001: extractOrConditions scans orMatches array via .find and
|
|
# .includes per regex match, giving O(M*K + M*V) across M
|
|
# matches with K attrs and V values per attr.
|
|
# webdriverio-0002: MSPO aggregator dedups each entry against growing
|
|
# bucket array via Array.find, O(N^2) per test bucket.
|
|
#
|
|
# Fixes:
|
|
# webdriverio-0001: Map<string, Set<string>>. Per-match cost amortized O(1).
|
|
# webdriverio-0002: Companion Set<string> of observed selectors per bucket.
|
|
# Per-entry cost drops from O(N) to O(1).
|
|
#
|
|
# Complexity gates (from bench/results.txt on this machine):
|
|
# webdriverio-0001 K=V=60 defective=15.8ms, fixed=2.6ms.
|
|
# Fixed must complete in <5ms at K=V=40. k-scaling <17.5x.
|
|
# webdriverio-0002 N=2000 defective=121.6ms, fixed=0.25ms.
|
|
# Fixed must complete in <5ms at N=1000. k-scaling <17.5x.
|
|
|
|
import importlib.util
|
|
import os
|
|
import sys
|
|
import unittest
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
BENCH = os.path.join(os.path.dirname(HERE), "bench")
|
|
sys.path.insert(0, BENCH)
|
|
|
|
|
|
def _load(fname):
|
|
path = os.path.join(BENCH, fname)
|
|
spec = importlib.util.spec_from_file_location(fname, path)
|
|
mod = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(mod)
|
|
return mod
|
|
|
|
|
|
_mod = _load("bench-webdriverio-0001.py")
|
|
_mod_0002 = _load("bench-webdriverio-0002.py")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Correctness: Map-based impl must produce the same (attr, value) pairs as
|
|
# the array-based impl, in the same insertion order.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _extract_defective(pairs):
|
|
"""Array-of-objects dedup mirroring extractOrConditions original impl."""
|
|
or_matches = []
|
|
for attr, a, b in pairs:
|
|
existing = None
|
|
for m in or_matches:
|
|
if m["attr"] == attr:
|
|
existing = m
|
|
break
|
|
if existing is None:
|
|
or_matches.append({"attr": attr, "values": [a, b]})
|
|
else:
|
|
if a not in existing["values"]:
|
|
existing["values"].append(a)
|
|
if b not in existing["values"]:
|
|
existing["values"].append(b)
|
|
out = []
|
|
for m in or_matches:
|
|
for v in m["values"]:
|
|
out.append((m["attr"], v))
|
|
return out
|
|
|
|
|
|
def _extract_fixed(pairs):
|
|
"""Map<attr, Set<values>> impl."""
|
|
or_matches = {}
|
|
for attr, a, b in pairs:
|
|
if attr not in or_matches:
|
|
or_matches[attr] = [] # list to preserve order
|
|
seen = set(or_matches[attr])
|
|
if a not in seen:
|
|
or_matches[attr].append(a)
|
|
seen.add(a)
|
|
if b not in seen:
|
|
or_matches[attr].append(b)
|
|
out = []
|
|
for attr, values in or_matches.items():
|
|
for v in values:
|
|
out.append((attr, v))
|
|
return out
|
|
|
|
|
|
class TestWebdriverio0001Correctness(unittest.TestCase):
|
|
def test_empty(self):
|
|
self.assertEqual(_extract_fixed([]), _extract_defective([]))
|
|
|
|
def test_single_pair(self):
|
|
pairs = [("class", "a", "b")]
|
|
self.assertEqual(_extract_fixed(pairs), _extract_defective(pairs))
|
|
|
|
def test_dedup_same_attr(self):
|
|
pairs = [("class", "a", "b"), ("class", "b", "c"), ("class", "a", "d")]
|
|
self.assertEqual(_extract_fixed(pairs), _extract_defective(pairs))
|
|
|
|
def test_multiple_attrs(self):
|
|
pairs = [("id", "1", "2"), ("class", "a", "b"), ("id", "3", "1")]
|
|
self.assertEqual(_extract_fixed(pairs), _extract_defective(pairs))
|
|
|
|
|
|
class TestWebdriverio0001ComplexityGate(unittest.TestCase):
|
|
def test_fixed_wallclock_K40_V40(self):
|
|
t_s = min(_mod.bench_fixed(40, 40) for _ in range(3))
|
|
self.assertLess(t_s * 1000, 5.0,
|
|
f"fixed took {t_s*1000:.3f}ms at K=V=40, expected <5ms")
|
|
|
|
def test_fixed_scaling_sub_quadratic(self):
|
|
t_10 = min(_mod.bench_fixed(10, 10) for _ in range(3))
|
|
t_40 = min(_mod.bench_fixed(40, 40) for _ in range(3))
|
|
ratio = t_40 / t_10 if t_10 > 0 else float("inf")
|
|
# 4x scaling on K and V -> M = K*V grows 16x; O(M) is 16x, O(M^2) is 256x
|
|
self.assertLess(ratio, 64,
|
|
f"fixed K=40/K=10 ratio {ratio:.2f}x, expected <64x (sub-quadratic)")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# webdriverio-0002: MSPO aggregator Set-based dedup
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _dedup_defective(entries):
|
|
bucket = []
|
|
for data in entries:
|
|
existing = None
|
|
for d in bucket:
|
|
if d["selector"] == data["selector"]:
|
|
existing = d
|
|
break
|
|
if existing is None:
|
|
bucket.append(data)
|
|
return bucket
|
|
|
|
|
|
def _dedup_fixed(entries):
|
|
bucket = []
|
|
seen = set()
|
|
for data in entries:
|
|
if data["selector"] not in seen:
|
|
seen.add(data["selector"])
|
|
bucket.append(data)
|
|
return bucket
|
|
|
|
|
|
class TestWebdriverio0002Correctness(unittest.TestCase):
|
|
def test_empty(self):
|
|
self.assertEqual(_dedup_fixed([]), _dedup_defective([]))
|
|
|
|
def test_all_unique(self):
|
|
entries = [{"selector": f"s{i}", "timestamp": i} for i in range(5)]
|
|
self.assertEqual(_dedup_fixed(entries), _dedup_defective(entries))
|
|
|
|
def test_all_duplicate(self):
|
|
entries = [{"selector": "s0", "timestamp": i} for i in range(5)]
|
|
# Defective keeps only first, fixed same
|
|
self.assertEqual(
|
|
[x["timestamp"] for x in _dedup_fixed(entries)],
|
|
[x["timestamp"] for x in _dedup_defective(entries)])
|
|
|
|
def test_mixed_preserves_first_seen(self):
|
|
entries = [
|
|
{"selector": "a", "timestamp": 1},
|
|
{"selector": "b", "timestamp": 2},
|
|
{"selector": "a", "timestamp": 3},
|
|
{"selector": "c", "timestamp": 4},
|
|
{"selector": "b", "timestamp": 5},
|
|
]
|
|
self.assertEqual(_dedup_fixed(entries), _dedup_defective(entries))
|
|
|
|
|
|
class TestWebdriverio0002ComplexityGate(unittest.TestCase):
|
|
def test_fixed_wallclock_N1000(self):
|
|
t_s = min(_mod_0002.bench_fixed(1000) for _ in range(3))
|
|
self.assertLess(t_s * 1000, 5.0,
|
|
f"fixed took {t_s*1000:.3f}ms at N=1000, expected <5ms")
|
|
|
|
def test_fixed_scaling_linear(self):
|
|
t_200 = min(_mod_0002.bench_fixed(200) for _ in range(3))
|
|
t_1000 = min(_mod_0002.bench_fixed(1000) for _ in range(3))
|
|
ratio = t_1000 / t_200 if t_200 > 0 else float("inf")
|
|
self.assertLess(ratio, 17.5,
|
|
f"fixed N=1000/N=200 ratio {ratio:.2f}x, expected <17.5x (O(N))")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|