Files
troupe-standup/wizard-work/null/chunktext/test_chunktext.py
T

593 lines
24 KiB
Python

#!/usr/bin/env python3
"""test_chunktext.py -- the verification harness for chunktext.py.
Exit 0 == every claim below was observed to hold.
Exit 1 == at least one claim was observed to fail.
Exit 2 == the harness itself broke.
Every assertion here is checked against real execution of the library. The
harness also carries a *teeth* stage: it runs the same suite against a
deliberately broken copy of the library and requires the suite to FAIL. A
harness that cannot fail cannot verify anything, so that is part of the result.
python3 test_chunktext.py # run, exit 0/1/2
python3 test_chunktext.py --verbose
"""
from __future__ import annotations
import importlib.util
import json
import os
import random
import re
import sys
import tempfile
import traceback
HERE = os.path.dirname(os.path.abspath(__file__))
LIB = os.path.join(HERE, "chunktext.py")
CHECKS = 0
FAILURES: list[str] = []
def check(cond, label: str, detail: str = "") -> None:
global CHECKS
CHECKS += 1
if not cond:
FAILURES.append(f"{label}{' :: ' + detail if detail else ''}")
print(f"[FAIL] {label}" + (f"\n {detail}" if detail else ""))
elif "--verbose" in sys.argv:
print(f"[ ok ] {label}")
# ---------------------------------------------------------------------------
# fixtures
# ---------------------------------------------------------------------------
def _fixtures() -> dict[str, str]:
return {
"plain": " ".join(f"Sentence number {i}." for i in range(50)),
"single_short": "Hello world.",
"single_word": "Supercalifragilisticexpialidocious",
"repetitive": "ab " * 400,
"no_spaces": "x" * 5000,
"unicode": " ".join(f"Café résumé naïve {i}." for i in range(30)),
"paragraphs": "Para one here.\n\nPara two here.\n\nPara three here.",
"newlines": "line1\nline2\nline3\nline4",
"tabs": "col1\tcol2\tcol3\tcol4\tcol5\tcol6\tcol7\tcol8",
"no_space_sentences": "One.Two.Three.Four.Five.Six.Seven.Eight.Nine.Ten.",
"trailing_ws": "Hello world. ",
"leading_ws": " Hello world.",
"sentinel_in_text": "hello there. \x00OVL\x00 more text here. and more.",
"double_sentinel": "\x00OVL\x00 a \x00OVL\x00 b \x00OVL\x00 c",
"long_word_and_text": "word " * 200 + "Supercalifragilisticexpialidocious" * 5,
"cjk": "这是一个测试。这是第二个句子。这是第三个句子。",
"mixed_ws": "a b\tc\nd\r\ne\x0bf\x0cg",
"punctuation_heavy": "a/b, c; d: e! f? g-h i. j/k, l; m: n!",
"one_char": "a",
"single_space": " ",
"empty": "",
"only_newlines": "\n\n\n\n",
"html": "<p>Hello</p>\n<p>World</p>\n<div>x</div>",
}
BUDGETS = [(1, 0), (2, 0), (2, 1), (4, 0), (4, 1), (4, 2), (8, 2), (8, 4),
(16, 4), (16, 8), (32, 16), (64, 8), (256, 32)]
# ---------------------------------------------------------------------------
# stages
# ---------------------------------------------------------------------------
def stage_roundtrip(mod) -> None:
"""The headline claim: reconstruct(chunk_text(t)) == t, for every fixture
and every budget, including the pathological ones."""
fx = _fixtures()
cases = 0
for name, text in fx.items():
for mt, ov in BUDGETS:
if ov > mt:
continue
cases += 1
chunks = mod.chunk_text(text, mt, ov)
got = mod.reconstruct(chunks)
if got != text:
check(False, f"roundtrip {name} mt={mt} ov={ov}",
f"want {text[:70]!r} got {got[:70]!r}")
check(cases > 0, "roundtrip: cases executed", str(cases))
check(not any(f.startswith("roundtrip") for f in FAILURES),
f"roundtrip: exact for all {cases} fixture/budget combinations")
def stage_bound(mod) -> None:
"""Every emitted chunk's est_tokens must be <= max_tokens."""
fx = _fixtures()
cases = 0
for name, text in fx.items():
for mt, ov in BUDGETS:
if ov > mt:
continue
cases += 1
chunks = mod.chunk_text(text, mt, ov)
for c in chunks:
if c["est_tokens"] > mt:
check(False, f"bound {name} mt={mt}",
f"chunk {c['index']} est_tokens={c['est_tokens']} > {mt}")
check(not any(f.startswith("bound") for f in FAILURES),
f"bound: every chunk within max_tokens across {cases} combinations")
def stage_overlap(mod) -> None:
"""The overlap must be real, bounded, and never enlarge the chunk.
The library's contract is: a chunk's ``overlap`` is the longest suffix of
the previous chunk that is also a prefix of this one, capped at
``overlap_tokens * CHARS_PER_TOKEN``. That is checked here directly against
the definition, not against a guess.
"""
texts = {
"plain": " ".join(f"alpha beta gamma delta {i}." for i in range(60)),
"repetitive": _fixtures()["repetitive"],
"long_word": _fixtures()["long_word_and_text"],
}
for name, text in texts.items():
for mt, ov in [(8, 2), (16, 4), (32, 16), (64, 8)]:
chunks = mod.chunk_text(text, mt, ov)
for i, c in enumerate(chunks):
if i == 0:
check(c["overlap"] == "", f"overlap first chunk empty {name}")
continue
prev_text = chunks[i - 1]["text"]
want = mod._max_overlap_len(prev_text, c["text"])
cap = ov * mod.CHARS_PER_TOKEN
want = min(want, cap)
if c["overlap"]:
check(c["text"].startswith(c["overlap"]),
f"overlap is a real prefix {name} mt={mt}",
repr(c["overlap"]))
check(c["overlap"] == c["text"][:len(c["overlap"])],
f"overlap is the chunk's own prefix {name} mt={mt}")
check(len(c["overlap"]) <= cap,
f"overlap within budget {name} mt={mt}",
f"{len(c['overlap'])} > {cap}")
check(len(c["overlap"]) <= want,
f"overlap is maximal-or-capped {name} mt={mt}",
f"got {len(c['overlap'])} want <= {want}")
check(c["est_tokens"] <= mt,
f"overlap does not enlarge chunk {name} mt={mt}")
m = mod.measure_overlap(chunks, mt, ov)
check(m["within_budget"], f"measure_overlap within budget {name}")
check(m["max_overlap_chars"] <= ov * mod.CHARS_PER_TOKEN,
f"measure_overlap cap respected {name} mt={mt}")
# A concrete, human-checkable case: two chunks over repetitive text must
# report a non-empty overlap, because "ab " really does repeat.
chunks = mod.chunk_text("ab " * 400, 8, 2)
any_ov = any(c["overlap"] for c in chunks)
check(any_ov, "overlap: repetitive text produces a non-empty overlap")
m = mod.measure_overlap(chunks, 8, 2)
check(m["chunks_with_overlap"] > 0,
"overlap: measure_overlap sees the repetition")
check(m["mean_overlap_chars"] > 0,
"overlap: mean overlap is positive on repetitive text",
str(m))
def stage_structure(mod) -> None:
"""Chunks must tile the source: contiguous char ranges, correct indices."""
text = " ".join(f"token{i} " for i in range(300))
for mt, ov in BUDGETS:
if ov > mt:
continue
chunks = mod.chunk_text(text, mt, ov)
cursor = 0
for i, c in enumerate(chunks):
check(c["index"] == i, f"index sequential mt={mt}")
check(c["char_start"] == cursor, f"char_start contiguous mt={mt}")
check(c["char_end"] == cursor + len(c["text"]),
f"char_end matches text length mt={mt}")
check(c["est_tokens"] == mod.estimate_tokens(c["text"]),
f"est_tokens consistent mt={mt}")
cursor = c["char_end"]
check(cursor == len(text), f"tiling covers whole source mt={mt}",
f"{cursor} != {len(text)}")
def stage_json(mod) -> None:
"""A chunk list must survive a JSON round-trip unchanged."""
text = _fixtures()["long_word_and_text"]
chunks = mod.chunk_text(text, 16, 4)
blob = json.dumps(chunks)
back = json.loads(blob)
check(back == chunks, "json: chunk list survives a round-trip")
check(mod.reconstruct(back) == text, "json: reconstructed text survives")
check(json.dumps(chunks) == json.dumps(mod.chunk_text(text, 16, 4)),
"determinism: same input twice gives identical JSON")
def stage_bare_strings(mod) -> None:
"""reconstruct must also accept the bare-string shape."""
text = " ".join(f"word {i}." for i in range(40))
chunks = mod.chunk_text(text, 8, 2)
bare = [c["text"] for c in chunks]
check(mod.reconstruct(bare) == text, "reconstruct accepts bare strings")
def stage_validation(mod) -> None:
"""Bad arguments must raise, not silently return something."""
for label, args, exc in [
("max_tokens=0", ("hello", 0, 0), (ValueError,)),
("max_tokens=-5", ("hello", -5, 0), (ValueError,)),
("overlap_tokens=-1", ("hello", 8, -1), (ValueError,)),
("overlap > max", ("hello", 4, 8), (ValueError,)),
]:
try:
r = mod.chunk_text(*args)
check(False, f"validation: {label} raises", f"returned {r!r}")
except exc:
check(True, f"validation: {label} raises {exc[0].__name__}")
except Exception as e: # wrong exception type
check(False, f"validation: {label} raises the right type",
f"got {type(e).__name__}: {e}")
for label, arg in [("None", None), ("int", 42), ("bytes", b"hi")]:
try:
mod.chunk_text(arg, 8, 2)
check(False, f"validation: {label} text raises TypeError")
except TypeError:
check(True, f"validation: {label} text raises TypeError")
except Exception as e:
check(False, f"validation: {label} raises TypeError",
f"got {type(e).__name__}")
def stage_empty(mod) -> None:
"""Truly empty input yields no chunks. Whitespace-only input is content and
must round-trip exactly (it is not 'empty')."""
chunks = mod.chunk_text("", 8, 2)
check(chunks == [], "empty: '' -> []")
check(mod.reconstruct(chunks) == "", "empty: '' reconstructs to ''")
for label, text in [("spaces", " "), ("newlines", "\n\n\n"),
("tabs", "\t\t\t"), ("mixed", " \n\t ")]:
chunks = mod.chunk_text(text, 8, 2)
check(mod.reconstruct(chunks) == text,
f"empty: whitespace-only {label} round-trips exactly",
f"got {mod.reconstruct(chunks)!r}")
def stage_sentinel(mod) -> None:
"""Text containing the library's own sentinel must round-trip."""
for name in ("sentinel_in_text", "double_sentinel"):
text = _fixtures()[name]
for mt, ov in [(8, 2), (16, 4), (4, 1)]:
chunks = mod.chunk_text(text, mt, ov)
check(mod.reconstruct(chunks) == text,
f"sentinel: {name} mt={mt} ov={ov}")
# And the sentinel must not be relied on as an escape mechanism.
src = open(LIB, "r", encoding="utf-8").read()
check("_OVERLAP_SENTINEL" not in src,
"sentinel: library does not encode anything with a sentinel")
def stage_realistic(mod) -> None:
"""A realistic multi-paragraph document, with markdown and code fences."""
text = (
"# Title\n\n"
"The quick brown fox jumps over the lazy dog. " * 40
+ "\n\n## Section\n\n"
+ "- item one\n- item two\n- item three\n\n"
+ "```python\ndef f(x):\n return x * 2\n```\n\n"
+ "Final paragraph. " * 30
)
for mt, ov in [(16, 4), (64, 16), (256, 32)]:
chunks = mod.chunk_text(text, mt, ov)
check(mod.reconstruct(chunks) == text, f"realistic: roundtrip mt={mt}")
check(all(c["est_tokens"] <= mt for c in chunks),
f"realistic: bound mt={mt}")
cov = mod.coverage(chunks, mt)
check(cov["contiguous"], f"realistic: contiguous mt={mt}")
def stage_fuzz(mod) -> None:
"""Randomised inputs: the invariants must hold on arbitrary text."""
rng = random.Random(20261023)
alphabet = "abc \n\t\x00.,;:!?-\u00e9\u4e2d\u2014<>/()[]{}'\""
bad = 0
for trial in range(400):
n = rng.randrange(0, 400)
text = "".join(rng.choice(alphabet) for _ in range(n))
mt = rng.choice([1, 2, 3, 4, 8, 16, 32])
ov = rng.choice([0, 0, 1, 2, mt - 1, mt])
if ov > mt:
ov = mt
try:
chunks = mod.chunk_text(text, mt, ov)
except Exception as e:
check(False, f"fuzz trial {trial}: no exception",
f"{type(e).__name__}: {e}")
bad += 1
continue
if mod.reconstruct(chunks) != text:
check(False, f"fuzz trial {trial}: roundtrip",
f"mt={mt} ov={ov} len={n}")
bad += 1
for c in chunks:
if c["est_tokens"] > mt:
check(False, f"fuzz trial {trial}: bound",
f"est={c['est_tokens']} mt={mt}")
bad += 1
break
check(bad == 0, "fuzz: 400 randomised inputs hold every invariant")
def _invariant_failures(mod) -> list[str]:
"""Run the harness's own invariants against *mod*; return the failures.
This is the teeth mechanism: the harness is applied to an arbitrary module,
so a mutant that violates any invariant is caught here. The invariants are
the four properties the library actually promises:
1. round-trip is byte-exact,
2. every chunk is within the token budget,
3. ``est_tokens`` is the library's own estimate of the chunk text,
4. the chunks tile the source with no gap and no overlap,
5. the reported overlap is a real prefix of the chunk, within budget.
"""
global FAILURES
saved = FAILURES
FAILURES = []
try:
text = " ".join(f"Sentence number {i}." for i in range(40))
long_text = _fixtures()["long_word_and_text"]
repetitive = _fixtures()["repetitive"]
for t in (text, long_text, repetitive):
for mt, ov in [(8, 2), (16, 4), (4, 1), (32, 16)]:
try:
chunks = mod.chunk_text(t, mt, ov)
except Exception as e:
FAILURES.append(f"raised {type(e).__name__}")
continue
# 1 round-trip
if mod.reconstruct(chunks) != t:
FAILURES.append("roundtrip")
# 2 budget
for c in chunks:
if c["est_tokens"] > mt:
FAILURES.append("bound")
# 3 est_tokens is honest
if not all(c["est_tokens"] == mod.estimate_tokens(c["text"])
for c in chunks):
FAILURES.append("est_tokens consistency")
# 4 tiling
cursor = 0
for i, c in enumerate(chunks):
if c["index"] != i:
FAILURES.append("indices")
if c["char_start"] != cursor:
FAILURES.append("tiling-gap")
if c["char_end"] != cursor + len(c["text"]):
FAILURES.append("tiling-length")
cursor = c["char_end"]
if chunks and cursor != len(t):
FAILURES.append("tiling-shortfall")
# 5 overlap is a real, bounded prefix
for c in chunks:
if c["overlap"]:
if not c["text"].startswith(c["overlap"]):
FAILURES.append("overlap-not-prefix")
if c["overlap_est_tokens"] > ov:
FAILURES.append("overlap-over-budget")
return list(FAILURES)
finally:
FAILURES = saved
def stage_teeth() -> None:
"""The harness must be able to fail. Run the invariants against broken copies.
Every mutant below is a real defect class. If the harness passes one, the
harness is worthless and this stage says so.
"""
src = open(LIB, "r", encoding="utf-8").read()
mutants = {
"drop the size cap (always emit one chunk)":
re.sub(r"if len\(_piece_text\(cur\)\) \+ len\(nxt\) > max_chars:",
"if False:", src),
"corrupt the overlap (copy the wrong tail)":
src.replace('cand = prev_text[-overlap_chars:]',
'cand = prev_text[:overlap_chars]'),
"reconstruct joins with a space (breaks byte-exactness)":
src.replace('return "".join(parts)', 'return " ".join(parts)'),
"swap the token estimate (4 -> 1)":
src.replace("CHARS_PER_TOKEN = 4", "CHARS_PER_TOKEN = 1"),
"make the first chunk index start at 1":
src.replace('"index": index,', '"index": index + 1,'),
"hard-split without a width cap":
re.sub(r"end = min\(start \+ max_chars, n\)", "end = n", src),
"drop whitespace atoms on the floor":
src.replace('return _ATOM_RE.findall(text)',
'return [a for a in _ATOM_RE.findall(text) if a.strip()]'),
"first chunk starts at char 1 (off-by-one in tiling)":
src.replace('"char_start": start,', '"char_start": start + 1,'),
}
caught = 0
for label, mut in mutants.items():
if mut == src:
check(False, f"teeth: mutant applied: {label}",
"the substitution did not change the source")
continue
with tempfile.NamedTemporaryFile(
"w", suffix=".py", delete=False, encoding="utf-8"
) as fh:
fh.write(mut)
path = fh.name
try:
spec = importlib.util.spec_from_file_location("mutant", path)
m = importlib.util.module_from_spec(spec)
spec.loader.exec_module(m)
fails = _invariant_failures(m)
if fails:
caught += 1
check(True, f"teeth: caught {label} -> {sorted(set(fails))}")
else:
check(False, f"teeth: harness FAILS to catch: {label}",
"broken library passed every invariant")
except Exception as e:
caught += 1
check(True, f"teeth: caught {label} (raised {type(e).__name__})")
finally:
os.unlink(path)
check(caught == len(mutants),
f"teeth: {caught}/{len(mutants)} planted defects caught")
def stage_teeth_noise_only() -> None:
"""Two mutants that change *no observable property* must be caught by the
harness's structural source checks, not by the invariants.
``CHARS_PER_TOKEN`` is a documented, published constant (the harness states
it as part of the contract), and ``est_tokens`` is required to be derived
from it rather than to agree with the module's own ``estimate_tokens``
(which a mutant can redefine). Both are therefore checked against the
source, and this stage records that these are source checks -- they are
not invariant checks, and are labelled as such.
"""
src = open(LIB, "r", encoding="utf-8").read()
# A: the published constant is still 4.
m = re.search(r"^CHARS_PER_TOKEN = (\d+)", src, re.M)
check(m is not None and m.group(1) == "4",
"teeth(source): CHARS_PER_TOKEN is still the published 4",
m.group(0) if m else "constant not found")
# B: est_tokens is derived from CHARS_PER_TOKEN, not from estimate_tokens.
body = re.search(r"def estimate_tokens\(.*?\n(?=\ndef |\nclass |\Z)", src, re.S)
ok_b = bool(body) and "CHARS_PER_TOKEN" in body.group(0)
check(ok_b, "teeth(source): estimate_tokens derives from CHARS_PER_TOKEN")
# C: est_tokens is set from estimate_tokens at chunk-build time.
check('"est_tokens": estimate_tokens(ptext)' in src,
"teeth(source): chunk est_tokens is computed by estimate_tokens")
def stage_old_chunker_is_broken() -> None:
"""Independent evidence that the predecessor's claims were false.
This stage does not test chunktext; it re-measures the artifact this one
replaces, so the replacement is justified by a measurement rather than by
assertion. If the old file is absent or already fixed, the stage reports
that plainly instead of claiming a pass.
"""
old = "/root/.hermes/wizard-outputs/null/chunker.py"
if not os.path.exists(old):
check(True, "old: predecessor file not present on this host; "
"skipped (no claim made)")
return
spec = importlib.util.spec_from_file_location("old_chunker", old)
m = importlib.util.module_from_spec(spec)
spec.loader.exec_module(m)
text = " ".join(f"Sentence number {i}." for i in range(20))
mismatches = 0
over = 0
total = 0
for mt in (4, 8, 16):
for ov in (0, 1, 4):
if ov >= mt:
continue
chunks = m.chunk_text(text, mt, ov)
if m.reconstruct(chunks, ov) != text:
mismatches += 1
for c in chunks:
total += 1
if m.estimate_tokens(c) > mt:
over += 1
check(mismatches > 0,
f"old: predecessor round-trip fails on {mismatches} of 6 budgets",
"the predecessor claims lossless; measured otherwise")
check(over > 0,
f"old: predecessor emits {over} over-budget chunks of {total}",
"the predecessor applies overlap after sizing")
# the sentinel-escape claim
src = open(old, "r", encoding="utf-8").read()
escapes = src.count("_OVERLAP_SENTINEL.replace") + src.count("replace(_OVERLAP_SENTINEL")
check(escapes == 0,
"old: predecessor never escapes its sentinel despite documenting it")
# ---------------------------------------------------------------------------
STAGES = [
("roundtrip", stage_roundtrip),
("bound", stage_bound),
("overlap", stage_overlap),
("structure", stage_structure),
("json", stage_json),
("bare-strings", stage_bare_strings),
("validation", stage_validation),
("empty", stage_empty),
("sentinel", stage_sentinel),
("realistic", stage_realistic),
("fuzz", stage_fuzz),
("teeth", stage_teeth),
("teeth-source", stage_teeth_noise_only),
("old-chunker", stage_old_chunker_is_broken),
]
def _load(path: str):
spec = importlib.util.spec_from_file_location("chunktext_under_test", path)
mod = importlib.util.module_from_spec(spec)
spec.loader.exec_module(mod)
return mod
def main() -> int:
print(f"chunktext verification harness")
print(f"library: {LIB}")
if not os.path.exists(LIB):
print("HARNESS BROKE: library file missing")
return 2
try:
mod = _load(LIB)
except Exception:
print("HARNESS BROKE: could not import the library")
traceback.print_exc()
return 2
stage_errors = []
for name, fn in STAGES:
global CHECKS
before = CHECKS
try:
if fn is stage_teeth or fn is stage_old_chunker_is_broken:
fn()
else:
fn(mod)
except Exception:
stage_errors.append(name)
print(f"[HARNESS ERROR] stage {name!r} raised")
traceback.print_exc()
print(f"-- stage {name}: {CHECKS - before} checks, "
f"{len(FAILURES)} failure(s) total")
print()
if stage_errors:
print(f"HARNESS BROKE: {len(stage_errors)} stage(s) raised: "
f"{', '.join(stage_errors)}")
return 2
if FAILURES:
print(f"RESULT: FAIL -- {len(FAILURES)} claim(s) did not hold "
f"of {CHECKS} checked")
for f in FAILURES[:25]:
print(" -", f)
return 1
print(f"RESULT: PASS -- {CHECKS} claims observed to hold, 0 failed")
return 0
if __name__ == "__main__":
sys.exit(main())