255 lines
11 KiB
Python
255 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""Canonical, source-bound unittest harness."""
|
|
import argparse
|
|
import ast
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
ROLES = ("common", "spec", "impl", "review")
|
|
SOURCE_ROOT = Path(__file__).resolve().parent
|
|
REPO_ROOT = SOURCE_ROOT.parents[1]
|
|
|
|
class CanonicalError(Exception):
|
|
pass
|
|
|
|
def fail(message):
|
|
raise CanonicalError("canonical harness: " + message)
|
|
|
|
def rel(path):
|
|
return path.relative_to(REPO_ROOT).as_posix()
|
|
|
|
def sha(path):
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
def _reject_source_tree_symlinks():
|
|
"""Reject links before any source bytes are read or imported."""
|
|
def visit(directory):
|
|
try:
|
|
entries = list(os.scandir(directory))
|
|
except OSError as exc:
|
|
fail(f"cannot inspect {rel(Path(directory))}: {exc}")
|
|
return
|
|
for entry in entries:
|
|
path = Path(entry.path)
|
|
if entry.is_symlink():
|
|
fail(f"symlinked source-tree path: {rel(path)}")
|
|
if entry.is_dir(follow_symlinks=False):
|
|
visit(path)
|
|
visit(SOURCE_ROOT)
|
|
|
|
def module_name(path):
|
|
return ".".join(path.relative_to(REPO_ROOT).with_suffix("").parts)
|
|
|
|
def _package_manifest():
|
|
"""Return raw-byte fingerprints for all source and package markers."""
|
|
_reject_source_tree_symlinks()
|
|
files = {}
|
|
if not SOURCE_ROOT.is_dir():
|
|
fail("missing source root")
|
|
for directory in [SOURCE_ROOT, *(SOURCE_ROOT / role for role in ROLES)]:
|
|
marker = directory / "__init__.py"
|
|
if not marker.is_file():
|
|
fail(f"missing package marker: {rel(marker)}")
|
|
for path in SOURCE_ROOT.rglob("*"):
|
|
if path.is_file() and (path.name == "__init__.py" or path.name.startswith("test") and path.suffix == ".py"):
|
|
files[rel(path)] = path.read_bytes()
|
|
return files
|
|
|
|
def _reject_package_load_tests(manifest=None):
|
|
manifest = _package_manifest() if manifest is None else manifest
|
|
for name in manifest:
|
|
if not name.endswith("__init__.py"):
|
|
continue
|
|
path = REPO_ROOT / name
|
|
try:
|
|
tree = ast.parse(path.read_bytes(), filename=str(path))
|
|
except (OSError, SyntaxError) as exc:
|
|
fail(f"cannot parse {rel(path)}: {exc}")
|
|
if any(isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef)) and n.name == "load_tests" for n in tree.body):
|
|
fail(f"unsupported dynamic load_tests declaration in {rel(path)}")
|
|
|
|
def package_dirs(manifest=None):
|
|
_reject_package_load_tests(manifest)
|
|
for path in SOURCE_ROOT.rglob("test*.py"):
|
|
current = path.parent
|
|
while current != SOURCE_ROOT.parent:
|
|
if current in (SOURCE_ROOT, *(SOURCE_ROOT / role for role in ROLES)):
|
|
current = current.parent
|
|
continue
|
|
marker = current / "__init__.py"
|
|
if not marker.is_file():
|
|
fail(f"nested test package missing marker: {rel(marker)}")
|
|
current = current.parent
|
|
|
|
def source_files(manifest=None):
|
|
package_dirs(manifest)
|
|
files = sorted(SOURCE_ROOT.rglob("test*.py"))
|
|
role_roots = {SOURCE_ROOT / role for role in ROLES}
|
|
unexpected = [p for p in files if not any(p.is_relative_to(d) for d in role_roots)]
|
|
if unexpected:
|
|
fail("unexpected test file outside role dirs: " + ", ".join(rel(p) for p in unexpected))
|
|
by_role = {role: sorted((SOURCE_ROOT / role).rglob("test*.py")) for role in ROLES}
|
|
for role, paths in by_role.items():
|
|
if not paths:
|
|
fail(f"required role has no test files: {role}")
|
|
return by_role
|
|
|
|
def aliases(tree):
|
|
names = {"TestCase", "unittest.TestCase"}
|
|
for node in tree.body:
|
|
if isinstance(node, ast.ImportFrom) and node.module == "unittest":
|
|
for item in node.names:
|
|
if item.name == "TestCase":
|
|
names.add(item.asname or item.name)
|
|
elif isinstance(node, ast.Import):
|
|
for item in node.names:
|
|
if item.name == "unittest":
|
|
names.add((item.asname or item.name) + ".TestCase")
|
|
return names
|
|
|
|
def inventory_file(path, raw_bytes=None):
|
|
try:
|
|
raw_bytes = path.read_bytes() if raw_bytes is None else raw_bytes
|
|
tree = ast.parse(raw_bytes, filename=str(path))
|
|
except (OSError, SyntaxError) as exc:
|
|
fail(f"cannot parse {rel(path)}: {exc}")
|
|
def conditional_declarations(nodes, context):
|
|
controls = (ast.If, ast.For, ast.AsyncFor, ast.While, ast.Try, ast.With, ast.AsyncWith, ast.Match)
|
|
for node in nodes:
|
|
if isinstance(node, controls):
|
|
stack = list(ast.iter_child_nodes(node))
|
|
while stack:
|
|
child = stack.pop()
|
|
if isinstance(child, (ast.ClassDef, ast.FunctionDef, ast.AsyncFunctionDef)):
|
|
if child.name.startswith("test") or isinstance(child, ast.ClassDef):
|
|
fail(f"unsupported conditional test declaration in {rel(path)}: {context}.{child.name}")
|
|
if isinstance(child, (ast.FunctionDef, ast.AsyncFunctionDef)):
|
|
continue
|
|
stack.extend(ast.iter_child_nodes(child))
|
|
conditional_declarations(tree.body, "module")
|
|
if any(isinstance(n, (ast.FunctionDef, ast.AsyncFunctionDef)) and n.name == "load_tests" for n in tree.body):
|
|
fail(f"unsupported dynamic load_tests declaration in {rel(path)}")
|
|
testcase_names = aliases(tree)
|
|
classes = {}
|
|
for node in tree.body:
|
|
if isinstance(node, ast.ClassDef):
|
|
if node.name in classes:
|
|
fail(f"duplicate test class declaration in {rel(path)}: {node.name}")
|
|
classes[node.name] = node
|
|
conditional_declarations(node.body, node.name)
|
|
child_names = [c.name for c in node.body if isinstance(c, (ast.FunctionDef, ast.AsyncFunctionDef))]
|
|
if len(child_names) != len(set(child_names)):
|
|
fail(f"duplicate declaration in {rel(path)}: {node.name}")
|
|
state = {}
|
|
memo = {}
|
|
def base_name(base):
|
|
if isinstance(base, ast.Name): return base.id
|
|
if isinstance(base, ast.Attribute) and isinstance(base.value, ast.Name): return base.value.id + "." + base.attr
|
|
fail(f"unsupported test class base in {rel(path)}: {ast.unparse(base)}")
|
|
def is_case(name):
|
|
if name in memo: return memo[name]
|
|
if state.get(name) == "active":
|
|
fail(f"cyclic test class declaration in {rel(path)}: {name}")
|
|
node = classes.get(name)
|
|
if node is None:
|
|
result = name in testcase_names
|
|
if not result and name not in testcase_names:
|
|
fail(f"unknown test class base in {rel(path)}: {name}")
|
|
return result
|
|
state[name] = "active"
|
|
result = False
|
|
for base in node.bases: # eager: never short-circuit validation
|
|
result = is_case(base_name(base)) or result
|
|
state.pop(name, None)
|
|
memo[name] = result
|
|
return result
|
|
def local_methods(name, stack=()):
|
|
if name in stack:
|
|
fail(f"cyclic test inheritance in {rel(path)}: {' -> '.join(stack + (name,))}")
|
|
node = classes[name]
|
|
found = {c.name for c in node.body if isinstance(c, (ast.FunctionDef, ast.AsyncFunctionDef)) and c.name.startswith("test")}
|
|
for base in node.bases:
|
|
b = base_name(base)
|
|
if b in classes:
|
|
is_case(b)
|
|
found |= local_methods(b, stack + (name,))
|
|
return found
|
|
methods = {}
|
|
free = []
|
|
for node in tree.body:
|
|
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name.startswith("test"):
|
|
free.append(node.name)
|
|
for name in classes:
|
|
case = is_case(name)
|
|
own = {c.name for c in classes[name].body if isinstance(c, (ast.FunctionDef, ast.AsyncFunctionDef)) and c.name.startswith("test")}
|
|
if case:
|
|
methods[name] = local_methods(name)
|
|
if free:
|
|
fail(f"unsupported free-function test in {rel(path)}: {free[0]}")
|
|
ids = {f"{module_name(path)}.{name}.{method}" for name, ms in methods.items() for method in ms}
|
|
if not ids:
|
|
fail(f"file has zero inventoried tests: {rel(path)}")
|
|
return ids
|
|
|
|
def inventory(files, manifest=None):
|
|
ids, role_ids, hashes = set(), {}, {}
|
|
manifest = _package_manifest() if manifest is None else manifest
|
|
for role, paths in files.items():
|
|
role_ids[role] = set()
|
|
for path in paths:
|
|
raw_bytes = manifest.get(rel(path))
|
|
if raw_bytes is None:
|
|
fail(f"source missing from package manifest: {rel(path)}")
|
|
hashes[rel(path)] = hashlib.sha256(raw_bytes).hexdigest()
|
|
found = inventory_file(path, raw_bytes)
|
|
overlap = ids & found
|
|
if overlap: fail("duplicate test IDs: " + ", ".join(sorted(overlap)))
|
|
ids |= found; role_ids[role] |= found
|
|
if not role_ids[role]: fail(f"role has zero inventoried tests: {role}")
|
|
return ids, role_ids, hashes
|
|
|
|
def flatten(suite):
|
|
for item in suite:
|
|
if isinstance(item, unittest.TestSuite): yield from flatten(item)
|
|
else: yield item
|
|
|
|
def collect(files, loader=None):
|
|
loader = loader or unittest.defaultTestLoader
|
|
suite = loader.discover(str(SOURCE_ROOT), pattern="test*.py", top_level_dir=str(REPO_ROOT))
|
|
flat = list(flatten(suite))
|
|
failed = [test for test in flat if isinstance(test, unittest.loader._FailedTest)]
|
|
if failed: fail("import/loader failure: " + "; ".join(str(test) for test in failed[:20]))
|
|
raw = [test.id() for test in flat]
|
|
seen = set(); duplicates = []
|
|
for test_id in raw:
|
|
if test_id in seen and test_id not in duplicates: duplicates.append(test_id)
|
|
seen.add(test_id)
|
|
if duplicates: fail("duplicate loader test IDs: " + ", ".join(duplicates[:20]))
|
|
return set(raw), suite
|
|
|
|
def validate():
|
|
before_manifest = _package_manifest()
|
|
files = source_files(before_manifest)
|
|
expected, role_ids, before = inventory(files, before_manifest)
|
|
actual, suite = collect(files)
|
|
after_manifest = _package_manifest()
|
|
if before_manifest != after_manifest: fail("source/package manifest drift detected during collection")
|
|
if expected != actual:
|
|
fail(f"inventory/loader mismatch: missing={sorted(expected-actual)!r} extra={sorted(actual-expected)!r}")
|
|
return expected, role_ids, before, suite
|
|
|
|
def main(argv=None):
|
|
parser = argparse.ArgumentParser(add_help=True); parser.add_argument("--collect-only", action="store_true")
|
|
args = parser.parse_args(argv); sys.path.insert(0, str(REPO_ROOT))
|
|
try: ids, role_ids, hashes, suite = validate()
|
|
except CanonicalError as exc: print(str(exc), file=sys.stderr); return 2
|
|
if args.collect_only:
|
|
print(json.dumps({"collection_only": True, "role_counts": {r: len(role_ids[r]) for r in ROLES}, "ids": sorted(ids), "source_sha256": hashes}, sort_keys=True)); return 0
|
|
result = unittest.TextTestRunner(verbosity=2).run(suite); return 0 if result.wasSuccessful() else 1
|
|
|
|
if __name__ == "__main__": raise SystemExit(main())
|