|
| 1 | +"""Tests for subtree-population crossover (docs/web_research_port_candidates_2026-08.md #8). |
| 2 | +
|
| 3 | +Port of GRIIN (ASE '23) / Grammarinator x AFL++ (2026): grammar-aware tree |
| 4 | +crossover should be able to splice in a subtree harvested from a *different* |
| 5 | +corpus entry, not just regenerate one from scratch or clone within the same |
| 6 | +tree. ``TreeMutator``'s docstring already promised this as op 4 |
| 7 | +("Subtree splice") but the implementation never existed until now. |
| 8 | +""" |
| 9 | + |
| 10 | +import random |
| 11 | + |
| 12 | +from fuzzer_tool.core.grammar import Grammar, SubtreePopulation, TreeMutator |
| 13 | +from fuzzer_tool.services.operators import OperatorEngine |
| 14 | + |
| 15 | +from .support.operator_env import make_minimal_fuzzer |
| 16 | + |
| 17 | + |
| 18 | +def _json_grammar() -> Grammar: |
| 19 | + g = Grammar() |
| 20 | + g.parse('root = {"key":"value"}') |
| 21 | + return g |
| 22 | + |
| 23 | + |
| 24 | +class TestSubtreePopulation: |
| 25 | + def test_sample_unseen_rule_returns_none(self): |
| 26 | + pop = SubtreePopulation() |
| 27 | + assert pop.sample("value") is None |
| 28 | + |
| 29 | + def test_add_harvests_interior_nodes(self): |
| 30 | + tm = TreeMutator(_json_grammar()) |
| 31 | + tree = tm.parse(b'{"key":"value"}') |
| 32 | + pop = SubtreePopulation() |
| 33 | + pop.add(tree, rng=random.Random(1)) |
| 34 | + assert len(pop) > 0 |
| 35 | + assert pop.sample("root") is not None |
| 36 | + |
| 37 | + @staticmethod |
| 38 | + def _interior_node(rule: str, marker: bytes): |
| 39 | + """A one-node-deep interior node: ``collect_interior`` only reports |
| 40 | + non-leaf nodes, so a bare ``TreeNode(rule=..., data=...)`` (a leaf) |
| 41 | + would never be harvested.""" |
| 42 | + from fuzzer_tool.core.grammar import TreeNode |
| 43 | + |
| 44 | + return TreeNode(rule=rule, children=[TreeNode(rule="leaf", data=marker)]) |
| 45 | + |
| 46 | + def test_reservoir_bounded_by_max_per_rule(self): |
| 47 | + """Falsification: harvesting more nodes than the cap must never |
| 48 | + grow the pool past ``max_per_rule`` for a single rule.""" |
| 49 | + pop = SubtreePopulation(max_per_rule=4) |
| 50 | + rng = random.Random(7) |
| 51 | + for i in range(200): |
| 52 | + pop.add(self._interior_node("value", str(i).encode()), rng=rng) |
| 53 | + assert len(pop._pools["value"]) == 4 |
| 54 | + |
| 55 | + def test_reservoir_sampling_reaches_late_items(self): |
| 56 | + """Adversarial: an item harvested long after the pool filled up |
| 57 | + must still have a nonzero chance of surviving eviction — a buggy |
| 58 | + reservoir (e.g. only replacing index 0) would never let it in.""" |
| 59 | + seen_late_item = False |
| 60 | + for trial in range(200): |
| 61 | + pop = SubtreePopulation(max_per_rule=4) |
| 62 | + rng = random.Random(trial) |
| 63 | + for i in range(50): |
| 64 | + pop.add(self._interior_node("value", str(i).encode()), rng=rng) |
| 65 | + if any(n.children[0].data == b"49" for n in pop._pools["value"]): |
| 66 | + seen_late_item = True |
| 67 | + break |
| 68 | + assert seen_late_item, "item #49 never survived reservoir sampling across 200 trials" |
| 69 | + |
| 70 | + |
| 71 | +class TestTreeSplice: |
| 72 | + def test_splice_falls_back_without_population(self): |
| 73 | + """No population supplied -> behaves like a plain subtree swap, |
| 74 | + never raises and always returns bytes that respect max_len.""" |
| 75 | + tm = TreeMutator(_json_grammar()) |
| 76 | + tm._rng = random.Random(0) |
| 77 | + tree = tm.parse(b'{"key":"value"}') |
| 78 | + result = tm._tree_splice(tree, max_len=64, population=None) |
| 79 | + assert isinstance(result, bytes) |
| 80 | + assert len(result) <= 64 |
| 81 | + |
| 82 | + def test_splice_falls_back_on_empty_population(self): |
| 83 | + tm = TreeMutator(_json_grammar()) |
| 84 | + tm._rng = random.Random(0) |
| 85 | + tree = tm.parse(b'{"key":"value"}') |
| 86 | + result = tm._tree_splice(tree, max_len=64, population=SubtreePopulation()) |
| 87 | + assert isinstance(result, bytes) |
| 88 | + |
| 89 | + def test_splice_grafts_donor_subtree(self): |
| 90 | + """Splicing from a population seeded with a donor tree must be able |
| 91 | + to pull in bytes that never appeared in the original tree.""" |
| 92 | + grammar = _json_grammar() |
| 93 | + tm = TreeMutator(grammar) |
| 94 | + target_tree = tm.parse(b'{"key":"value"}') |
| 95 | + donor_tree = tm.parse(b'{"other":"DONOR_MARKER_XYZ"}') |
| 96 | + |
| 97 | + pop = SubtreePopulation() |
| 98 | + pop.add(donor_tree, rng=random.Random(3)) |
| 99 | + |
| 100 | + found_donor_bytes = False |
| 101 | + for seed in range(64): |
| 102 | + tm._rng = random.Random(seed) |
| 103 | + clone = tm._clone_tree(target_tree) |
| 104 | + out = tm._tree_splice(clone, max_len=4096, population=pop) |
| 105 | + if b"DONOR_MARKER_XYZ" in out or b"other" in out: |
| 106 | + found_donor_bytes = True |
| 107 | + break |
| 108 | + assert found_donor_bytes, "splice never grafted in bytes from the donor tree" |
| 109 | + |
| 110 | + def test_mutate_tree_op3_is_splice(self): |
| 111 | + """The op-3 branch in mutate_tree must route to _tree_splice, not |
| 112 | + silently stay a no-op (regression for the missing implementation).""" |
| 113 | + grammar = _json_grammar() |
| 114 | + tm = TreeMutator(grammar) |
| 115 | + tree = tm.parse(b'{"key":"value"}') |
| 116 | + donor_tree = tm.parse(b'{"other":"DONOR_MARKER_XYZ"}') |
| 117 | + pop = SubtreePopulation() |
| 118 | + pop.add(donor_tree, rng=random.Random(3)) |
| 119 | + |
| 120 | + class _FixedOpRng: |
| 121 | + """Forces mutate_tree's op selection to pick splice (op index 3).""" |
| 122 | + |
| 123 | + def randint(self, a, b): |
| 124 | + if b == 5: # the op-selection roll in mutate_tree |
| 125 | + return 3 |
| 126 | + return random.Random(0).randint(a, b) |
| 127 | + |
| 128 | + found_donor_bytes = False |
| 129 | + for _ in range(32): |
| 130 | + clone = tm._clone_tree(tree) |
| 131 | + out = tm.mutate_tree(clone, max_len=4096, rng=_FixedOpRng(), population=pop) |
| 132 | + if b"DONOR_MARKER_XYZ" in out or b"other" in out: |
| 133 | + found_donor_bytes = True |
| 134 | + break |
| 135 | + assert found_donor_bytes |
| 136 | + |
| 137 | + |
| 138 | +class TestGrammarTreeMutateOperatorWiring: |
| 139 | + def _engine_with_grammar(self, corpus): |
| 140 | + f = make_minimal_fuzzer(seed=0x5EED) |
| 141 | + f.grammar = _json_grammar() |
| 142 | + f.corpus = corpus |
| 143 | + return OperatorEngine(f) |
| 144 | + |
| 145 | + def test_builds_and_reuses_population_across_calls(self): |
| 146 | + corpus = [b'{"a":"seed_one"}', b'{"b":"seed_two_marker"}'] |
| 147 | + engine = self._engine_with_grammar(corpus) |
| 148 | + buf = bytearray(b'{"key":"value"}') |
| 149 | + engine._op_grammar_tree_mutate(buf, 0, bytes(buf)) |
| 150 | + f = engine.f |
| 151 | + assert hasattr(f, "_subtree_population") |
| 152 | + assert len(f._subtree_population) > 0 |
| 153 | + assert f._subtree_pop_next_idx == len(corpus) |
| 154 | + |
| 155 | + # Corpus growth is picked up incrementally on the next call. |
| 156 | + corpus.append(b'{"c":"seed_three"}') |
| 157 | + engine._op_grammar_tree_mutate(buf, 0, bytes(buf)) |
| 158 | + assert f._subtree_pop_next_idx == len(corpus) |
| 159 | + |
| 160 | + def test_grammar_tree_mutate_returns_bytes(self): |
| 161 | + corpus = [b'{"a":"seed_one"}'] |
| 162 | + engine = self._engine_with_grammar(corpus) |
| 163 | + buf = bytearray(b'{"key":"value"}') |
| 164 | + result = engine._op_grammar_tree_mutate(buf, 0, bytes(buf)) |
| 165 | + assert isinstance(result, bytearray) |
| 166 | + |
| 167 | + def test_no_grammar_returns_none(self): |
| 168 | + """Adversarial: without a grammar the op must be a clean no-op, |
| 169 | + never touching the (nonexistent) population machinery.""" |
| 170 | + f = make_minimal_fuzzer(seed=1) |
| 171 | + f.grammar = None |
| 172 | + engine = OperatorEngine(f) |
| 173 | + buf = bytearray(b"whatever") |
| 174 | + assert engine._op_grammar_tree_mutate(buf, 0, bytes(buf)) is None |
0 commit comments