From 735c897d36afb8aa3e3b55e386afe84e86dd72e9 Mon Sep 17 00:00:00 2001 From: Joshua Wink Date: Tue, 13 Oct 2026 12:05:38 -0500 Subject: [PATCH 1/2] 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 --- 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 +----------------- tests/conformance/test_case_runner.py | 110 +++++ tests/lint/invariants/test_repo_is_clean.py | 11 +- 24 files changed, 845 insertions(+), 369 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/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")] From 855eb329d18556a4ccd94e9b53508f8197ba08e9 Mon Sep 17 00:00:00 2001 From: Joshua Wink Date: Tue, 13 Oct 2026 12:07:05 -0500 Subject: [PATCH 2/2] 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 --- scorecard/vocab.py | 22 ++++++++++++---------- 1 file changed, 12 insertions(+), 10 deletions(-) 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))