From fa0d42e0fe4b3f25a789d887e93135d0c320a339 Mon Sep 17 00:00:00 2001 From: Joshua Wink Date: Tue, 13 Oct 2026 11:48:57 -0500 Subject: [PATCH 1/2] refactor(conformance): the judge is a codeupipe pipeline The judge was eight plain functions in run.py. It is now a pipeline in conformance/judging: read the cases, choose the targets, run each target, compare, count, format, and write the results. Each step is a primitive in its own file and has tests that start no process. run.py keeps its command line and the function names that the tests and bench.py use. Each one calls a step of the pipeline, so there is one judge. The new judge was run beside the old one. Their reports are the same byte for byte on all 7 targets, and they gave the same verdict for each of 6550 results, 2700 of them wrong on purpose. Co-Authored-By: Claude Opus 5.5 --- conformance/README.md | 3 + conformance/judging/__init__.py | 5 + conformance/judging/default_targets_filter.py | 30 +++ conformance/judging/format_matrix_filter.py | 32 +++ conformance/judging/judge_pipeline.py | 36 ++++ conformance/judging/judge_results_filter.py | 93 +++++++++ conformance/judging/load_cases_filter.py | 28 +++ .../judging/no_targets_named_predicate.py | 10 + conformance/judging/node_probe.py | 20 ++ conformance/judging/process_launcher.py | 29 +++ conformance/judging/run_targets_filter.py | 29 +++ conformance/judging/summarize_filter.py | 37 ++++ conformance/judging/write_results_tap.py | 20 ++ conformance/run.py | 190 ++++-------------- cup.config.json | 2 +- tests/conformance/test_judge_pipeline.py | 113 +++++++++++ tests/lint/invariants/test_repo_is_clean.py | 2 +- 17 files changed, 521 insertions(+), 158 deletions(-) create mode 100644 conformance/judging/__init__.py create mode 100644 conformance/judging/default_targets_filter.py create mode 100644 conformance/judging/format_matrix_filter.py create mode 100644 conformance/judging/judge_pipeline.py create mode 100644 conformance/judging/judge_results_filter.py create mode 100644 conformance/judging/load_cases_filter.py create mode 100644 conformance/judging/no_targets_named_predicate.py create mode 100644 conformance/judging/node_probe.py create mode 100644 conformance/judging/process_launcher.py create mode 100644 conformance/judging/run_targets_filter.py create mode 100644 conformance/judging/summarize_filter.py create mode 100644 conformance/judging/write_results_tap.py create mode 100644 tests/conformance/test_judge_pipeline.py diff --git a/conformance/README.md b/conformance/README.md index 2871a48..eb94ad2 100644 --- a/conformance/README.md +++ b/conformance/README.md @@ -4,6 +4,9 @@ The rules are in **[API.md](API.md)**, the CUP API contract. This suite checks them. One set of language-neutral cases, a runner, and **one judge** (`run.py`). +The judge is itself a codeupipe pipeline, in [judging/](judging/): one step +reads the cases, one runs each target, one compares, and one writes the +matrix. Each step is tested alone in `tests/conformance/test_judge_pipeline.py`. The dynamic run and each compiled form are held to exactly the same comparison. diff --git a/conformance/judging/__init__.py b/conformance/judging/__init__.py new file mode 100644 index 0000000..7e322bb --- /dev/null +++ b/conformance/judging/__init__.py @@ -0,0 +1,5 @@ +"""The judge of the conformance suite, as a codeupipe pipeline (see ``judge_pipeline.py``). + +A runner executes the cases and reports what happened. Only the judge +decides pass or fail, so every target is held to the same comparison. +""" diff --git a/conformance/judging/default_targets_filter.py b/conformance/judging/default_targets_filter.py new file mode 100644 index 0000000..7c3a182 --- /dev/null +++ b/conformance/judging/default_targets_filter.py @@ -0,0 +1,30 @@ +"""DefaultTargetsFilter: each target that this machine can run.""" + +from codeupipe import Payload + + +class DefaultTargetsFilter: + """Filter (sync). + + Input keys: ``targets`` ({name: how to start its runner}), ``needs`` + ({name: paths that must exist for the target}). + Output keys: ``names``. A Python target always runs. A TypeScript target + needs a Node that runs ``.ts`` files, and the paths of ``needs``. + + ``node_runs_typescript`` is a function with no arguments that says if + the machine has such a Node. + """ + + def __init__(self, node_runs_typescript): + self.node_runs_typescript = node_runs_typescript + + def call(self, payload: Payload) -> Payload: + return payload.insert("names", self.choose(payload.get("targets"), payload.get("needs", {}))) + + def choose(self, targets, needs) -> list: + return [name for name in targets if self.can_run(name, needs)] + + def can_run(self, name, needs) -> bool: + if not name.startswith("typescript:"): + return True + return self.node_runs_typescript() and all(path.exists() for path in needs.get(name, [])) diff --git a/conformance/judging/format_matrix_filter.py b/conformance/judging/format_matrix_filter.py new file mode 100644 index 0000000..3d54bfa --- /dev/null +++ b/conformance/judging/format_matrix_filter.py @@ -0,0 +1,32 @@ +"""FormatMatrixFilter: the matrix and the counts as Markdown.""" + +from codeupipe import Payload + +_MARK = {"pass": "✅", "partial": "✅*", "FAIL": "❌", "unsupported": "➖", "—": "—"} + + +class FormatMatrixFilter: + """Filter (sync). + + Input keys: ``cases``, ``names``, ``matrix``, ``report``. + Output keys: ``markdown``. + """ + + def call(self, payload: Payload) -> Payload: + cases, names = payload.get("cases"), payload.get("names") + matrix, report = payload.get("matrix"), payload.get("report") + lines = ["| case | " + " | ".join(names) + " |", "|---|" + "---|" * len(names)] + for case in cases: + lines.append(f"| `{case['id']}` | " + " | ".join(_MARK[matrix[case['id']][n]] for n in names) + " |") + lines.append("") + for name in names: + target = report["targets"][name] + if "error" in target: + lines.append(f"- **{name}**: not run: {target['error']}") + else: + lines.append(f"- **{name}**: {target['pass']} pass, {target['partial']} pass*, {target['FAIL']} fail, " + f"{target['unsupported']} unsupported") + lines.append("") + lines.append("✅* passed everything the target can observe; it does not expose State or Hooks " + "(the fast mode). ➖ the target cannot express the case.") + return payload.insert("markdown", "\n".join(lines) + "\n") diff --git a/conformance/judging/judge_pipeline.py b/conformance/judging/judge_pipeline.py new file mode 100644 index 0000000..72f638d --- /dev/null +++ b/conformance/judging/judge_pipeline.py @@ -0,0 +1,36 @@ +"""JudgePipeline: run the conformance cases on each target and judge them one way.""" + +from codeupipe import Pipeline + +from .default_targets_filter import DefaultTargetsFilter +from .format_matrix_filter import FormatMatrixFilter +from .judge_results_filter import JudgeResultsFilter +from .load_cases_filter import LoadCasesFilter +from .no_targets_named_predicate import NoTargetsNamedPredicate +from .run_targets_filter import RunTargetsFilter +from .summarize_filter import SummarizeFilter +from .write_results_tap import WriteResultsTap + + +class JudgePipeline(Pipeline): + """The judge. + + Input keys: ``cases_dir``, ``results_dir``, ``targets`` ({name: how to + start its runner}), ``needs`` ({name: paths that the target needs}), and + ``names`` (the targets that the caller named; empty for each target that + this machine can run). + Output keys: ``report``, ``matrix``, ``markdown``, ``exit_code``. + + ``launcher`` starts the runner of a target. ``node_runs_typescript`` says + if the machine can run the TypeScript targets. + """ + + def __init__(self, launcher, node_runs_typescript): + super().__init__() + self.add_filter(LoadCasesFilter()) + self.add_valve(NoTargetsNamedPredicate(), DefaultTargetsFilter(node_runs_typescript)) + self.add_filter(RunTargetsFilter(launcher)) + self.add_filter(JudgeResultsFilter()) + self.add_filter(SummarizeFilter()) + self.add_filter(FormatMatrixFilter()) + self.add_tap(WriteResultsTap()) diff --git a/conformance/judging/judge_results_filter.py b/conformance/judging/judge_results_filter.py new file mode 100644 index 0000000..1f9a43c --- /dev/null +++ b/conformance/judging/judge_results_filter.py @@ -0,0 +1,93 @@ +"""JudgeResultsFilter: compare what each target reported with what each case expects. + +This is the one place that decides pass or fail. A verdict is a status and +its problems: + + pass every expectation matched + partial every expectation that the target can show matched, but it + does not show some of them at all (the fast compiled forms + show no State and no Hooks) + FAIL something that the target shows is different + unsupported the target cannot express the case. Never counted as a pass +""" + +from codeupipe import Payload + +# What a target may not show at all. A case that expects one of these, on a target that does not report it, is "partial". +_NOT_ALWAYS_SHOWN = ("state", "hooks", "inner_states", "pipeline_state") + + +class JudgeResultsFilter: + """Filter (sync). + + Input keys: ``cases``, ``results`` ({name: {case id: result}}). + Output keys: ``verdicts`` ({name: {case id: (status, problems)}}). + """ + + def call(self, payload: Payload) -> Payload: + cases = payload.get("cases") + verdicts = {name: {case["id"]: self.judge(case, found.get(case["id"])) for case in cases} + for name, found in payload.get("results").items()} + return payload.insert("verdicts", verdicts) + + def judge(self, case, result): + """The verdict for one case on one target: ``(status, problems)``.""" + if result is None: + return "FAIL", ["no result from runner"] + if "unsupported" in result: + return "unsupported", [result["unsupported"]] + expect = case["expect"] + problems = self.output_problems(expect, result) + + if self.norm(result.get("input_after")) != self.norm(case["input"]): + problems.append(f"input was mutated: {result.get('input_after')!r}") + + if "taps" in expect and self.norm(result.get("taps")) != self.norm(expect["taps"]): + problems.append(f"taps {result.get('taps')!r} != {expect['taps']!r}") + + if "state" in expect and "state" in result: + for key, want in expect["state"].items(): + got = result["state"].get(key) + if got != want: + problems.append(f"state.{key} {got!r} != {want!r}") + + for key in ("inner_states", "pipeline_state"): + if key in expect and key in result and result[key] != expect[key]: + problems.append(f"{key} {result[key]!r} != {expect[key]!r}") + + if "hooks" in expect and "hooks" in result and result["hooks"] != expect["hooks"]: + problems.append(f"hooks {result['hooks']!r} != {expect['hooks']!r}") + + if problems: + return "FAIL", problems + unchecked = [key for key in _NOT_ALWAYS_SHOWN if key in expect and key not in result] + if unchecked: + return "partial", [f"target cannot observe {', '.join(unchecked)}"] + return "pass", [] + + def output_problems(self, expect, result) -> list: + """What is wrong with the output or the error of a run.""" + problems = [] + if "output" in expect: + if "error" in result: + problems.append(f"expected output, got error {result['error']!r}") + elif self.norm(result.get("output")) != self.norm(expect["output"]): + problems.append(f"output {result.get('output')!r} != {expect['output']!r}") + if "error" in expect: + if "error" not in result: + problems.append(f"expected an error, got output {result.get('output')!r}") + elif expect["error"].get("message_contains", "") not in result["error"]: + problems.append(f"error {result['error']!r} lacks {expect['error']['message_contains']!r}") + return problems + + def norm(self, value): + """JSON equality: 1 == 1.0, but true is not 1.""" + if isinstance(value, bool) or value is None or isinstance(value, str): + return value + if isinstance(value, (int, float)): + return float(value) + if isinstance(value, list): + return [self.norm(v) for v in value] + if isinstance(value, dict): + return {k: self.norm(v) for k, v in value.items()} + return value diff --git a/conformance/judging/load_cases_filter.py b/conformance/judging/load_cases_filter.py new file mode 100644 index 0000000..b3649f7 --- /dev/null +++ b/conformance/judging/load_cases_filter.py @@ -0,0 +1,28 @@ +"""LoadCasesFilter: read the conformance cases from disk.""" + +import json +import pathlib + +from codeupipe import Payload + + +class LoadCasesFilter: + """Filter (sync). + + Input keys: ``cases_dir`` (the directory of the ``*.json`` case files). + Output keys: ``cases`` (each case, in the order of the file names). Two + cases with the same id stop the run. + """ + + def call(self, payload: Payload) -> Payload: + return payload.insert("cases", self.load(payload.get("cases_dir"))) + + def load(self, cases_dir) -> list: + cases = [] + for path in sorted(pathlib.Path(cases_dir).glob("*.json")): + cases.extend(json.loads(path.read_text())) + ids = [case["id"] for case in cases] + duplicates = {i for i in ids if ids.count(i) > 1} + if duplicates: + raise SystemExit(f"duplicate case ids: {sorted(duplicates)}") + return cases diff --git a/conformance/judging/no_targets_named_predicate.py b/conformance/judging/no_targets_named_predicate.py new file mode 100644 index 0000000..9adbeda --- /dev/null +++ b/conformance/judging/no_targets_named_predicate.py @@ -0,0 +1,10 @@ +"""NoTargetsNamedPredicate: did the caller name no target?""" + +from codeupipe import Payload + + +class NoTargetsNamedPredicate: + """True when ``names`` is empty: the judge then takes each target that this machine can run.""" + + def test(self, payload: Payload) -> bool: + return not payload.get("names") diff --git a/conformance/judging/node_probe.py b/conformance/judging/node_probe.py new file mode 100644 index 0000000..80bd764 --- /dev/null +++ b/conformance/judging/node_probe.py @@ -0,0 +1,20 @@ +"""NodeProbe: what this machine has for the TypeScript targets.""" + +import shutil +import subprocess + + +# cup: allow-plain the edge to the machine: it asks which Node is installed +class NodeProbe: + """Says if there is a Node that runs a ``.ts`` file as it is (22.18 or later).""" + + def runs_typescript(self) -> bool: + node = shutil.which("node") + if node is None: + return False + try: + version = subprocess.run([node, "--version"], capture_output=True, text=True, timeout=30).stdout + major, minor = (int(part) for part in version.strip().lstrip("v").split(".")[:2]) + except (OSError, ValueError, subprocess.SubprocessError): + return False + return (major, minor) >= (22, 18) diff --git a/conformance/judging/process_launcher.py b/conformance/judging/process_launcher.py new file mode 100644 index 0000000..70b76f8 --- /dev/null +++ b/conformance/judging/process_launcher.py @@ -0,0 +1,29 @@ +"""ProcessLauncher: starts the runner of one target and reads what it reports.""" + +import json +import os +import subprocess + + +# cup: allow-plain the edge to the runners: it starts a process, gives it the cases and reads its report +class ProcessLauncher: + """Runs one target. ``spec`` has ``cmd``, ``env`` and ``cwd``. + + Gives ``(results, None)``, with ``results`` as {case id: result}, or + ``(None, reason)`` when the runner did not end well or its output + cannot be read. + """ + + def run(self, spec, cases): + env = dict(os.environ) + env.update(spec["env"]) + proc = subprocess.run( + spec["cmd"], input=json.dumps(cases), capture_output=True, text=True, + cwd=spec["cwd"], env=env, timeout=600, + ) + if proc.returncode != 0: + return None, f"runner exited {proc.returncode}: {proc.stderr.strip()[-2000:]}" + try: + return {r["id"]: r for r in json.loads(proc.stdout)}, None + except (json.JSONDecodeError, KeyError) as error: + return None, f"unreadable runner output ({error}): {proc.stdout[:500]}" diff --git a/conformance/judging/run_targets_filter.py b/conformance/judging/run_targets_filter.py new file mode 100644 index 0000000..2f7c56e --- /dev/null +++ b/conformance/judging/run_targets_filter.py @@ -0,0 +1,29 @@ +"""RunTargetsFilter: run the cases on each target.""" + +from codeupipe import Payload + + +class RunTargetsFilter: + """Filter (sync). + + Input keys: ``names`` (the targets to run, in order), ``targets`` + ({name: how to start its runner}), ``cases``. + Output keys: ``results`` ({name: {case id: result}}) for each target that + ran, and ``target_errors`` ({name: reason}) for each one that did not. + + ``launcher`` runs one target: ``launcher.run(spec, cases)``. + """ + + def __init__(self, launcher): + self.launcher = launcher + + def call(self, payload: Payload) -> Payload: + targets, cases = payload.get("targets"), payload.get("cases") + results, errors = {}, {} + for name in payload.get("names"): + found, error = self.launcher.run(targets[name], cases) + if error: + errors[name] = error + else: + results[name] = found + return payload.insert("results", results).insert("target_errors", errors) diff --git a/conformance/judging/summarize_filter.py b/conformance/judging/summarize_filter.py new file mode 100644 index 0000000..2d060cc --- /dev/null +++ b/conformance/judging/summarize_filter.py @@ -0,0 +1,37 @@ +"""SummarizeFilter: the counts for each target, the matrix, and the exit code.""" + +from codeupipe import Payload + + +class SummarizeFilter: + """Filter (sync). + + Input keys: ``cases``, ``names``, ``verdicts``, ``target_errors``. + Output keys: ``report`` (the number of cases, and for each target its + counts, its failures and the cases that it cannot express, or the + reason that it did not run), ``matrix`` ({case id: {name: status}}) and + ``exit_code`` (1 when a case failed or a target did not run). + """ + + def call(self, payload: Payload) -> Payload: + cases, verdicts, errors = payload.get("cases"), payload.get("verdicts"), payload.get("target_errors") + report = {"cases": len(cases), "targets": {}} + matrix = {case["id"]: {} for case in cases} + for name in payload.get("names"): + if name in errors: + report["targets"][name] = {"error": errors[name]} + for case in cases: + matrix[case["id"]][name] = "—" + continue + summary = {"pass": 0, "partial": 0, "FAIL": 0, "unsupported": 0, "failures": {}, "unsupported_cases": {}} + for case in cases: + status, problems = verdicts[name][case["id"]] + summary[status] += 1 + matrix[case["id"]][name] = status + if status == "FAIL": + summary["failures"][case["id"]] = problems + elif status == "unsupported": + summary["unsupported_cases"][case["id"]] = problems[0] + report["targets"][name] = summary + failed = any(target.get("FAIL") or "error" in target for target in report["targets"].values()) + return payload.insert("report", report).insert("matrix", matrix).insert("exit_code", 1 if failed else 0) diff --git a/conformance/judging/write_results_tap.py b/conformance/judging/write_results_tap.py new file mode 100644 index 0000000..5bacf03 --- /dev/null +++ b/conformance/judging/write_results_tap.py @@ -0,0 +1,20 @@ +"""WriteResultsTap: put the report and the matrix on disk.""" + +import json +import pathlib + +from codeupipe import Payload + + +class WriteResultsTap: + """Tap (sync). + + Reads ``results_dir``, ``report`` and ``markdown``, and writes + ``latest.json`` and ``latest.md`` in that directory. + """ + + def observe(self, payload: Payload) -> None: + directory = pathlib.Path(payload.get("results_dir")) + directory.mkdir(exist_ok=True) + (directory / "latest.json").write_text(json.dumps(payload.get("report"), indent=2)) + (directory / "latest.md").write_text(payload.get("markdown")) diff --git a/conformance/run.py b/conformance/run.py index ed10c61..9e11f24 100644 --- a/conformance/run.py +++ b/conformance/run.py @@ -19,17 +19,23 @@ nested pipeline); reported, never counted as a pass """ -import json -import os import pathlib -import shutil -import subprocess import sys HERE = pathlib.Path(__file__).resolve().parent REPO = HERE.parent CASES = HERE / "cases" RESULTS = HERE / "results" +sys.path.insert(0, str(HERE)) + +from codeupipe import Payload # noqa: E402 + +from judging.default_targets_filter import DefaultTargetsFilter # noqa: E402 +from judging.judge_pipeline import JudgePipeline # noqa: E402 +from judging.judge_results_filter import JudgeResultsFilter # noqa: E402 +from judging.load_cases_filter import LoadCasesFilter # noqa: E402 +from judging.node_probe import NodeProbe # noqa: E402 +from judging.process_launcher import ProcessLauncher # noqa: E402 PYTHON = sys.executable @@ -51,20 +57,6 @@ def _typescript(target): } -# cup: allow-plain the judge of the conformance suite: it runs each target and compares each result with what the case expects -def _node_runs_typescript(): - """Whether there is a Node that runs a .ts file as it is (22.18 or later).""" - node = shutil.which("node") - if node is None: - return False - try: - version = subprocess.run([node, "--version"], capture_output=True, text=True, timeout=30).stdout - major, minor = (int(part) for part in version.strip().lstrip("v").split(".")[:2]) - except (OSError, ValueError, subprocess.SubprocessError): - return False - return (major, minor) >= (22, 18) - - TARGETS = { "python:run": _python("run"), "python:run_async": _python("run_async"), @@ -84,158 +76,44 @@ def _node_runs_typescript(): } -# cup: allow-plain the judge of the conformance suite: it runs each target and compares each result with what the case expects +# The judge is the pipeline in judging/. The functions below give its parts to the tests and to bench.py +# under the names that they had. Each one calls a step of the pipeline, so there is one judge. + +def _node_runs_typescript(): + """Whether there is a Node that runs a .ts file as it is (22.18 or later).""" + return NodeProbe().runs_typescript() + + def default_targets(): """Every target that this machine can run. TypeScript needs Node, and the static TypeScript target needs its package to be installed.""" - def can_run(name): - if not name.startswith("typescript:"): - return True - return _node_runs_typescript() and all(path.exists() for path in _TYPESCRIPT_NEEDS.get(name, [])) - return [name for name in TARGETS if can_run(name)] + return DefaultTargetsFilter(lambda: _node_runs_typescript()).choose(TARGETS, _TYPESCRIPT_NEEDS) -# cup: allow-plain the judge of the conformance suite: it runs each target and compares each result with what the case expects def load_cases(): - cases = [] - for path in sorted(CASES.glob("*.json")): - cases.extend(json.loads(path.read_text())) - ids = [c["id"] for c in cases] - duplicates = {i for i in ids if ids.count(i) > 1} - if duplicates: - raise SystemExit(f"duplicate case ids: {sorted(duplicates)}") - return cases + return LoadCasesFilter().load(CASES) -# cup: allow-plain the judge of the conformance suite: it runs each target and compares each result with what the case expects def run_target(name, spec, cases): - env = dict(os.environ) - env.update(spec["env"]) - proc = subprocess.run( - spec["cmd"], input=json.dumps(cases), capture_output=True, text=True, - cwd=spec["cwd"], env=env, timeout=600, - ) - if proc.returncode != 0: - return None, f"runner exited {proc.returncode}: {proc.stderr.strip()[-2000:]}" - try: - return {r["id"]: r for r in json.loads(proc.stdout)}, None - except (json.JSONDecodeError, KeyError) as error: - return None, f"unreadable runner output ({error}): {proc.stdout[:500]}" - - -# cup: allow-plain the judge of the conformance suite: it runs each target and compares each result with what the case expects + return ProcessLauncher().run(spec, cases) + + def _norm(value): """JSON equality: 1 == 1.0, but true is not 1.""" - if isinstance(value, bool) or value is None or isinstance(value, str): - return value - if isinstance(value, (int, float)): - return float(value) - if isinstance(value, list): - return [_norm(v) for v in value] - if isinstance(value, dict): - return {k: _norm(v) for k, v in value.items()} - return value - - -# cup: allow-plain the judge of the conformance suite: it runs each target and compares each result with what the case expects + return JudgeResultsFilter().norm(value) + + def judge(case, result): """Return (status, problems).""" - if result is None: - return "FAIL", ["no result from runner"] - if "unsupported" in result: - return "unsupported", [result["unsupported"]] - expect = case["expect"] - problems = [] - - if "output" in expect: - if "error" in result: - problems.append(f"expected output, got error {result['error']!r}") - elif _norm(result.get("output")) != _norm(expect["output"]): - problems.append(f"output {result.get('output')!r} != {expect['output']!r}") - if "error" in expect: - if "error" not in result: - problems.append(f"expected an error, got output {result.get('output')!r}") - elif expect["error"].get("message_contains", "") not in result["error"]: - problems.append(f"error {result['error']!r} lacks {expect['error']['message_contains']!r}") - - if _norm(result.get("input_after")) != _norm(case["input"]): - problems.append(f"input was mutated: {result.get('input_after')!r}") - - if "taps" in expect and _norm(result.get("taps")) != _norm(expect["taps"]): - problems.append(f"taps {result.get('taps')!r} != {expect['taps']!r}") - - if "state" in expect and "state" in result: - for key, want in expect["state"].items(): - got = result["state"].get(key) - if got != want: - problems.append(f"state.{key} {got!r} != {want!r}") - - for key in ("inner_states", "pipeline_state"): - if key in expect and key in result and result[key] != expect[key]: - problems.append(f"{key} {result[key]!r} != {expect[key]!r}") - - if "hooks" in expect and "hooks" in result and result["hooks"] != expect["hooks"]: - problems.append(f"hooks {result['hooks']!r} != {expect['hooks']!r}") - - if problems: - return "FAIL", problems - unchecked = [k for k in ("state", "hooks", "inner_states", "pipeline_state") if k in expect and k not in result] - if unchecked: - return "partial", [f"target cannot observe {', '.join(unchecked)}"] - return "pass", [] - - -# cup: allow-plain the judge of the conformance suite: it runs each target and compares each result with what the case expects + return JudgeResultsFilter().judge(case, result) + + def main(argv): - cases = load_cases() - names = argv or default_targets() - report = {"cases": len(cases), "targets": {}} - matrix = {c["id"]: {} for c in cases} - - for name in names: - results, error = run_target(name, TARGETS[name], cases) - if error: - report["targets"][name] = {"error": error} - for c in cases: - matrix[c["id"]][name] = "—" - continue - summary = {"pass": 0, "partial": 0, "FAIL": 0, "unsupported": 0, "failures": {}, "unsupported_cases": {}} - for c in cases: - status, problems = judge(c, results.get(c["id"])) - summary[status] += 1 - matrix[c["id"]][name] = status - if status == "FAIL": - summary["failures"][c["id"]] = problems - elif status == "unsupported": - summary["unsupported_cases"][c["id"]] = problems[0] - report["targets"][name] = summary - - RESULTS.mkdir(exist_ok=True) - (RESULTS / "latest.json").write_text(json.dumps(report, indent=2)) - (RESULTS / "latest.md").write_text(_markdown(cases, names, matrix, report)) - print(_markdown(cases, names, matrix, report)) - failed = any(t.get("FAIL") or "error" in t for t in report["targets"].values()) - return 1 if failed else 0 - - -# cup: allow-plain the judge of the conformance suite: it runs each target and compares each result with what the case expects -def _markdown(cases, names, matrix, report): - mark = {"pass": "✅", "partial": "✅*", "FAIL": "❌", "unsupported": "➖", "—": "—"} - lines = ["| case | " + " | ".join(names) + " |", "|---|" + "---|" * len(names)] - for c in cases: - lines.append(f"| `{c['id']}` | " + " | ".join(mark[matrix[c['id']][n]] for n in names) + " |") - lines.append("") - for name in names: - t = report["targets"][name] - if "error" in t: - lines.append(f"- **{name}**: not run: {t['error']}") - else: - lines.append(f"- **{name}**: {t['pass']} pass, {t['partial']} pass*, {t['FAIL']} fail, " - f"{t['unsupported']} unsupported") - lines.append("") - lines.append("✅* passed everything the target can observe; it does not expose State or Hooks " - "(the fast mode). ➖ the target cannot express the case.") - return "\n".join(lines) + "\n" + pipeline = JudgePipeline(ProcessLauncher(), lambda: _node_runs_typescript()) + out = pipeline.run(Payload({"cases_dir": CASES, "results_dir": RESULTS, "targets": TARGETS, + "needs": _TYPESCRIPT_NEEDS, "names": list(argv)})) + print(out.get("markdown")) + return out.get("exit_code") if __name__ == "__main__": diff --git a/cup.config.json b/cup.config.json index e2b5f45..8da157a 100644 --- a/cup.config.json +++ b/cup.config.json @@ -1,6 +1,6 @@ { "lint": { - "paths": ["packages/codeupipe/codeupipe", "packages/codeupipe-lint/codeupipe_lint", "scorecard", "conformance/runners/python", "examples"], + "paths": ["packages/codeupipe/codeupipe", "packages/codeupipe-lint/codeupipe_lint", "scorecard", "conformance/runners/python", "conformance/judging", "conformance/run.py", "examples"], "exclude": ["*/tests/lint/fixtures/*"] } } diff --git a/tests/conformance/test_judge_pipeline.py b/tests/conformance/test_judge_pipeline.py new file mode 100644 index 0000000..d9049fe --- /dev/null +++ b/tests/conformance/test_judge_pipeline.py @@ -0,0 +1,113 @@ +"""The judge is a pipeline, so each of its steps is tested alone, with no runner and no process.""" + +import json +import pathlib +import sys + +import pytest + +CONFORMANCE = pathlib.Path(__file__).resolve().parents[2] / "conformance" +sys.path.insert(0, str(CONFORMANCE)) + +from codeupipe import Payload # noqa: E402 + +from judging.default_targets_filter import DefaultTargetsFilter # noqa: E402 +from judging.judge_pipeline import JudgePipeline # noqa: E402 +from judging.judge_results_filter import JudgeResultsFilter # noqa: E402 +from judging.load_cases_filter import LoadCasesFilter # noqa: E402 + +CASE = {"id": "a/one", "steps": [], "input": {"n": 1}, + "expect": {"output": {"n": 2}, "state": {"executed": ["s"], "skipped": [], "errors": []}, "hooks": ["before:"]}} +GOOD = {"id": "a/one", "output": {"n": 2.0}, "state": {"executed": ["s"], "skipped": [], "errors": []}, + "hooks": ["before:"], "taps": [], "input_after": {"n": 1}} + + +def without(result, *keys): + return {key: value for key, value in result.items() if key not in keys} + + +@pytest.mark.parametrize("result, status, problem", [ + (GOOD, "pass", None), + (None, "FAIL", "no result from runner"), + ({"id": "a/one", "unsupported": "no nested pipeline"}, "unsupported", "no nested pipeline"), + (without(GOOD, "state", "hooks"), "partial", "target cannot observe state, hooks"), + ({**GOOD, "output": {"n": 3}}, "FAIL", "output {'n': 3} != {'n': 2}"), + ({**GOOD, "output": {"n": True}}, "FAIL", "output {'n': True} != {'n': 2}"), + ({**without(GOOD, "output"), "error": "ValueError: no"}, "FAIL", "expected output, got error 'ValueError: no'"), + ({**GOOD, "input_after": {"n": 9}}, "FAIL", "input was mutated: {'n': 9}"), + ({**GOOD, "state": {"executed": [], "skipped": [], "errors": []}}, "FAIL", "state.executed [] != ['s']"), + ({**GOOD, "hooks": []}, "FAIL", "hooks [] != ['before:']"), +]) +def test_the_verdict_for_one_result(result, status, problem): + got_status, problems = JudgeResultsFilter().judge(CASE, result) + assert got_status == status + assert problems == ([] if problem is None else [problem]) + + +def test_an_error_is_expected_by_a_part_of_its_message(): + case = {"id": "a/two", "steps": [], "input": {}, "expect": {"error": {"message_contains": "boom"}}} + judge = JudgeResultsFilter().judge + assert judge(case, {"error": "ValueError: boom!", "input_after": {}})[0] == "pass" + assert judge(case, {"error": "ValueError: other", "input_after": {}})[0] == "FAIL" + assert judge(case, {"output": {}, "input_after": {}})[0] == "FAIL" + + +def test_a_python_target_always_runs_and_a_typescript_target_needs_node_and_its_files(tmp_path): + there = tmp_path / "built.js" + there.write_text("") + targets = {"python:run": {}, "typescript:run": {}, "typescript:static": {}, "typescript:other": {}} + needs = {"typescript:static": [there], "typescript:other": [tmp_path / "missing.js"]} + assert DefaultTargetsFilter(lambda: True).choose(targets, needs) == ["python:run", "typescript:run", "typescript:static"] + assert DefaultTargetsFilter(lambda: False).choose(targets, needs) == ["python:run"] + + +def test_two_cases_with_one_id_stop_the_run(tmp_path): + (tmp_path / "a.json").write_text(json.dumps([{"id": "x"}, {"id": "x"}])) + with pytest.raises(SystemExit, match="duplicate case ids"): + LoadCasesFilter().load(tmp_path) + + +class FakeLauncher: + """A launcher that starts no process: it gives the results that the test made.""" + + def __init__(self, by_command): + self.by_command = by_command + + def run(self, spec, cases): + return self.by_command[spec["cmd"]] + + +def test_the_whole_judge_with_no_process(tmp_path): + cases_dir, results_dir = tmp_path / "cases", tmp_path / "results" + cases_dir.mkdir() + (cases_dir / "a.json").write_text(json.dumps([CASE])) + launcher = FakeLauncher({ + "good": ({"a/one": GOOD}, None), + "wrong": ({"a/one": {**GOOD, "output": {"n": 3}}}, None), + "broken": (None, "runner exited 3: no"), + }) + targets = {"t:good": {"cmd": "good"}, "t:wrong": {"cmd": "wrong"}, "t:broken": {"cmd": "broken"}} + pipeline = JudgePipeline(launcher, lambda: False) + + def run(names): + return pipeline.run(Payload({"cases_dir": cases_dir, "results_dir": results_dir, "targets": targets, + "needs": {}, "names": names})) + + out = run(["t:good"]) + assert out.get("exit_code") == 0 + assert out.get("matrix") == {"a/one": {"t:good": "pass"}} + assert json.loads((results_dir / "latest.json").read_text()) == out.get("report") + assert (results_dir / "latest.md").read_text() == out.get("markdown") + + out = run(["t:good", "t:wrong", "t:broken"]) + assert out.get("exit_code") == 1 + assert out.get("matrix") == {"a/one": {"t:good": "pass", "t:wrong": "FAIL", "t:broken": "—"}} + assert out.get("report")["targets"]["t:wrong"]["failures"] == {"a/one": ["output {'n': 3} != {'n': 2}"]} + assert out.get("report")["targets"]["t:broken"] == {"error": "runner exited 3: no"} + assert "- **t:broken**: not run: runner exited 3: no" in out.get("markdown") + + # With no target named, the judge takes each target that the machine can run. + assert set(run([]).get("matrix")["a/one"]) == {"t:good", "t:wrong", "t:broken"} + assert pipeline.state.executed[1] == "DefaultTargetsFilter" + run(["t:good"]) + assert pipeline.state.skipped == ["DefaultTargetsFilter"] diff --git a/tests/lint/invariants/test_repo_is_clean.py b/tests/lint/invariants/test_repo_is_clean.py index 5456e21..0fb44b3 100644 --- a/tests/lint/invariants/test_repo_is_clean.py +++ b/tests/lint/invariants/test_repo_is_clean.py @@ -7,7 +7,7 @@ from ._support import REPO # The places with a `cup: allow-plain` comment. -PLAIN_BY_DESIGN = 81 +PLAIN_BY_DESIGN = 75 CHECKED = ["packages/codeupipe/codeupipe", "conformance", "scorecard", "packages/codeupipe-lint/codeupipe_lint", "tests", "examples"] From c27cb15af3d93eab879036a2627c72dbf94155f8 Mon Sep 17 00:00:00 2001 From: Joshua Wink <60934381+JoshuaWink@users.noreply.github.com> Date: Tue, 13 Oct 2026 13:57:31 -0500 Subject: [PATCH 2/2] refactor(conformance): the Python runner is codeupipe pipelines (#16) * refactor(conformance): the Python runner is codeupipe pipelines The runner was 15 plain functions in runner.py. It is now two pipelines in conformance/runners/python/case_runner: one plays a case on a target and one times a scenario. They share the steps that read the target, write the Predicates, build the pipeline and compile it. Each step is a primitive in its own file with tests of its own. runner.py is the command line. The new runner was run beside the old one on each Python target, with all 30 cases and 7 more that were made to fail. The reports are the same byte for byte. In the timing mode the outputs and the refusals are the same, and the times are the same within the noise of a run. Co-Authored-By: Claude Opus 5.5 * fix(scorecard): use the steps of the new runner The differential test tool used two functions of the old runner file. It now uses the same steps that the runner uses. Co-Authored-By: Claude Opus 5.5 --------- Co-authored-by: Claude Opus 5.5 --- conformance/README.md | 5 + .../runners/python/case_runner/__init__.py | 6 + .../case_runner/attach_noop_hook_filter.py | 19 + .../case_runner/attach_record_hook_filter.py | 22 + .../case_runner/bench_each_scenario_filter.py | 38 ++ .../case_runner/bench_scenario_pipeline.py | 30 ++ .../case_runner/build_pipeline_filter.py | 79 ++++ .../python/case_runner/export_filter.py | 48 +++ .../case_runner/finish_result_filter.py | 16 + .../case_runner/is_compiled_predicate.py | 13 + .../case_runner/make_run_function_filter.py | 33 ++ .../python/case_runner/measure_filter.py | 61 +++ .../python/case_runner/name_steps_filter.py | 35 ++ .../python/case_runner/parse_target_filter.py | 36 ++ .../python/case_runner/run_case_pipeline.py | 36 ++ .../python/case_runner/run_compiled_filter.py | 55 +++ .../python/case_runner/run_dynamic_filter.py | 56 +++ .../case_runner/run_each_case_filter.py | 40 ++ .../python/case_runner/runner_pipeline.py | 15 + .../runners/python/case_runner/unsupported.py | 9 + .../case_runner/write_predicates_filter.py | 55 +++ conformance/runners/python/runner.py | 386 +----------------- scorecard/vocab.py | 22 +- tests/conformance/test_case_runner.py | 110 +++++ tests/lint/invariants/test_repo_is_clean.py | 11 +- 25 files changed, 857 insertions(+), 379 deletions(-) create mode 100644 conformance/runners/python/case_runner/__init__.py create mode 100644 conformance/runners/python/case_runner/attach_noop_hook_filter.py create mode 100644 conformance/runners/python/case_runner/attach_record_hook_filter.py create mode 100644 conformance/runners/python/case_runner/bench_each_scenario_filter.py create mode 100644 conformance/runners/python/case_runner/bench_scenario_pipeline.py create mode 100644 conformance/runners/python/case_runner/build_pipeline_filter.py create mode 100644 conformance/runners/python/case_runner/export_filter.py create mode 100644 conformance/runners/python/case_runner/finish_result_filter.py create mode 100644 conformance/runners/python/case_runner/is_compiled_predicate.py create mode 100644 conformance/runners/python/case_runner/make_run_function_filter.py create mode 100644 conformance/runners/python/case_runner/measure_filter.py create mode 100644 conformance/runners/python/case_runner/name_steps_filter.py create mode 100644 conformance/runners/python/case_runner/parse_target_filter.py create mode 100644 conformance/runners/python/case_runner/run_case_pipeline.py create mode 100644 conformance/runners/python/case_runner/run_compiled_filter.py create mode 100644 conformance/runners/python/case_runner/run_dynamic_filter.py create mode 100644 conformance/runners/python/case_runner/run_each_case_filter.py create mode 100644 conformance/runners/python/case_runner/runner_pipeline.py create mode 100644 conformance/runners/python/case_runner/unsupported.py create mode 100644 conformance/runners/python/case_runner/write_predicates_filter.py create mode 100644 tests/conformance/test_case_runner.py diff --git a/conformance/README.md b/conformance/README.md index eb94ad2..307cd0f 100644 --- a/conformance/README.md +++ b/conformance/README.md @@ -7,6 +7,11 @@ One set of language-neutral cases, a runner, and **one judge** (`run.py`). The judge is itself a codeupipe pipeline, in [judging/](judging/): one step reads the cases, one runs each target, one compares, and one writes the matrix. Each step is tested alone in `tests/conformance/test_judge_pipeline.py`. + +The Python runner is built the same way, in +[runners/python/case_runner/](runners/python/case_runner/): one pipeline plays +a case and one times a scenario. Its steps are tested alone in +`tests/conformance/test_case_runner.py`. The dynamic run and each compiled form are held to exactly the same comparison. diff --git a/conformance/runners/python/case_runner/__init__.py b/conformance/runners/python/case_runner/__init__.py new file mode 100644 index 0000000..a453e76 --- /dev/null +++ b/conformance/runners/python/case_runner/__init__.py @@ -0,0 +1,6 @@ +"""The Python conformance runner, as codeupipe pipelines. + +``run_case_pipeline.py`` plays one case on one target and reports what +happened. ``bench_scenario_pipeline.py`` times one scenario. A runner never +judges: ``conformance/judging`` does. +""" diff --git a/conformance/runners/python/case_runner/attach_noop_hook_filter.py b/conformance/runners/python/case_runner/attach_noop_hook_filter.py new file mode 100644 index 0000000..8170d41 --- /dev/null +++ b/conformance/runners/python/case_runner/attach_noop_hook_filter.py @@ -0,0 +1,19 @@ +"""AttachNoopHookFilter: a Hook that does nothing, for a scenario that measures the cost of Hooks.""" + +import cup_ops as ops +from codeupipe import Payload + + +class AttachNoopHookFilter: + """Filter (sync). + + Input keys: ``case`` (a scenario), ``pipeline``. + Output keys: ``hooks``: a list with one NoopHook for a scenario with + ``"hooks": "noop"``, or None. The pipeline gets its own NoopHook. + """ + + def call(self, payload: Payload) -> Payload: + if payload.get("case").get("hooks") != "noop": + return payload.insert("hooks", None) + payload.get("pipeline").use_hook(ops.NoopHook()) + return payload.insert("hooks", [ops.NoopHook()]) diff --git a/conformance/runners/python/case_runner/attach_record_hook_filter.py b/conformance/runners/python/case_runner/attach_record_hook_filter.py new file mode 100644 index 0000000..314be3b --- /dev/null +++ b/conformance/runners/python/case_runner/attach_record_hook_filter.py @@ -0,0 +1,22 @@ +"""AttachRecordHookFilter: the Hook that records the lifecycle, for a case that asks for it.""" + +import cup_ops as ops +from codeupipe import Payload + + +class AttachRecordHookFilter: + """Filter (sync). + + Input keys: ``case``, ``pipeline``, ``compiled``, ``observe``. + Output keys: ``hooks``: the Hooks of the run, or None. A case with + ``"hooks"`` gets a RecordHook, on the pipeline too. Compiled code calls + Hooks only when it was compiled with ``observe``. + """ + + def call(self, payload: Payload) -> Payload: + wanted = payload.get("case").get("hooks") and (not payload.get("compiled") or payload.get("observe")) + if not wanted: + return payload.insert("hooks", None) + hook = ops.RecordHook() + payload.get("pipeline").use_hook(hook) + return payload.insert("hooks", [hook]) diff --git a/conformance/runners/python/case_runner/bench_each_scenario_filter.py b/conformance/runners/python/case_runner/bench_each_scenario_filter.py new file mode 100644 index 0000000..f176275 --- /dev/null +++ b/conformance/runners/python/case_runner/bench_each_scenario_filter.py @@ -0,0 +1,38 @@ +"""BenchEachScenarioFilter: time each scenario, one after the other.""" + +import pathlib +import tempfile + +from codeupipe import Payload + +from .unsupported import ExportRefused + + +class BenchEachScenarioFilter: + """Filter (sync). + + Input keys: ``cases`` (the scenarios), ``target``. + Output keys: ``results``: for each scenario ``id``, ``us_per_run``, + ``relative`` and ``output``, or ``{"id", "unsupported": reason}`` when + the compiler refuses its pipeline. + + ``each`` is the pipeline that times one scenario. + """ + + def __init__(self, each): + self.each = each + + def call(self, payload: Payload) -> Payload: + target = payload.get("target") + results = [self.one(case, target, index) for index, case in enumerate(payload.get("cases"))] + return payload.insert("results", results) + + def one(self, case, target, index) -> dict: + workdir = pathlib.Path(tempfile.mkdtemp(prefix="cup_bench_")) + try: + out = self.each.run(Payload({"case": case, "target": target, "workdir": workdir, + "module_name": f"bench_predicates_{index}"})) + except ExportRefused as reason: + return {"id": case["id"], "unsupported": str(reason)} + return {"id": case["id"], "us_per_run": out.get("us_per_run"), "relative": out.get("relative"), + "output": out.get("output")} diff --git a/conformance/runners/python/case_runner/bench_scenario_pipeline.py b/conformance/runners/python/case_runner/bench_scenario_pipeline.py new file mode 100644 index 0000000..fd8a41a --- /dev/null +++ b/conformance/runners/python/case_runner/bench_scenario_pipeline.py @@ -0,0 +1,30 @@ +"""BenchScenarioPipeline: time one scenario on one target.""" + +from codeupipe import Pipeline + +from .attach_noop_hook_filter import AttachNoopHookFilter +from .build_pipeline_filter import BuildPipelineFilter +from .export_filter import ExportFilter +from .is_compiled_predicate import IsCompiledPredicate +from .make_run_function_filter import MakeRunFunctionFilter +from .measure_filter import MeasureFilter +from .parse_target_filter import ParseTargetFilter +from .write_predicates_filter import WritePredicatesFilter + +TARGETS = ("run", "static-steps", "static-inline", "static-observe") + + +class BenchScenarioPipeline(Pipeline): + """Input keys: ``case`` (a scenario), ``target``, ``workdir``, ``module_name``. + Output keys: ``us_per_run``, ``relative``, ``output``. + """ + + def __init__(self): + super().__init__() + self.add_filter(ParseTargetFilter(TARGETS, "bench target")) + self.add_filter(WritePredicatesFilter()) + self.add_filter(BuildPipelineFilter()) + self.add_filter(AttachNoopHookFilter()) + self.add_valve(IsCompiledPredicate(), ExportFilter()) + self.add_filter(MakeRunFunctionFilter()) + self.add_filter(MeasureFilter()) diff --git a/conformance/runners/python/case_runner/build_pipeline_filter.py b/conformance/runners/python/case_runner/build_pipeline_filter.py new file mode 100644 index 0000000..d62bb4a --- /dev/null +++ b/conformance/runners/python/case_runner/build_pipeline_filter.py @@ -0,0 +1,79 @@ +"""BuildPipelineFilter: the pipeline of a case, made from its steps.""" + +import cup_ops as ops +from codeupipe import Payload, Pipeline + +from .unsupported import Unsupported + +# The reference vocabulary (conformance/README.md): for each op, how its step is made from the case. +_FILTERS = { + "set": lambda spec: ops.SetFilter(spec["key"], spec["value"]), + "copy": lambda spec: ops.CopyFilter(spec["from"], spec["to"]), + "add": lambda spec: ops.AddFilter(spec["key"], spec["by"]), + "append": lambda spec: ops.AppendFilter(spec["key"], spec["value"]), + "remove": lambda spec: ops.RemoveFilter(spec["key"]), + "merge": lambda spec: ops.MergeFilter(spec["value"]), + "has": lambda spec: ops.HasFilter(spec["key"], spec["to"]), + "mutate": lambda spec: ops.MutateFilter(spec["key"], spec["value"]), + "fail": lambda spec: ops.FailFilter(spec["message"]), + "run_again": lambda spec: ops.RunAgainFilter(spec["input"]), +} +_ASYNC_FILTERS = {**_FILTERS, "run_again": lambda spec: ops.AsyncRunAgainFilter(spec["input"])} +_TAPS = { + "record": lambda spec: ops.RecordTap(spec["key"]), + "fail": lambda spec: ops.FailTap(spec["message"]), + "noop": lambda spec: ops.NoopTap(), +} + + +class BuildPipelineFilter: + """Filter (sync). + + Input keys: ``case``, ``predicates`` (from WritePredicatesFilter), + ``run_async``. + Output keys: ``pipeline``, and ``label``: for each step label that the + pipeline reports, the name that the case gives that step. Steps are + reported by label (conformance/API.md) and the cases name them, and this + Filter built each step, so it knows which is which. A label that is not + in the map stays as it is, and fails the comparison. + """ + + def call(self, payload: Payload) -> Payload: + names = {} + pipeline = self.build(payload.get("case")["steps"], Pipeline, payload.get("predicates"), + names, payload.get("run_async")) + label = {step_label: names[id(step)] for step_label, step, _type in pipeline._steps} + return payload.insert("pipeline", pipeline).insert("label", label) + + def build(self, steps, cls, predicates, names, run_async=False): + pipeline = cls() + for step in steps: + kind, name = step["kind"], step["name"] + if kind == "filter": + made = self.filter(step, run_async) + pipeline.add_filter(made) + elif kind == "tap": + made = self.tap(step) + pipeline.add_tap(made) + elif kind == "valve": + pipeline.add_valve(getattr(predicates, f"when_{name}"), self.filter(step["filter"])) + made = pipeline._steps[-1][1] # the Valve that the pipeline made: the step that the label names + elif kind == "pipeline": + made = self.build(step["steps"], ops.NestedPipeline, predicates, names) + pipeline.add_pipeline(made) + else: + raise Unsupported(f"unknown step kind {kind!r}") + names[id(made)] = name + return pipeline + + def filter(self, spec, run_async=False): + make = (_ASYNC_FILTERS if run_async else _FILTERS).get(spec["op"]) + if make is None: + raise Unsupported(f"unknown filter op {spec['op']!r}") + return make(spec) + + def tap(self, spec): + make = _TAPS.get(spec["op"]) + if make is None: + raise Unsupported(f"unknown tap op {spec['op']!r}") + return make(spec) diff --git a/conformance/runners/python/case_runner/export_filter.py b/conformance/runners/python/case_runner/export_filter.py new file mode 100644 index 0000000..54b04ca --- /dev/null +++ b/conformance/runners/python/case_runner/export_filter.py @@ -0,0 +1,48 @@ +"""ExportFilter: compile the pipeline of the case and import the package from disk, as a customer would.""" + +import importlib +import sys + +from codeupipe import Payload + +from .unsupported import ExportRefused + + +class ExportFilter: + """Filter (sync). + + Input keys: ``pipeline``, ``lowering``, ``observe``, ``workdir``. + Output keys: ``package``: the compiled code, with its ``run`` function. + + The compiler refuses a pipeline that it cannot keep equal. For the fast + forms that is an ExportRefused: the target cannot express the case. With + ``observe`` the error is given as it is. + """ + + def call(self, payload: Payload) -> Payload: + try: + package = self.load(payload.get("pipeline"), payload.get("lowering"), payload.get("observe"), + payload.get("workdir")) + except Exception as error: # noqa: BLE001 — the compiler refused the shape (for example Hooks) + if payload.get("observe"): + raise + raise ExportRefused(f"{type(error).__name__}: {error}") from error + return payload.insert("package", package) + + def load(self, pipeline, lowering, observe, workdir): + from codeupipe.compile import build_export_pipeline_v2 + + out = build_export_pipeline_v2(write=False).run( + Payload({"pipeline": pipeline, "pattern": "flat", "lowering": lowering, "observe": observe})) + for file in out.get("files") or []: + target = workdir / file["path"] + target.parent.mkdir(parents=True, exist_ok=True) + target.write_text(file["content"]) + package_parent = str(workdir / "src") + for name in [m for m in sys.modules if m == "steps" or m.startswith("steps.")]: + del sys.modules[name] + sys.path.insert(0, package_parent) + try: + return importlib.import_module("steps") + finally: + sys.path.remove(package_parent) diff --git a/conformance/runners/python/case_runner/finish_result_filter.py b/conformance/runners/python/case_runner/finish_result_filter.py new file mode 100644 index 0000000..6f0c991 --- /dev/null +++ b/conformance/runners/python/case_runner/finish_result_filter.py @@ -0,0 +1,16 @@ +"""FinishResultFilter: the id of the case and what its Taps recorded.""" + +import conformance_sink as sink +from codeupipe import Payload + + +class FinishResultFilter: + """Filter (sync). + + Input keys: ``case``, ``result``. + Output keys: ``result``, with ``id`` and ``taps``. + """ + + def call(self, payload: Payload) -> Payload: + result = {**payload.get("result"), "id": payload.get("case")["id"], "taps": list(sink.TAPS)} + return payload.insert("result", result) diff --git a/conformance/runners/python/case_runner/is_compiled_predicate.py b/conformance/runners/python/case_runner/is_compiled_predicate.py new file mode 100644 index 0000000..be83eca --- /dev/null +++ b/conformance/runners/python/case_runner/is_compiled_predicate.py @@ -0,0 +1,13 @@ +"""IsCompiledPredicate: does the case run as compiled code?""" + +from codeupipe import Payload + + +class IsCompiledPredicate: + """True when ``compiled`` is the value that was given (True unless said).""" + + def __init__(self, compiled=True): + self.compiled = compiled + + def test(self, payload: Payload) -> bool: + return payload.get("compiled") is self.compiled diff --git a/conformance/runners/python/case_runner/make_run_function_filter.py b/conformance/runners/python/case_runner/make_run_function_filter.py new file mode 100644 index 0000000..284d34e --- /dev/null +++ b/conformance/runners/python/case_runner/make_run_function_filter.py @@ -0,0 +1,33 @@ +"""MakeRunFunctionFilter: the one function that a measurement calls again and again.""" + +from codeupipe import Payload + + +class MakeRunFunctionFilter: + """Filter (sync). + + Input keys: ``case`` (a scenario), ``compiled``, ``observe``, ``hooks``, + and ``pipeline`` or ``package``. + Output keys: ``fn``: a function with no arguments that does one run of + the scenario on a new copy of its input. ``output``: what one run gives. + """ + + def call(self, payload: Payload) -> Payload: + fn = self.make(payload) + return payload.insert("fn", fn).insert("output", self.output(fn(), payload.get("compiled"))) + + def make(self, payload: Payload): + data = payload.get("case")["input"] + if not payload.get("compiled"): + pipeline = payload.get("pipeline") + return lambda: pipeline.run(Payload(dict(data))) + run = payload.get("package").run + if not payload.get("observe"): + return lambda: run(dict(data)) + hooks = payload.get("hooks") + state = {} + return lambda: run(dict(data), state=state, hooks=hooks) + + def output(self, out, compiled): + """The dynamic run gives a Payload, and compiled code gives a dict.""" + return out if compiled else out.to_dict() diff --git a/conformance/runners/python/case_runner/measure_filter.py b/conformance/runners/python/case_runner/measure_filter.py new file mode 100644 index 0000000..092fa5e --- /dev/null +++ b/conformance/runners/python/case_runner/measure_filter.py @@ -0,0 +1,61 @@ +"""MeasureFilter: how long one run takes.""" + +import statistics +import time + +from codeupipe import Payload + +WARMUP = 20000 +RUNS = 20000 +REPEATS = 5 + +_REFERENCE_KEYS = tuple("k%d" % i for i in range(10)) + + +def _reference_write(data, key): + data[key] = data.get(key, 0) + 1 + return data + + +class MeasureFilter: + """Filter (sync). + + Input keys: ``fn`` (from MakeRunFunctionFilter). + Output keys: ``us_per_run`` (µs for one run: the best of REPEATS x RUNS) + and ``relative`` (the time of ``fn`` as a ratio to a reference workload). + + The relative time is for comparing runs on different machines and under + different load: each repeat times the reference workload and then + ``fn``, back to back, so whatever slows one slows the other, and the + median of the ratios of the repeats is reported. + """ + + def call(self, payload: Payload) -> Payload: + us_per_run, relative = self.measure(payload.get("fn")) + return payload.insert("us_per_run", us_per_run).insert("relative", relative) + + def measure(self, fn): + for _ in range(WARMUP): + fn() + self.reference() + best, ratios = None, [] + for _ in range(REPEATS): + reference = self.timed(self.reference) + elapsed = self.timed(fn) + ratios.append(elapsed / reference) + best = elapsed if best is None else min(best, elapsed) + return best / RUNS * 1e6, statistics.median(ratios) + + def timed(self, fn) -> float: + started = time.perf_counter() + for _ in range(RUNS): + fn() + return time.perf_counter() - started + + @staticmethod + def reference(): + """The reference workload: ten dict writes behind ten function calls, with no framework.""" + data = {"a": 1} + for key in _REFERENCE_KEYS: + data = _reference_write(data, key) + return data diff --git a/conformance/runners/python/case_runner/name_steps_filter.py b/conformance/runners/python/case_runner/name_steps_filter.py new file mode 100644 index 0000000..5661b5a --- /dev/null +++ b/conformance/runners/python/case_runner/name_steps_filter.py @@ -0,0 +1,35 @@ +"""NameStepsFilter: give each step in the result the name that the case gives it.""" + +from codeupipe import Payload + + +class NameStepsFilter: + """Filter (sync). + + Input keys: ``result``, ``label`` ({step label: name in the case}). + Output keys: ``result``, with the name of the case for each step in + ``state``, ``pipeline_state``, ``inner_states`` and ``hooks``. A label + that is not in ``label`` stays as it is. + """ + + def call(self, payload: Payload) -> Payload: + label = payload.get("label") + named = {key: self.named(key, value, label) for key, value in payload.get("result").items()} + return payload.insert("result", named) + + def named(self, key, value, label): + """The value of one key of the result, with the names of the case where it has steps.""" + if key in ("state", "pipeline_state"): + return self.state(value, label) + if key == "inner_states": + return [self.state(one, label) for one in value] + if key == "hooks": + return [self.event(event, label) for event in value] + return value + + def state(self, state, label) -> dict: + return {key: [label.get(step, step) for step in steps] for key, steps in state.items()} + + def event(self, event, label) -> str: + phase, _, step = event.partition(":") + return f"{phase}:{label.get(step, step)}" diff --git a/conformance/runners/python/case_runner/parse_target_filter.py b/conformance/runners/python/case_runner/parse_target_filter.py new file mode 100644 index 0000000..fab6c17 --- /dev/null +++ b/conformance/runners/python/case_runner/parse_target_filter.py @@ -0,0 +1,36 @@ +"""ParseTargetFilter: what the name of a target means.""" + +from codeupipe import Payload + +_DYNAMIC = {"compiled": False, "run_async": False, "lowering": "", "observe": False} +_TARGETS = { + "run": _DYNAMIC, + "run_async": {**_DYNAMIC, "run_async": True}, + "static-steps": {"compiled": True, "run_async": False, "lowering": "steps", "observe": False}, + "static-inline": {"compiled": True, "run_async": False, "lowering": "inline", "observe": False}, + "static-observe": {"compiled": True, "run_async": False, "lowering": "steps", "observe": True}, +} + + +class ParseTargetFilter: + """Filter (sync). + + Input keys: ``target`` (``run``, ``run_async``, ``static-steps``, + ``static-inline`` or ``static-observe``). + Output keys: ``compiled`` (the case runs as compiled code), ``run_async``, + ``lowering`` (the layout of the compiled code) and ``observe`` (the + compiled code records State and calls Hooks). + + ``allowed`` is the targets that this run takes, and ``what`` is the word + for them in the message for another name. + """ + + def __init__(self, allowed=tuple(_TARGETS), what="target"): + self.allowed = allowed + self.what = what + + def call(self, payload: Payload) -> Payload: + target = payload.get("target") + if target not in self.allowed: + raise SystemExit(f"unknown {self.what} {target!r}") + return payload.merge(Payload(dict(_TARGETS[target]))) diff --git a/conformance/runners/python/case_runner/run_case_pipeline.py b/conformance/runners/python/case_runner/run_case_pipeline.py new file mode 100644 index 0000000..6b72a81 --- /dev/null +++ b/conformance/runners/python/case_runner/run_case_pipeline.py @@ -0,0 +1,36 @@ +"""RunCasePipeline: play one conformance case on one target and report what happened.""" + +from codeupipe import Pipeline + +from .attach_record_hook_filter import AttachRecordHookFilter +from .build_pipeline_filter import BuildPipelineFilter +from .export_filter import ExportFilter +from .finish_result_filter import FinishResultFilter +from .is_compiled_predicate import IsCompiledPredicate +from .name_steps_filter import NameStepsFilter +from .parse_target_filter import ParseTargetFilter +from .run_compiled_filter import RunCompiledFilter +from .run_dynamic_filter import RunDynamicFilter +from .write_predicates_filter import WritePredicatesFilter + + +class RunCasePipeline(Pipeline): + """Input keys: ``case``, ``target``, ``workdir``, ``module_name``. + Output keys: ``result`` (conformance/README.md, "Runner protocol"). + + The runner never judges. A step that raises Unsupported, or any other + error before the case runs, means that the target cannot express the + case: RunEachCaseFilter reports that. + """ + + def __init__(self): + super().__init__() + self.add_filter(ParseTargetFilter()) + self.add_filter(WritePredicatesFilter()) + self.add_filter(BuildPipelineFilter()) + self.add_filter(AttachRecordHookFilter()) + self.add_valve(IsCompiledPredicate(), ExportFilter()) + self.add_valve(IsCompiledPredicate(False), RunDynamicFilter()) + self.add_valve(IsCompiledPredicate(), RunCompiledFilter()) + self.add_filter(NameStepsFilter()) + self.add_filter(FinishResultFilter()) diff --git a/conformance/runners/python/case_runner/run_compiled_filter.py b/conformance/runners/python/case_runner/run_compiled_filter.py new file mode 100644 index 0000000..ebef4d5 --- /dev/null +++ b/conformance/runners/python/case_runner/run_compiled_filter.py @@ -0,0 +1,55 @@ +"""RunCompiledFilter: run the case as compiled code.""" + +import copy + +import conformance_sink as sink +from codeupipe import Payload + + +class RunCompiledFilter: + """Filter (sync). + + Input keys: ``case``, ``package`` (from ExportFilter), ``observe``, + ``hooks``. + Output keys: ``result``: ``output`` or ``error``, and ``input_after``. + With ``observe`` also ``state``, ``inner_states`` and, for a case that + asks for them, ``hooks``. Compiled code is a function, so it has no + ``pipeline_state``. The steps in the result are labels: NameStepsFilter + gives them the names of the case. + """ + + def call(self, payload: Payload) -> Payload: + case, package, observe = payload.get("case"), payload.get("package"), payload.get("observe") + hooks = payload.get("hooks") + data = copy.deepcopy(case["input"]) + result = {} + state = {} + inner_states = [] + + def again(inner): + # The compiled form is a function. To run the pipeline again is to call that function again, + # and the caller gives each call its own State. + if not observe: + return package.run(inner) + inner_states.append({}) + return package.run(inner, state=inner_states[-1], hooks=None) + + sink.RUN_AGAIN[0] = again + try: + if observe: + result["output"] = package.run(data, state=state, hooks=hooks) + else: + result["output"] = package.run(data) + except Exception as error: # noqa: BLE001 + result["error"] = f"{type(error).__name__}: {error}" + if observe: + result["state"] = self.report(state) + result["inner_states"] = [self.report(one) for one in inner_states] + if case.get("hooks"): + result["hooks"] = list(sink.HOOKS) + result["input_after"] = data + return payload.insert("result", result) + + def report(self, state) -> dict: + return {"executed": list(state["executed"]), "skipped": list(state["skipped"]), + "errors": [step for step, _error in state["errors"]]} diff --git a/conformance/runners/python/case_runner/run_dynamic_filter.py b/conformance/runners/python/case_runner/run_dynamic_filter.py new file mode 100644 index 0000000..8bd4d0b --- /dev/null +++ b/conformance/runners/python/case_runner/run_dynamic_filter.py @@ -0,0 +1,56 @@ +"""RunDynamicFilter: run the case with ``Pipeline.run`` or ``Pipeline.run_async``.""" + +import asyncio +import copy + +import conformance_sink as sink +from codeupipe import Payload, State + + +class RunDynamicFilter: + """Filter (sync). + + Input keys: ``case``, ``pipeline``, ``run_async``, ``hooks``. + Output keys: ``result``: ``output`` or ``error``, ``state`` (the State of + the run), ``pipeline_state`` (``state`` on the Pipeline after the run), + ``inner_states`` (the State of each run that a run_again step started), + ``hooks`` for a case that asks for them, and ``input_after``. The steps + in it are labels: NameStepsFilter gives them the names of the case. + """ + + def call(self, payload: Payload) -> Payload: + case, pipeline, run_async = payload.get("case"), payload.get("pipeline"), payload.get("run_async") + data = copy.deepcopy(case["input"]) + # With "given_state", the caller gives the State of the run. If not, the run makes one. + given = [State()] if case.get("given_state") else [] + result = {} + + def again(inner): + state = State() + sink.INNER_STATES.append(state) + return pipeline.run_async(Payload(inner), state) if run_async else pipeline.run(Payload(inner), state) + + sink.RUN_AGAIN[0] = again + try: + if run_async: + out = asyncio.run(pipeline.run_async(Payload(data), *given)) + else: + out = pipeline.run(Payload(data), *given) + result["output"] = out.to_dict() + except Exception as error: # noqa: BLE001 — the error is the result + result["error"] = f"{type(error).__name__}: {error}" + + # The State that these two lines read is that of the pipeline of the case, which this step ran. + # It is what the case asks for, not the State of the run that this step is in. + ran = pipeline.state # cup: allow-state-read the State of the pipeline of the case is the result that this step reports + result["state"] = self.report(given[0] if given else ran) + result["pipeline_state"] = self.report(ran) + result["inner_states"] = [self.report(state) for state in sink.INNER_STATES] + if payload.get("hooks"): + result["hooks"] = list(sink.HOOKS) + result["input_after"] = data + return payload.insert("result", result) + + def report(self, state) -> dict: + return {"executed": list(state.executed), "skipped": list(state.skipped), + "errors": [step for step, _error in state.errors]} diff --git a/conformance/runners/python/case_runner/run_each_case_filter.py b/conformance/runners/python/case_runner/run_each_case_filter.py new file mode 100644 index 0000000..fd3d8e0 --- /dev/null +++ b/conformance/runners/python/case_runner/run_each_case_filter.py @@ -0,0 +1,40 @@ +"""RunEachCaseFilter: play each case, one after the other, each in its own directory.""" + +import pathlib +import tempfile + +import conformance_sink as sink +from codeupipe import Payload + +from .unsupported import Unsupported + + +class RunEachCaseFilter: + """Filter (sync). + + Input keys: ``cases``, ``target``. + Output keys: ``results``: for each case its result, or ``{"id", + "unsupported": reason}`` when the target cannot express it. + + ``each`` is the pipeline that plays one case. + """ + + def __init__(self, each): + self.each = each + + def call(self, payload: Payload) -> Payload: + target = payload.get("target") + results = [self.one(case, target, index) for index, case in enumerate(payload.get("cases"))] + return payload.insert("results", results) + + def one(self, case, target, index) -> dict: + sink.reset() + workdir = pathlib.Path(tempfile.mkdtemp(prefix="cup_conf_")) + try: + out = self.each.run(Payload({"case": case, "target": target, "workdir": workdir, + "module_name": f"conf_predicates_{index}"})) + except Unsupported as reason: + return {"id": case["id"], "unsupported": str(reason)} + except Exception as error: # noqa: BLE001 — for example the compiler refuses a shape + return {"id": case["id"], "unsupported": f"{type(error).__name__}: {error}"} + return out.get("result") diff --git a/conformance/runners/python/case_runner/runner_pipeline.py b/conformance/runners/python/case_runner/runner_pipeline.py new file mode 100644 index 0000000..0ca304c --- /dev/null +++ b/conformance/runners/python/case_runner/runner_pipeline.py @@ -0,0 +1,15 @@ +"""RunnerPipeline: the whole runner for one call: each case, or each scenario, on one target.""" + +from codeupipe import Pipeline + + +class RunnerPipeline(Pipeline): + """Input keys: ``cases``, ``target``. Output keys: ``results``. + + ``each`` is the Filter that goes through the cases: RunEachCaseFilter to + play them, or BenchEachScenarioFilter to time them. + """ + + def __init__(self, each): + super().__init__() + self.add_filter(each) diff --git a/conformance/runners/python/case_runner/unsupported.py b/conformance/runners/python/case_runner/unsupported.py new file mode 100644 index 0000000..0d2cb73 --- /dev/null +++ b/conformance/runners/python/case_runner/unsupported.py @@ -0,0 +1,9 @@ +"""The errors that say "this target cannot express this case".""" + + +class Unsupported(Exception): + """The target cannot express the case. The message is the reason.""" + + +class ExportRefused(Unsupported): + """The compiler refused the pipeline of the case.""" diff --git a/conformance/runners/python/case_runner/write_predicates_filter.py b/conformance/runners/python/case_runner/write_predicates_filter.py new file mode 100644 index 0000000..09145ff --- /dev/null +++ b/conformance/runners/python/case_runner/write_predicates_filter.py @@ -0,0 +1,55 @@ +"""WritePredicatesFilter: the Predicates of a case, as real classes in a module on disk.""" + +import importlib +import json +import sys + +from codeupipe import Payload + +from .unsupported import Unsupported + + +class WritePredicatesFilter: + """Filter (sync). + + Input keys: ``case``, ``workdir`` (a directory for this case), + ``module_name``. + Output keys: ``predicates``: a module that has, for each Valve of the + case, an instance ``when_`` of a Predicate class. + + The Predicates are written as source code and imported, because the + compiler reads the source of a Predicate and refuses anything else. + """ + + def call(self, payload: Payload) -> Payload: + workdir, module_name = payload.get("workdir"), payload.get("module_name") + sources = self.sources(payload.get("case")["steps"]) + (workdir / f"{module_name}.py").write_text("\n\n".join(sources) or "\n") + sys.path.insert(0, str(workdir)) + try: + return payload.insert("predicates", importlib.import_module(module_name)) + finally: + sys.path.remove(str(workdir)) + + def sources(self, steps) -> list: + """The source of the Predicate of each Valve, with those of the nested pipelines.""" + found = [] + for step in steps: + if step["kind"] == "valve": + found.append(self.source(step["name"], step["when"])) + elif step["kind"] == "pipeline": + found.extend(self.sources(step["steps"])) + return found + + def source(self, name, when) -> str: + key = json.dumps(when["key"]) + if when["op"] == "equals": + body = f"payload.get({key}) == {when['value']!r}" + elif when["op"] == "truthy": + body = f"bool(payload.get({key}))" + else: + raise Unsupported(f"unknown valve op {when['op']!r}") + # a predicate object (conformance/API.md, "Valve"): a class with test(), + # plus a module-level instance under the name the builders look up + return (f"class When_{name}:\n def test(self, payload):\n return {body}\n\n\n" + f"when_{name} = When_{name}()\n") diff --git a/conformance/runners/python/runner.py b/conformance/runners/python/runner.py index d450a58..77d5d5a 100644 --- a/conformance/runners/python/runner.py +++ b/conformance/runners/python/runner.py @@ -9,391 +9,45 @@ static-inline codeupipe.compile export, lowering="inline", loaded from disk static-observe codeupipe.compile export, observe=True: State and Hooks too +With ``bench:`` it times the scenarios that it gets, and does not +play cases (conformance/bench.py). + +The runner is two codeupipe pipelines, in ``case_runner/``: one plays a +case, and one times a scenario. This file is its command line. + The runner never judges; conformance/run.py compares every target's results against the same expectations. The fast static targets have no State or Hooks, so they omit those fields and the orchestrator skips those checks; static-observe reports both and is judged on them. """ -import asyncio -import copy -import importlib import json import pathlib import sys -import tempfile HERE = pathlib.Path(__file__).resolve().parent sys.path.insert(0, str(HERE)) -import conformance_sink as sink # noqa: E402 -import cup_ops as ops # noqa: E402 -from codeupipe import Payload, Pipeline, State # noqa: E402 - # noqa: E402 - - -class Unsupported(Exception): - pass - - -# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges -def _filter(spec, run_async=False): - op = spec["op"] - if op == "run_again": - return (ops.AsyncRunAgainFilter if run_async else ops.RunAgainFilter)(spec["input"]) - if op == "set": - return ops.SetFilter(spec["key"], spec["value"]) - if op == "copy": - return ops.CopyFilter(spec["from"], spec["to"]) - if op == "add": - return ops.AddFilter(spec["key"], spec["by"]) - if op == "append": - return ops.AppendFilter(spec["key"], spec["value"]) - if op == "remove": - return ops.RemoveFilter(spec["key"]) - if op == "merge": - return ops.MergeFilter(spec["value"]) - if op == "has": - return ops.HasFilter(spec["key"], spec["to"]) - if op == "mutate": - return ops.MutateFilter(spec["key"], spec["value"]) - if op == "fail": - return ops.FailFilter(spec["message"]) - raise Unsupported(f"unknown filter op {op!r}") - - -# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges -def _tap(spec): - if spec["op"] == "record": - return ops.RecordTap(spec["key"]) - if spec["op"] == "fail": - return ops.FailTap(spec["message"]) - if spec["op"] == "noop": - return ops.NoopTap() - raise Unsupported(f"unknown tap op {spec['op']!r}") - - -# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges -def _predicate_source(name, when): - """Valve predicates are generated as real predicate classes so static - lowering can recover their source (it refuses anything else).""" - key = json.dumps(when["key"]) - if when["op"] == "equals": - body = f"payload.get({key}) == {when['value']!r}" - elif when["op"] == "truthy": - body = f"bool(payload.get({key}))" - else: - raise Unsupported(f"unknown valve op {when['op']!r}") - # a predicate object (conformance/API.md, "Valve"): a class with test(), - # plus a module-level instance under the name the builders look up - return (f"class When_{name}:\n def test(self, payload):\n return {body}\n\n\n" - f"when_{name} = When_{name}()\n") - - -# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges -def _predicates(steps, module_dir, module_name): - sources = [] - - def walk(items): - for step in items: - if step["kind"] == "valve": - sources.append(_predicate_source(step["name"], step["when"])) - elif step["kind"] == "pipeline": - walk(step["steps"]) - - walk(steps) - path = module_dir / f"{module_name}.py" - path.write_text("\n\n".join(sources) or "\n") - sys.path.insert(0, str(module_dir)) - try: - return importlib.import_module(module_name) - finally: - sys.path.remove(str(module_dir)) - - -# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges -def build(case, cls, predicates, names, run_async=False): - p = cls() - for step in case["steps"]: - kind, name = step["kind"], step["name"] - if kind == "filter": - obj = _filter(step, run_async) - p.add_filter(obj) - elif kind == "tap": - obj = _tap(step) - p.add_tap(obj) - elif kind == "valve": - p.add_valve(getattr(predicates, f"when_{name}"), _filter(step["filter"])) - obj = p._steps[-1][1] # the Valve that the pipeline made: the step that labels() names - elif kind == "pipeline": - if not hasattr(p, "add_pipeline"): - raise Unsupported(f"{cls.__name__} has no add_pipeline") - obj = build({"steps": step["steps"]}, ops.NestedPipeline, predicates, names) - p.add_pipeline(obj) - else: - raise Unsupported(f"unknown step kind {kind!r}") - names[id(obj)] = name - return p - - -def labels(p, names): - """label -> the case's step name, for the steps of ``p``. - - Steps are reported by label (conformance/API.md), while the - shared cases name them. The runner built every step, so it knows which - case step each label belongs to. A label reported that is not in - this map (wrong, or inconsistent between modes) stays as is and fails - the comparison. - """ - return {label: names[id(step)] for label, step, _type in p._steps} - - -def _named(steps, label): - return [label.get(step, step) for step in steps] - - -# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges -def _named_hooks(events, label): - out = [] - for event in events: - phase, _, step = event.partition(":") - out.append(f"{phase}:{label.get(step, step)}") - return out - - -# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges -def _dynamic(case, predicates, run_async=False): - names = {} - p = build(case, Pipeline, predicates, names, run_async) - label = labels(p, names) - if case.get("hooks"): - p.use_hook(ops.RecordHook()) - data = copy.deepcopy(case["input"]) - # With "given_state", the caller gives the State of the run. If not, the run makes one. - given = [State()] if case.get("given_state") else [] - result = {} - - def again(inner): - state = State() - sink.INNER_STATES.append(state) - return p.run_async(Payload(inner), state) if run_async else p.run(Payload(inner), state) - - sink.RUN_AGAIN[0] = again - try: - if run_async: - out = asyncio.run(p.run_async(Payload(data), *given)) - else: - out = p.run(Payload(data), *given) - result["output"] = out.to_dict() - except Exception as error: # noqa: BLE001 — the error is the result - result["error"] = f"{type(error).__name__}: {error}" - - def report(state): - return { - "executed": _named(state.executed, label), - "skipped": _named(state.skipped, label), - "errors": _named([step for step, _ in state.errors], label), - } - - result["state"] = report(given[0] if given else p.state) - result["pipeline_state"] = report(p.state) - result["inner_states"] = [report(state) for state in sink.INNER_STATES] - if case.get("hooks"): - result["hooks"] = _named_hooks(sink.HOOKS, label) - result["input_after"] = data - return result - - -# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges -def _load_export(p, lowering, observe, workdir): - """Export ``p`` and import the package from disk, as a customer would.""" - from codeupipe.compile import build_export_pipeline_v2 - - out = build_export_pipeline_v2(write=False).run( - Payload({"pipeline": p, "pattern": "flat", "lowering": lowering, "observe": observe})) - for f in out.get("files") or []: - target = workdir / f["path"] - target.parent.mkdir(parents=True, exist_ok=True) - target.write_text(f["content"]) - package_parent = str(workdir / "src") - for name in [m for m in sys.modules if m == "steps" or m.startswith("steps.")]: - del sys.modules[name] - sys.path.insert(0, package_parent) - try: - return importlib.import_module("steps") - finally: - sys.path.remove(package_parent) - - -# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges -def _static(case, predicates, lowering, workdir, observe=False): - # The compiled form always gets its State from the caller, so "given_state" needs nothing more. - names = {} - p = build(case, Pipeline, predicates, names) - label = labels(p, names) - hooks = [ops.RecordHook()] if observe and case.get("hooks") else None - if hooks: - p.use_hook(hooks[0]) - package = _load_export(p, lowering, observe, workdir) - data = copy.deepcopy(case["input"]) - result = {} - state = {} - inner_states = [] - - def again(inner): - # The compiled form is a function. To run the pipeline again is to call that function again, - # and the caller gives each call its own State. - if not observe: - return package.run(inner) - inner_states.append({}) - return package.run(inner, state=inner_states[-1], hooks=None) - - sink.RUN_AGAIN[0] = again - try: - if observe: - result["output"] = package.run(data, state=state, hooks=hooks) - else: - result["output"] = package.run(data) - except Exception as error: # noqa: BLE001 - result["error"] = f"{type(error).__name__}: {error}" - if observe: - def report(one): - return { - "executed": _named(one["executed"], label), - "skipped": _named(one["skipped"], label), - "errors": _named([step for step, _ in one["errors"]], label), - } - - result["state"] = report(state) - result["inner_states"] = [report(one) for one in inner_states] - if case.get("hooks"): - result["hooks"] = _named_hooks(sink.HOOKS, label) - result["input_after"] = data - return result - - -# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges -def run_case(case, target, index): - sink.reset() - tmp = pathlib.Path(tempfile.mkdtemp(prefix="cup_conf_")) - try: - predicates = _predicates(case["steps"], tmp, f"conf_predicates_{index}") - if target == "run": - result = _dynamic(case, predicates) - elif target == "run_async": - result = _dynamic(case, predicates, run_async=True) - elif target in ("static-steps", "static-inline"): - result = _static(case, predicates, target.split("-", 1)[1], tmp) - elif target == "static-observe": - result = _static(case, predicates, "steps", tmp, observe=True) - else: - raise SystemExit(f"unknown target {target!r}") - except Unsupported as reason: - return {"id": case["id"], "unsupported": str(reason)} - except Exception as error: # noqa: BLE001 — e.g. export refusing a shape - return {"id": case["id"], "unsupported": f"{type(error).__name__}: {error}"} - result["id"] = case["id"] - result["taps"] = list(sink.TAPS) - return result - - -BENCH_WARMUP = 20000 -BENCH_RUNS = 20000 -BENCH_REPEATS = 5 - - -_REFERENCE_KEYS = tuple("k%d" % i for i in range(10)) - - -def _reference_write(data, key): - data[key] = data.get(key, 0) + 1 - return data - - -# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges -def _reference(): - """The reference workload: ten dict writes behind ten function calls, - with no framework. A pipeline's time is reported relative to it.""" - data = {"a": 1} - for key in _REFERENCE_KEYS: - data = _reference_write(data, key) - return data - - -# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges -def _timed(fn): - import time - t0 = time.perf_counter() - for _ in range(BENCH_RUNS): - fn() - return time.perf_counter() - t0 - - -# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges -def _measure(fn): - """(µs per run, time relative to the reference workload). - - µs per run is the best of BENCH_REPEATS x BENCH_RUNS. The relative time - is for comparing runs on different machines and under different load: - each repeat times the reference workload and then ``fn``, back to back, - so whatever slows one slows the other, and the median of the repeats' - ratios is reported. - """ - import statistics - for _ in range(BENCH_WARMUP): - fn() - _reference() - best, ratios = None, [] - for _ in range(BENCH_REPEATS): - reference = _timed(_reference) - elapsed = _timed(fn) - ratios.append(elapsed / reference) - best = elapsed if best is None else min(best, elapsed) - return best / BENCH_RUNS * 1e6, statistics.median(ratios) +from codeupipe import Payload # noqa: E402 +from case_runner.bench_each_scenario_filter import BenchEachScenarioFilter # noqa: E402 +from case_runner.bench_scenario_pipeline import BenchScenarioPipeline # noqa: E402 +from case_runner.run_case_pipeline import RunCasePipeline # noqa: E402 +from case_runner.run_each_case_filter import RunEachCaseFilter # noqa: E402 +from case_runner.runner_pipeline import RunnerPipeline # noqa: E402 -# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges -def bench(target, scenarios): - results = [] - for index, case in enumerate(scenarios): - tmp = pathlib.Path(tempfile.mkdtemp(prefix="cup_bench_")) - predicates = _predicates(case["steps"], tmp, f"bench_predicates_{index}") - p = build(case, Pipeline, predicates, {}) - if case.get("hooks") == "noop": - p.use_hook(ops.NoopHook()) - data = case["input"] - if target == "run": - fn = lambda: p.run(Payload(dict(data))) # noqa: E731 - output = fn().to_dict() - elif target in ("static-inline", "static-steps"): - try: - run = _load_export(p, target.split("-", 1)[1], False, tmp).run - except Exception as error: # noqa: BLE001 — export refused the shape (e.g. Hooks) - results.append({"id": case["id"], "unsupported": f"{type(error).__name__}: {error}"}) - continue - fn = lambda: run(dict(data)) # noqa: E731 - output = fn() - elif target == "static-observe": - run = _load_export(p, "steps", True, tmp).run - hooks = [ops.NoopHook()] if case.get("hooks") == "noop" else None - state = {} - fn = lambda: run(dict(data), state=state, hooks=hooks) # noqa: E731 - output = fn() - else: - raise SystemExit(f"unknown bench target {target!r}") - us_per_run, relative = _measure(fn) - results.append({"id": case["id"], "us_per_run": us_per_run, "relative": relative, "output": output}) - return results +# Is the first argument "bench:"? Then the runner times scenarios. If not, it plays cases. +_PIPELINES = { + False: lambda: RunnerPipeline(RunEachCaseFilter(RunCasePipeline())), + True: lambda: RunnerPipeline(BenchEachScenarioFilter(BenchScenarioPipeline())), +} -# cup: allow-plain the conformance harness: it builds each case, runs it on a target and reports; it never judges def main(): target = sys.argv[1] - if target.startswith("bench:"): - json.dump(bench(target.split(":", 1)[1], json.load(sys.stdin)), sys.stdout) - return - cases = json.load(sys.stdin) - json.dump([run_case(case, target, i) for i, case in enumerate(cases)], sys.stdout) + pipeline = _PIPELINES[target.startswith("bench:")]() + out = pipeline.run(Payload({"cases": json.load(sys.stdin), "target": target.removeprefix("bench:")})) + json.dump(out.get("results"), sys.stdout) if __name__ == "__main__": diff --git a/scorecard/vocab.py b/scorecard/vocab.py index a2790b5..606e47b 100644 --- a/scorecard/vocab.py +++ b/scorecard/vocab.py @@ -1,7 +1,7 @@ """The conformance vocabulary, reused (not reimplemented) from the Python -conformance runner: ``runner.build`` turns a case's steps into a live -Pipeline, ``runner._predicate_source`` writes Valve predicates as real -``def``s, and ``runner._load_export`` exports and imports the package from +conformance runner: ``BuildPipelineFilter`` turns a case's steps into a live +Pipeline, ``WritePredicatesFilter`` writes Valve predicates as real +classes, and ``ExportFilter`` exports and imports the package from disk, as a customer would. The only thing done differently here: predicate ``def``s get index-based @@ -27,11 +27,13 @@ import conformance_sink as sink # noqa: E402 import cup_ops as ops # noqa: E402 -import runner # noqa: E402 +from case_runner.build_pipeline_filter import BuildPipelineFilter # noqa: E402 +from case_runner.export_filter import ExportFilter # noqa: E402 +from case_runner.write_predicates_filter import WritePredicatesFilter # noqa: E402 from codeupipe import Payload, Pipeline # noqa: E402 from codeupipe.compile.errors import UnsupportedFilterShape # noqa: E402 -__all__ = ["sink", "ops", "runner", "Payload", "Pipeline", "UnsupportedFilterShape", +__all__ = ["sink", "ops", "Payload", "Pipeline", "UnsupportedFilterShape", "make_predicates", "build_pipeline", "load_export", "new_workdir", "walk_steps"] _COUNTER = itertools.count() @@ -55,7 +57,7 @@ def make_predicates(steps, workdir): """A namespace with ``when_`` for every Valve, each a real ``def`` in a real file (static lowering needs its source).""" valves = [s for s in walk_steps(steps) if s["kind"] == "valve"] - sources = [runner._predicate_source(str(i), s["when"]) for i, s in enumerate(valves)] + sources = [WritePredicatesFilter().source(str(i), s["when"]) for i, s in enumerate(valves)] name = f"scorecard_predicates_{next(_COUNTER)}" path = pathlib.Path(workdir) / f"{name}.py" path.write_text("\n\n".join(sources) or "\n") @@ -71,13 +73,13 @@ def make_predicates(steps, workdir): # cup: allow-plain the differential test harness: it makes cases, runs them in each mode and compares the results def build_pipeline(case, predicates, hooks=()): - """The case as a live codeupipe Pipeline (runner.build), with ``hooks``.""" - pipeline = runner.build(case, Pipeline, predicates, {}) + """The case as a live codeupipe Pipeline (BuildPipelineFilter), with ``hooks``.""" + pipeline = BuildPipelineFilter().build(case["steps"], Pipeline, predicates, {}) for hook in hooks: pipeline.use_hook(hook) return pipeline def load_export(pipeline, lowering, observe, workdir): - """runner._load_export: export, write to disk, import package ``steps``.""" - return runner._load_export(pipeline, lowering, observe, pathlib.Path(workdir)) + """ExportFilter: export, write to disk, import package ``steps``.""" + return ExportFilter().load(pipeline, lowering, observe, pathlib.Path(workdir)) diff --git a/tests/conformance/test_case_runner.py b/tests/conformance/test_case_runner.py new file mode 100644 index 0000000..414fac4 --- /dev/null +++ b/tests/conformance/test_case_runner.py @@ -0,0 +1,110 @@ +"""The Python runner is two pipelines, so each of its steps is tested alone.""" + +import pathlib +import sys + +import pytest + +RUNNER = pathlib.Path(__file__).resolve().parents[2] / "conformance" / "runners" / "python" +sys.path.insert(0, str(RUNNER)) + +from codeupipe import Payload, Pipeline # noqa: E402 + +import conformance_sink as sink # noqa: E402 +from case_runner.attach_record_hook_filter import AttachRecordHookFilter # noqa: E402 +from case_runner.build_pipeline_filter import BuildPipelineFilter # noqa: E402 +from case_runner.name_steps_filter import NameStepsFilter # noqa: E402 +from case_runner.parse_target_filter import ParseTargetFilter # noqa: E402 +from case_runner.run_case_pipeline import RunCasePipeline # noqa: E402 +from case_runner.run_each_case_filter import RunEachCaseFilter # noqa: E402 +from case_runner.unsupported import Unsupported # noqa: E402 +from case_runner.write_predicates_filter import WritePredicatesFilter # noqa: E402 + +VALVE = {"kind": "valve", "name": "gate", "when": {"op": "truthy", "key": "on"}, "filter": {"op": "set", "key": "b", "value": 2}} +CASE = {"id": "t/one", "steps": [{"kind": "filter", "name": "first", "op": "set", "key": "a", "value": 1}, VALVE, + {"kind": "tap", "name": "audit", "op": "record", "key": "a"}], "input": {"on": True}} + + +@pytest.mark.parametrize("target, compiled, run_async, lowering, observe", [ + ("run", False, False, "", False), ("run_async", False, True, "", False), + ("static-steps", True, False, "steps", False), ("static-inline", True, False, "inline", False), + ("static-observe", True, False, "steps", True), +]) +def test_what_the_name_of_a_target_means(target, compiled, run_async, lowering, observe): + out = ParseTargetFilter().call(Payload({"target": target})) + assert (out.get("compiled"), out.get("run_async"), out.get("lowering"), out.get("observe")) == (compiled, run_async, lowering, observe) + + +def test_a_name_that_is_not_a_target_stops_the_runner(): + with pytest.raises(SystemExit, match="unknown target 'nope'"): + ParseTargetFilter().call(Payload({"target": "nope"})) + with pytest.raises(SystemExit, match="unknown bench target 'run_async'"): + ParseTargetFilter(("run",), "bench target").call(Payload({"target": "run_async"})) + + +def built(tmp_path, case=CASE, run_async=False): + payload = Payload({"case": case, "workdir": tmp_path, "module_name": f"t_predicates_{abs(hash(tmp_path))}", "run_async": run_async}) + return BuildPipelineFilter().call(WritePredicatesFilter().call(payload)) + + +def test_a_predicate_of_a_case_is_a_real_class_with_test(tmp_path): + predicates = WritePredicatesFilter().call(Payload({"case": CASE, "workdir": tmp_path, "module_name": "t_predicates_a"})).get("predicates") + assert predicates.when_gate.test(Payload({"on": 1})) is True + assert predicates.when_gate.test(Payload({})) is False + assert type(predicates.when_gate).__name__ == "When_gate" + + +def test_the_pipeline_of_a_case_and_the_name_of_each_label(tmp_path): + out = built(tmp_path) + assert isinstance(out.get("pipeline"), Pipeline) + assert out.get("label") == {"SetFilter": "first", "SetFilter#2": "gate", "RecordTap": "audit"} + + +@pytest.mark.parametrize("step, message", [ + ({"kind": "filter", "name": "a", "op": "nope"}, "unknown filter op 'nope'"), + ({"kind": "tap", "name": "a", "op": "nope"}, "unknown tap op 'nope'"), + ({"kind": "nope", "name": "a"}, "unknown step kind 'nope'"), +]) +def test_a_word_that_the_runner_does_not_know_is_unsupported(tmp_path, step, message): + with pytest.raises(Unsupported, match=message): + built(tmp_path, {"id": "t/x", "steps": [step], "input": {}}) + + +def test_a_hook_is_attached_for_a_case_that_asks_and_a_target_that_calls_hooks(tmp_path): + def hooks(case_hooks, compiled, observe): + payload = built(tmp_path, {**CASE, "hooks": case_hooks}).insert("compiled", compiled).insert("observe", observe) + return AttachRecordHookFilter().call(payload).get("hooks") + + assert hooks(False, False, False) is None + assert len(hooks(True, False, False)) == 1 # the dynamic run + assert hooks(True, True, False) is None # fast compiled code calls no Hooks + assert len(hooks(True, True, True)) == 1 # compiled with observe + + +def test_each_step_in_a_result_gets_the_name_of_the_case(): + result = {"output": {}, "state": {"executed": ["SetFilter", "Other"], "skipped": ["SetFilter#2"], "errors": []}, + "inner_states": [{"executed": ["SetFilter#2"], "skipped": [], "errors": ["SetFilter"]}], + "hooks": ["before:", "before:SetFilter", "error:SetFilter#2"]} + named = NameStepsFilter().call(Payload({"result": result, "label": {"SetFilter": "first", "SetFilter#2": "gate"}})).get("result") + assert named["state"] == {"executed": ["first", "Other"], "skipped": ["gate"], "errors": []} + assert named["inner_states"] == [{"executed": ["gate"], "skipped": [], "errors": ["first"]}] + assert named["hooks"] == ["before:", "before:first", "error:gate"] + + +@pytest.mark.parametrize("target", ["run", "run_async", "static-steps", "static-inline", "static-observe"]) +def test_one_case_on_each_target(tmp_path, target): + sink.reset() + out = RunCasePipeline().run(Payload({"case": CASE, "target": target, "workdir": tmp_path, "module_name": f"t_case_{target.replace('-', '_')}"})) + result = out.get("result") + assert result["output"] == {"on": True, "a": 1, "b": 2} + assert result["taps"] == [1] + assert result["input_after"] == {"on": True} + assert ("state" in result) == (target not in ("static-steps", "static-inline")) + assert result.get("state", {"executed": ["first", "gate", "audit"]})["executed"] == ["first", "gate", "audit"] + + +def test_a_case_that_a_target_cannot_express_is_reported_and_the_next_case_still_runs(): + cases = [{"id": "t/bad", "steps": [{"kind": "filter", "name": "a", "op": "nope"}], "input": {}}, CASE] + results = RunEachCaseFilter(RunCasePipeline()).call(Payload({"cases": cases, "target": "run"})).get("results") + assert results[0] == {"id": "t/bad", "unsupported": "unknown filter op 'nope'"} + assert results[1]["id"] == "t/one" and results[1]["output"] == {"on": True, "a": 1, "b": 2} diff --git a/tests/lint/invariants/test_repo_is_clean.py b/tests/lint/invariants/test_repo_is_clean.py index 0fb44b3..cc7de64 100644 --- a/tests/lint/invariants/test_repo_is_clean.py +++ b/tests/lint/invariants/test_repo_is_clean.py @@ -1,13 +1,14 @@ """This repo's own Python keeps the invariants it lints for.""" import os +import pathlib from codeupipe_lint.invariants.lint_pipeline import lint_paths from ._support import REPO # The places with a `cup: allow-plain` comment. -PLAIN_BY_DESIGN = 75 +PLAIN_BY_DESIGN = 60 CHECKED = ["packages/codeupipe/codeupipe", "conformance", "scorecard", "packages/codeupipe-lint/codeupipe_lint", "tests", "examples"] @@ -21,5 +22,9 @@ def test_repo_python_has_no_invariant_violations(): # the comment gives: a primitive itself, a visitor of a syntax tree, the entry of a program, a test # harness. The number is here so that a new one is a change that someone sees. allowed = out.get("suppressed") - assert {f.rule for f in allowed} == {"CUP-I008"} - assert len(allowed) == PLAIN_BY_DESIGN, sorted({f.path for f in allowed}) + plain = [f for f in allowed if f.rule == "CUP-I008"] + assert len(plain) == PLAIN_BY_DESIGN, sorted({f.path for f in plain}) + # One more comment, for another rule: the conformance runner reads the State of the pipeline of a + # case, because that State is what it reports. + other = [(f.rule, pathlib.Path(f.path).name) for f in allowed if f.rule != "CUP-I008"] + assert other == [("CUP-I003", "run_dynamic_filter.py")]