#!/usr/bin/env python3 """Run the upstream jq conformance corpus against a candidate implementation. This is the scoring instrument. It is supplied, not authored: it is staged verbatim into the build directory, hash-verified against the import, and restored before grading. Its exit status is the acceptance verdict. It is deliberately external to the implementation. Upstream jq grades itself, through its own ``--run-tests`` flag; a self-graded suite proves nothing here, so this runner re-implements the corpus protocol and drives the candidate as a subprocess, one process per case. Imported sources land in ``sources/`` inside the application directory, so this is normally invoked as ``python3 sources/run_conformance.py`` from that directory. Usage: JQ=./jq python3 sources/run_conformance.py # full corpus, the scored run JQ=./jq python3 sources/run_conformance.py -v # list passing cases too JQ=./jq python3 sources/run_conformance.py --json # machine-readable report JQ=./jq python3 sources/run_conformance.py --list # print cases, run nothing JQ=./jq python3 sources/run_conformance.py --select 'reduce' # develop one construct Environment: JQ command that runs the candidate. Required -- this harness is language-neutral and deliberately has no default implementation language. Exit codes: 0 every case that ran passed 1 at least one case failed or errored 2 the harness could not run: bad usage, missing corpus, or a stale exclusion """ from __future__ import annotations import argparse import json import os import re import shlex import subprocess import sys from dataclasses import dataclass, field from pathlib import Path HERE = Path(__file__).resolve().parent CORPUS = HERE / "jq.test" EXCLUSIONS = HERE / "exclusions.txt" #: A case that has not produced output in this long is not going to. jq's own suite runs the #: whole corpus in under a second; anything near this bound is a runaway generator. DEFAULT_TIMEOUT = 10.0 PASS, FAIL, ERROR, SKIP = "pass", "fail", "error", "skip" #: jq's documented exit codes, which this kit's interface contract adopts. The distinction is #: load-bearing: the corpus's %%FAIL cases are programs that must not compile, while an ordinary #: case may legitimately raise at run time part-way through its output and still be correct. EXIT_COMPILE_ERROR = 3 EXIT_RUNTIME_ERROR = 5 def split_lines(text: str) -> list[str]: """Split on newlines only. ``str.splitlines`` is Unicode-aware and also breaks on U+000B, U+000C, U+0085, U+2028, and U+2029. The corpus contains cases whose expected output embeds those code points inside JSON strings -- ``trim, ltrim, rtrim`` over the Unicode whitespace set is one -- and splitting there shreds one JSON value into several, failing a correct implementation. """ lines = text.split("\n") if lines and lines[-1] == "": lines.pop() return lines @dataclass class Case: """One corpus case: a program, an input, and what it must produce. ``expect_failure`` cases are the corpus's ``%%FAIL`` blocks. Upstream compares their diagnostic text; this runner requires only that the candidate reject the program. The expected strings are jq's exact C-implementation diagnostics, down to the caret art underlining the offending token -- reproducing them is reverse-engineering an implementation, not conforming to a specification. The text is parsed and reported so a reader can see what upstream said, and is never compared. """ line: int program: str stdin: str = "" expected: list[str] = field(default_factory=list) expect_failure: bool = False diagnostic: str = "" @dataclass class Result: case: Case status: str detail: str = "" actual: list[str] = field(default_factory=list) return_code: int | None = None stderr: str = "" class HarnessError(Exception): """A fault in the kit or its invocation, never a fault in the candidate.""" # -------------------------------------------------------------------------------------------- # Corpus parsing # -------------------------------------------------------------------------------------------- def parse_corpus(text: str) -> list[Case]: """Parse the jq test corpus. The format is documented in the corpus's own header: cases are groups of lines separated by blank lines; blank lines and lines starting with ``#`` are ignored. A case is a program line, an input line, and then zero or more expected output lines. A case preceded by a ``%%FAIL`` or ``%%FAIL IGNORE MSG`` marker is a program line followed by the diagnostic upstream jq emits, which may itself span several lines of source excerpt and caret art. """ cases: list[Case] = [] block: list[tuple[int, str]] = [] expect_failure = False lines = split_lines(text) def flush() -> None: nonlocal block, expect_failure if block: cases.append(_case_from_block(block, expect_failure)) block = [] expect_failure = False for number, raw in enumerate(lines, start=1): stripped = raw.strip() if not stripped: flush() continue if raw.startswith("%%FAIL"): flush() expect_failure = True continue # A '#' comment closes nothing: upstream places section banners between cases, always # with blank lines around them, and never inside a case. if raw.lstrip().startswith("#") and not block: continue if raw.lstrip().startswith("#"): continue block.append((number, raw)) flush() return cases def _case_from_block(block: list[tuple[int, str]], expect_failure: bool) -> Case: line, program = block[0] if expect_failure: diagnostic = "\n".join(text for _, text in block[1:]) return Case(line=line, program=program, expect_failure=True, diagnostic=diagnostic) if len(block) < 2: raise HarnessError( f"{CORPUS.name}:{line}: case has a program but no input line; the corpus is malformed" ) return Case( line=line, program=program, stdin=block[1][1], expected=[text for _, text in block[2:]], ) # -------------------------------------------------------------------------------------------- # Exclusions # -------------------------------------------------------------------------------------------- def parse_exclusions(path: Path) -> list[str]: """Read the declared exclusions: one verbatim program line per entry, ``#`` for reasons.""" if not path.is_file(): return [] return [ line for line in split_lines(path.read_text(encoding="utf-8")) if line.strip() and not line.lstrip().startswith("#") ] def apply_exclusions(cases: list[Case], exclusions: list[str]) -> set[int]: """Return the corpus line numbers of excluded cases. An exclusion that matches nothing is a hard error rather than a shrug. The corpus is pinned by hash, so a stale exclusion means the pin moved without the exclusion list being revisited -- and a silent no-op there would quietly re-admit a case the kit cannot run. """ by_program: dict[str, list[Case]] = {} for case in cases: by_program.setdefault(case.program, []).append(case) excluded: set[int] = set() stale: list[str] = [] for program in exclusions: matched = by_program.get(program) if not matched: stale.append(program) continue excluded.update(case.line for case in matched) if stale: listed = "\n".join(f" {program}" for program in stale) raise HarnessError( f"{EXCLUSIONS.name}: these exclusions match no case in {CORPUS.name}:\n{listed}\n" "The corpus and the exclusion list have drifted apart." ) return excluded # -------------------------------------------------------------------------------------------- # Comparison # -------------------------------------------------------------------------------------------- def jv_equal(left: object, right: object) -> bool: """Compare two decoded JSON values the way jq's own ``jv_equal`` does. Structural, not textual: ``1`` and ``1.0`` are the same jq value and the corpus relies on that. Python's ``==`` almost does this, but it also equates ``True`` with ``1`` and ``False`` with ``0``, which jq does not; booleans are therefore matched by identity of type before anything else. """ if isinstance(left, bool) or isinstance(right, bool): return isinstance(left, bool) and isinstance(right, bool) and left is right if isinstance(left, (int, float)) and isinstance(right, (int, float)): return left == right if isinstance(left, list) and isinstance(right, list): return len(left) == len(right) and all(jv_equal(a, b) for a, b in zip(left, right)) if isinstance(left, dict) and isinstance(right, dict): return left.keys() == right.keys() and all(jv_equal(left[k], right[k]) for k in left) if type(left) is not type(right): return False return left == right def _decode(line: str) -> tuple[bool, object]: try: return True, json.loads(line) except ValueError: return False, line def outputs_match(expected: list[str], actual: list[str]) -> bool: if len(expected) != len(actual): return False for want, got in zip(expected, actual): want_ok, want_value = _decode(want) got_ok, got_value = _decode(got) if want_ok and got_ok: if not jv_equal(want_value, got_value): return False elif want.strip() != got.strip(): return False return True # -------------------------------------------------------------------------------------------- # Execution # -------------------------------------------------------------------------------------------- def run_case(case: Case, argv: list[str], timeout: float) -> Result: try: completed = subprocess.run( [*argv, "-c", case.program], input=case.stdin, capture_output=True, text=True, timeout=timeout, ) except subprocess.TimeoutExpired: return Result(case, ERROR, detail=f"timed out after {timeout:g}s") except OSError as exc: raise HarnessError(f"cannot execute {shlex.join(argv)}: {exc}") from exc actual = split_lines(completed.stdout) stderr = completed.stderr.strip() code = completed.returncode first_diagnostic = split_lines(stderr)[0] if stderr else "" if case.expect_failure: # The corpus's %%FAIL cases are programs that must be rejected at compile time. Accepting # one and then failing at run time is a different, wrong behaviour, so the compile-error # code specifically -- not merely a non-zero exit -- is what passes here. if code == EXIT_COMPILE_ERROR: return Result(case, PASS, return_code=code, stderr=stderr) detail = ( "program was accepted, but the corpus marks it %%FAIL" if code == 0 else f"exited {code}; a rejected program must exit {EXIT_COMPILE_ERROR}" ) return Result(case, FAIL, detail=detail, actual=actual, return_code=code, stderr=stderr) if code == EXIT_COMPILE_ERROR: return Result( case, FAIL, detail=f"program did not compile: {first_diagnostic}", actual=actual, return_code=code, stderr=stderr, ) if code not in (0, EXIT_RUNTIME_ERROR): # A runtime error is legitimate: several cases raise part-way through a generator and are # judged on the outputs produced before the raise, exactly as upstream judges them. Any # other non-zero status is the program failing in a way the contract does not describe. return Result( case, FAIL, detail=f"exited {code}: {first_diagnostic}", actual=actual, return_code=code, stderr=stderr, ) if not outputs_match(case.expected, actual): return Result( case, FAIL, detail="output mismatch", actual=actual, return_code=code, stderr=stderr, ) return Result(case, PASS, actual=actual, return_code=code, stderr=stderr) # -------------------------------------------------------------------------------------------- # Reporting # -------------------------------------------------------------------------------------------- def _render_failure(result: Result) -> str: case = result.case lines = [ f"FAIL {CORPUS.name}:{case.line} {result.detail}", f" program: {case.program}", ] if not case.expect_failure: lines.append(f" input: {case.stdin}") lines.append(f" expected: {case.expected if case.expected else '(no output)'}") lines.append(f" actual: {result.actual if result.actual else '(no output)'}") if result.stderr: lines.append(f" stderr: {split_lines(result.stderr)[0]}") return "\n".join(lines) def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser( description="Run the upstream jq conformance corpus against a candidate implementation.", ) parser.add_argument("--jq", default=None, help="candidate command (default: $JQ)") parser.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT, help="seconds per case") parser.add_argument("--json", action="store_true", help="machine-readable report") parser.add_argument("-v", "--verbose", action="store_true", help="list passing cases too") parser.add_argument("--list", action="store_true", help="print the cases and run nothing") parser.add_argument( "--select", default=None, metavar="REGEX", help="run only cases whose program matches REGEX (development aid; the acceptance " "gate always runs the whole corpus)", ) args = parser.parse_args(argv) try: return _run(args) except HarnessError as exc: print(f"error: {exc}", file=sys.stderr) return 2 def _run(args: argparse.Namespace) -> int: if not CORPUS.is_file(): raise HarnessError(f"corpus not found at {CORPUS}") cases = parse_corpus(CORPUS.read_text(encoding="utf-8")) excluded = apply_exclusions(cases, parse_exclusions(EXCLUSIONS)) selector = re.compile(args.select) if args.select else None if selector is not None: cases = [case for case in cases if selector.search(case.program)] if args.list: for case in cases: mark = "skip" if case.line in excluded else "run " print(f"{mark} {CORPUS.name}:{case.line} {case.program}") print(f"\n{len(cases)} cases, {sum(1 for c in cases if c.line in excluded)} excluded") return 0 command = args.jq or os.environ.get("JQ") or "" if not command.strip(): raise HarnessError( "JQ is not set; give the command that runs your implementation, e.g.\n" ' JQ="$PWD/jq" python3 sources/run_conformance.py' ) jq_argv = shlex.split(command) results: list[Result] = [] for case in cases: if case.line in excluded: results.append(Result(case, SKIP, detail="declared in exclusions.txt")) continue results.append(run_case(case, jq_argv, args.timeout)) tally = {status: sum(1 for r in results if r.status == status) for status in (PASS, FAIL, ERROR, SKIP)} summary = ( f"jq conformance: {tally[PASS]} passed, {tally[FAIL]} failed, " f"{tally[ERROR]} errored, {tally[SKIP]} skipped " f"(corpus {CORPUS.name} @ jq-1.8.2)" ) if args.json: print(json.dumps( { "candidate": jq_argv, "corpus": CORPUS.name, "summary": tally, "cases": [ { "line": r.case.line, "program": r.case.program, "status": r.status, "detail": r.detail, "expect_failure": r.case.expect_failure, "expected": r.case.expected, "actual": r.actual, } for r in results if args.verbose or r.status != PASS ], }, indent=2, )) else: for result in results: if result.status in (FAIL, ERROR): print(_render_failure(result)) elif args.verbose and result.status == PASS: print(f"ok {CORPUS.name}:{result.case.line} {result.case.program}") elif args.verbose and result.status == SKIP: print(f"skip {CORPUS.name}:{result.case.line} {result.case.program}") print(summary) return 0 if tally[FAIL] == 0 and tally[ERROR] == 0 else 1 if __name__ == "__main__": raise SystemExit(main())