#!/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())
Project document