From 5b381dfd958366a1e22418b13cc41ec020d940eb Mon Sep 17 00:00:00 2001 From: Joshua Wink Date: Mon, 12 Oct 2026 22:45:20 -0500 Subject: [PATCH 1/2] fix: a State belongs to one run, not to the Pipeline Two runs of one Pipeline that were in progress at the same time wrote into one State, so their steps mixed. Each run now writes into a State of its own, and the caller can give it: run(payload, state). Python: the State is a local of the run. The Pipeline asks the Predicate of a Valve itself and no longer reads a flag from the Valve, which each run in progress shared. TypeScript: run takes an optional State. The contract has the rule, and three conformance cases: the caller gives the State, and a run that starts inside another run. Closes #11 Co-Authored-By: Claude Opus 5.5 --- README.md | 2 +- conformance/API.md | 25 +- conformance/MATRIX.md | 27 ++- conformance/README.md | 9 + conformance/cases/state.json | 225 ++++++++++++++++++ conformance/run.py | 4 + .../runners/python/conformance_sink.py | 3 + .../runners/python/cup_ops/__init__.py | 4 + .../python/cup_ops/async_run_again_filter.py | 20 ++ .../python/cup_ops/run_again_filter.py | 24 ++ conformance/runners/python/runner.py | 40 +++- .../typescript/cup-ops/run-again-filter.ts | 28 +++ .../runners/typescript/cup-ops/sink.ts | 3 + conformance/runners/typescript/runner.ts | 28 ++- docs/compile.md | 4 +- packages/codeupipe-ts/README.md | 6 +- packages/codeupipe-ts/src/pipeline.ts | 11 +- packages/codeupipe-ts/tests/pipeline.test.ts | 45 +++- packages/codeupipe/codeupipe/core/README.md | 5 +- packages/codeupipe/codeupipe/core/pipeline.py | 135 ++++++----- packages/codeupipe/codeupipe/core/valve.py | 4 + tests/conformance/test_python_targets.py | 13 +- tests/conformance/test_typescript_target.py | 5 +- tests/core/test_state_per_run.py | 91 +++++++ 24 files changed, 652 insertions(+), 109 deletions(-) create mode 100644 conformance/cases/state.json create mode 100644 conformance/runners/python/cup_ops/async_run_again_filter.py create mode 100644 conformance/runners/python/cup_ops/run_again_filter.py create mode 100644 conformance/runners/typescript/cup-ops/run-again-filter.ts create mode 100644 tests/core/test_state_per_run.py diff --git a/README.md b/README.md index 146e1fb..9dbdfca 100644 --- a/README.md +++ b/README.md @@ -183,7 +183,7 @@ An editable install needs pip 21.3 or later. | Command | What it does | |---|---| | `python -m pytest tests -q` | Runs the tests | -| `python conformance/run.py` | Runs the 27 conformance cases against the Python dynamic run and the three compiled forms, and against the TypeScript targets that the machine can run | +| `python conformance/run.py` | Runs the 30 conformance cases against the Python dynamic run and the three compiled forms, and against the TypeScript targets that the machine can run | | `python -m codeupipe_lint.anonymous_steps` | Runs the anonymous-step linter. Paths come from `cup.config.json` | | `python -m codeupipe_lint.invariants` | Runs the invariant linter. Paths come from `cup.config.json` | | `python -m codeupipe.compile examples.pricing.pricing_pipeline:PricingPipeline --out build/pricing` | Compiles the example to `build/pricing/src/steps/` | diff --git a/conformance/API.md b/conformance/API.md index 064b5db..ca65db6 100644 --- a/conformance/API.md +++ b/conformance/API.md @@ -120,12 +120,23 @@ A Hook reads the Payload and does not write to it. - Order on a filter failure: `… before(step)`, `on_error(step)`, `on_error(pipeline)`. No `after` events follow. -## State — what happened in the last run +## State — what happened in one run `executed` (labels, in order), `skipped` (labels), `errors` -(label + error), `timings` (label → seconds, when timing is on). A run -starts with a new State. **A failed run keeps the steps that completed -before the failure.** +(label + error), `timings` (label → seconds, when timing is on). **A failed +run keeps the steps that completed before the failure.** + +**A State belongs to one run, not to the Pipeline.** + +- Each run writes into a State of its own. Two runs of one Pipeline that + are in progress at the same time never write into the same State. +- The caller can give the State: `run(p, state)`. The run writes into that + State. This is how a caller with more than one run at a time gets the + State of each run. The caller gives a new State for each run. +- With no State from the caller, the run makes a new one. +- `state` on the Pipeline is the State of the run that started last, with + a State from the caller or not. It is for a caller that has one run at a + time. ## Pipeline @@ -137,9 +148,9 @@ before the failure.** | add a tap | `add_tap(t)` | | nest a pipeline | `add_pipeline(p)` | | add a hook | `use_hook(h)` | -| run (sync) | `run(p)` | -| run (async, opt-in) | `run_async(p)` | -| last run | `state` | +| run (sync) | `run(p)`, `run(p, state)` | +| run (async, opt-in) | `run_async(p)`, `run_async(p, state)` | +| the State of the run that started last | `state` | | timing and lineage | `observe(timing=True, lineage=False)` | | each step's label, in order | `labels` | diff --git a/conformance/MATRIX.md b/conformance/MATRIX.md index b336629..9a2c172 100644 --- a/conformance/MATRIX.md +++ b/conformance/MATRIX.md @@ -4,9 +4,13 @@ A snapshot of `python conformance/run.py`. Regenerate with that command; the live copy is `results/latest.md` (not committed). The command exits non-zero if any target has a ❌. -**Every target passes every case, and nothing is unsupported.** The targets -are the Python dynamic run, async run and three compiled forms, and the -TypeScript run and its compiled form. +**Every target passes every case that it can express.** The targets are the +Python dynamic run, async run and three compiled forms, and the TypeScript +run and its compiled form. + +Two cases are ➖ for the compiled forms. In them a step starts a second run +of its own Pipeline, and a compiled form is one function with no Pipeline to +run again. The fast modes (`python:static-steps`, `python:static-inline`, `typescript:static`) expose no State or Hooks, so they pass those cases as @@ -34,6 +38,9 @@ The fast modes (`python:static-steps`, `python:static-inline`, | `pipeline/steps-run-in-order` | ✅ | ✅ | ✅* | ✅* | ✅ | ✅ | ✅* | | `pipeline/filter-error-halts-the-run` | ✅ | ✅ | ✅* | ✅* | ✅ | ✅ | ✅* | | `pipeline/nested-pipeline-is-a-step` | ✅ | ✅ | ✅* | ✅* | ✅ | ✅ | ✅* | +| `state/the-caller-gives-the-state-of-a-run` | ✅ | ✅ | ✅* | ✅* | ✅ | ✅ | ✅* | +| `state/a-run-that-starts-inside-another-run-has-its-own-state` | ✅ | ✅ | ➖ | ➖ | ➖ | ✅ | ➖ | +| `state/with-no-state-from-the-caller-the-pipeline-has-the-state-of-the-run-that-started-last` | ✅ | ✅ | ➖ | ➖ | ➖ | ✅ | ➖ | | `tap/observes-without-changing-output` | ✅ | ✅ | ✅* | ✅* | ✅ | ✅ | ✅* | | `tap/failure-never-halts-the-pipeline` | ✅ | ✅ | ✅* | ✅* | ✅ | ✅ | ✅* | | `tap/failure-does-not-stop-later-taps` | ✅ | ✅ | ✅* | ✅* | ✅ | ✅ | ✅* | @@ -44,12 +51,12 @@ The fast modes (`python:static-steps`, `python:static-inline`, | `valve/sees-upstream-changes` | ✅ | ✅ | ✅* | ✅* | ✅ | ✅ | ✅* | | `valve/independent-gates` | ✅ | ✅ | ✅* | ✅* | ✅ | ✅ | ✅* | -- **python:run**: 27 pass, 0 pass*, 0 fail, 0 unsupported -- **python:run_async**: 27 pass, 0 pass*, 0 fail, 0 unsupported -- **python:static-steps**: 11 pass, 16 pass*, 0 fail, 0 unsupported -- **python:static-inline**: 11 pass, 16 pass*, 0 fail, 0 unsupported -- **python:static-observe**: 27 pass, 0 pass*, 0 fail, 0 unsupported -- **typescript:run**: 27 pass, 0 pass*, 0 fail, 0 unsupported -- **typescript:static**: 11 pass, 16 pass*, 0 fail, 0 unsupported +- **python:run**: 30 pass, 0 pass*, 0 fail, 0 unsupported +- **python:run_async**: 30 pass, 0 pass*, 0 fail, 0 unsupported +- **python:static-steps**: 11 pass, 17 pass*, 0 fail, 2 unsupported +- **python:static-inline**: 11 pass, 17 pass*, 0 fail, 2 unsupported +- **python:static-observe**: 28 pass, 0 pass*, 0 fail, 2 unsupported +- **typescript:run**: 30 pass, 0 pass*, 0 fail, 0 unsupported +- **typescript:static**: 11 pass, 17 pass*, 0 fail, 2 unsupported ✅* passed everything the target can observe; it does not expose State or Hooks (the fast mode). ➖ the target cannot express the case. diff --git a/conformance/README.md b/conformance/README.md index a489f06..44ad888 100644 --- a/conformance/README.md +++ b/conformance/README.md @@ -71,6 +71,14 @@ load a number can leave the band. "taps": [...], "hooks": ["before:", "before:s1", ...]}} ``` +Three more keys are for the cases about State and runs that overlap: + +- `"given_state": true` in a case: the runner gives the run a new State + (`run(p, state)`), and reports that State as `state`. +- `expect.inner_states`: the State of each run that a `run_again` step + started, in order. +- `expect.pipeline_state`: `state` on the Pipeline after the run ends. + Every case also checks, implicitly, that **the caller's input was not mutated**. Numbers compare as JSON numbers (`1 == 1.0`), but `true` is never `1`. @@ -87,6 +95,7 @@ Numbers compare as JSON numbers (`1 == 1.0`), but `true` is never `1`. | filter | `merge` value | `merge(Payload(value))` | | filter | `has` key, to | insert `to = has(key)` | | filter | `mutate` key, value | `insert(key, value)` as a statement of its own, then return the payload | +| filter | `run_again` input | when `get("inner")` is not true: run the pipeline that this step is in one more time, with `input` as a new Payload and a new State that the step gives, and keep that State for the report. Then return the payload with no change. This is a second run that starts and ends while the first one is in progress | | tap | `record` key | append `get(key)` to the runner's tap log | | tap | `fail` message | raise | | valve `when` | `equals` key, value | `get(key) == value` | diff --git a/conformance/cases/state.json b/conformance/cases/state.json new file mode 100644 index 0000000..bdd5219 --- /dev/null +++ b/conformance/cases/state.json @@ -0,0 +1,225 @@ +[ + { + "id": "state/the-caller-gives-the-state-of-a-run", + "given_state": true, + "steps": [ + { + "kind": "filter", + "name": "first", + "op": "set", + "key": "a", + "value": 1 + }, + { + "kind": "valve", + "name": "gate", + "when": { + "op": "truthy", + "key": "inner" + }, + "filter": { + "op": "set", + "key": "gated", + "value": true + } + }, + { + "kind": "tap", + "name": "audit", + "op": "fail", + "message": "sink down" + }, + { + "kind": "filter", + "name": "last", + "op": "add", + "key": "n", + "by": 1 + } + ], + "input": {}, + "expect": { + "output": { + "a": 1, + "n": 1 + }, + "state": { + "executed": [ + "first", + "audit", + "last" + ], + "skipped": [ + "gate" + ], + "errors": [ + "audit" + ] + }, + "pipeline_state": { + "executed": [ + "first", + "audit", + "last" + ], + "skipped": [ + "gate" + ], + "errors": [ + "audit" + ] + }, + "inner_states": [] + } + }, + { + "id": "state/a-run-that-starts-inside-another-run-has-its-own-state", + "given_state": true, + "steps": [ + { + "kind": "filter", + "name": "first", + "op": "set", + "key": "a", + "value": 1 + }, + { + "kind": "filter", + "name": "again", + "op": "run_again", + "input": { + "inner": true + } + }, + { + "kind": "valve", + "name": "gate", + "when": { + "op": "truthy", + "key": "inner" + }, + "filter": { + "op": "set", + "key": "gated", + "value": true + } + }, + { + "kind": "filter", + "name": "last", + "op": "add", + "key": "n", + "by": 1 + } + ], + "input": {}, + "expect": { + "output": { + "a": 1, + "n": 1 + }, + "state": { + "executed": [ + "first", + "again", + "last" + ], + "skipped": [ + "gate" + ], + "errors": [] + }, + "inner_states": [ + { + "executed": [ + "first", + "again", + "gate", + "last" + ], + "skipped": [], + "errors": [] + } + ], + "pipeline_state": { + "executed": [ + "first", + "again", + "gate", + "last" + ], + "skipped": [], + "errors": [] + } + } + }, + { + "id": "state/with-no-state-from-the-caller-the-pipeline-has-the-state-of-the-run-that-started-last", + "steps": [ + { + "kind": "filter", + "name": "first", + "op": "set", + "key": "a", + "value": 1 + }, + { + "kind": "filter", + "name": "again", + "op": "run_again", + "input": { + "inner": true + } + }, + { + "kind": "valve", + "name": "gate", + "when": { + "op": "truthy", + "key": "inner" + }, + "filter": { + "op": "set", + "key": "gated", + "value": true + } + }, + { + "kind": "filter", + "name": "last", + "op": "add", + "key": "n", + "by": 1 + } + ], + "input": {}, + "expect": { + "output": { + "a": 1, + "n": 1 + }, + "inner_states": [ + { + "executed": [ + "first", + "again", + "gate", + "last" + ], + "skipped": [], + "errors": [] + } + ], + "pipeline_state": { + "executed": [ + "first", + "again", + "gate", + "last" + ], + "skipped": [], + "errors": [] + } + } + } +] diff --git a/conformance/run.py b/conformance/run.py index 0cb333a..d2fabac 100644 --- a/conformance/run.py +++ b/conformance/run.py @@ -164,6 +164,10 @@ def judge(case, result): 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}") diff --git a/conformance/runners/python/conformance_sink.py b/conformance/runners/python/conformance_sink.py index 89b2a56..fcc42ec 100644 --- a/conformance/runners/python/conformance_sink.py +++ b/conformance/runners/python/conformance_sink.py @@ -9,8 +9,11 @@ TAPS = [] HOOKS = [] +# The State of each run that a run_again step started, in order. +INNER_STATES = [] def reset(): del TAPS[:] del HOOKS[:] + del INNER_STATES[:] diff --git a/conformance/runners/python/cup_ops/__init__.py b/conformance/runners/python/cup_ops/__init__.py index b9a7b47..384beca 100644 --- a/conformance/runners/python/cup_ops/__init__.py +++ b/conformance/runners/python/cup_ops/__init__.py @@ -6,6 +6,7 @@ """ from .add_filter import AddFilter +from .async_run_again_filter import AsyncRunAgainFilter from .append_filter import AppendFilter from .copy_filter import CopyFilter from .fail_filter import FailFilter @@ -19,11 +20,13 @@ from .record_hook import RecordHook from .record_tap import RecordTap from .remove_filter import RemoveFilter +from .run_again_filter import RunAgainFilter from .set_filter import SetFilter __all__ = [ "AddFilter", "AppendFilter", + "AsyncRunAgainFilter", "CopyFilter", "FailFilter", "FailTap", @@ -36,5 +39,6 @@ "RecordHook", "RecordTap", "RemoveFilter", + "RunAgainFilter", "SetFilter", ] diff --git a/conformance/runners/python/cup_ops/async_run_again_filter.py b/conformance/runners/python/cup_ops/async_run_again_filter.py new file mode 100644 index 0000000..4620e17 --- /dev/null +++ b/conformance/runners/python/cup_ops/async_run_again_filter.py @@ -0,0 +1,20 @@ +"""AsyncRunAgainFilter: the reference `run_again` step for ``run_async`` (conformance/README.md).""" + +from codeupipe import Payload, State + +from conformance_sink import INNER_STATES + + +class AsyncRunAgainFilter: + """``RunAgainFilter`` for a pipeline that runs with ``run_async``.""" + + def __init__(self, data): + self.data = data + self.pipeline = None + + async def call(self, payload): + if payload.get("inner") is not True: + state = State() + INNER_STATES.append(state) + await self.pipeline.run_async(Payload(dict(self.data)), state) + return payload diff --git a/conformance/runners/python/cup_ops/run_again_filter.py b/conformance/runners/python/cup_ops/run_again_filter.py new file mode 100644 index 0000000..d655f3c --- /dev/null +++ b/conformance/runners/python/cup_ops/run_again_filter.py @@ -0,0 +1,24 @@ +"""RunAgainFilter: the reference `run_again` step (conformance/README.md).""" + +from codeupipe import Payload, State + +from conformance_sink import INNER_STATES + + +class RunAgainFilter: + """Starts a second run of the pipeline that it is in, inside the first one. + + The runner sets ``pipeline`` after it builds the pipeline. The second run + has ``inner`` in its input, so it does not start a third one. + """ + + def __init__(self, data): + self.data = data + self.pipeline = None + + def call(self, payload): + if payload.get("inner") is not True: + state = State() + INNER_STATES.append(state) + self.pipeline.run(Payload(dict(self.data)), state) + return payload diff --git a/conformance/runners/python/runner.py b/conformance/runners/python/runner.py index 37b5859..279e316 100644 --- a/conformance/runners/python/runner.py +++ b/conformance/runners/python/runner.py @@ -28,7 +28,7 @@ import conformance_sink as sink # noqa: E402 import cup_ops as ops # noqa: E402 -from codeupipe import Payload, Pipeline # noqa: E402 +from codeupipe import Payload, Pipeline, State # noqa: E402 # noqa: E402 @@ -36,8 +36,10 @@ class Unsupported(Exception): pass -def _filter(spec): +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": @@ -105,12 +107,14 @@ def walk(items): sys.path.remove(str(module_dir)) -def build(case, cls, predicates, names): +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) + obj = _filter(step, run_async) + if step["op"] == "run_again": + obj.pipeline = p p.add_filter(obj) elif kind == "tap": obj = _tap(step) @@ -155,26 +159,33 @@ def _named_hooks(events, label): def _dynamic(case, predicates, run_async=False): names = {} - p = build(case, Pipeline, predicates, 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 = {} try: if run_async: - out = asyncio.run(p.run_async(Payload(data))) + out = asyncio.run(p.run_async(Payload(data), *given)) else: - out = p.run(Payload(data)) + 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}" - state = p.state - result["state"] = { - "executed": _named(state.executed, label), - "skipped": _named(state.skipped, label), - "errors": _named([step for step, _ in state.errors], label), - } + + 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 @@ -202,6 +213,9 @@ def _load_export(p, lowering, observe, workdir): def _static(case, predicates, lowering, workdir, observe=False): + # The compiled form always gets its State from the caller, so "given_state" needs nothing more. + if any(step.get("op") == "run_again" for step in case["steps"]): + raise Unsupported("the compiled form is one function: it has no Pipeline that a step can run again") names = {} p = build(case, Pipeline, predicates, names) label = labels(p, names) diff --git a/conformance/runners/typescript/cup-ops/run-again-filter.ts b/conformance/runners/typescript/cup-ops/run-again-filter.ts new file mode 100644 index 0000000..554144d --- /dev/null +++ b/conformance/runners/typescript/cup-ops/run-again-filter.ts @@ -0,0 +1,28 @@ +import { Payload, State } from "../../../../packages/codeupipe-ts/src/index.ts"; +import type { Filter, Pipeline } from "../../../../packages/codeupipe-ts/src/index.ts"; + +import { INNER_STATES } from "./sink.ts"; + +/** + * The reference `run_again` step (conformance/README.md): it starts a second + * run of the pipeline that it is in, inside the first one. The runner sets + * `pipeline` after it builds the pipeline. The second run has `inner` in its + * input, so it does not start a third one. + */ +export class RunAgainFilter implements Filter { + readonly data: Record; + pipeline: Pipeline | null = null; + + constructor(data: Record) { + this.data = data; + } + + async call(payload: Payload): Promise { + if (payload.get("inner") !== true && this.pipeline !== null) { + const state = new State(); + INNER_STATES.push(state); + await this.pipeline.run(new Payload({ ...this.data }), state); + } + return payload; + } +} diff --git a/conformance/runners/typescript/cup-ops/sink.ts b/conformance/runners/typescript/cup-ops/sink.ts index 05246a9..c4dce20 100644 --- a/conformance/runners/typescript/cup-ops/sink.ts +++ b/conformance/runners/typescript/cup-ops/sink.ts @@ -1,2 +1,5 @@ /** Where the Taps of the runner record what they see. The runner empties it before each case. */ export const TAPS: unknown[] = []; + +/** The State of each run that a `run_again` step started, in order. */ +export const INNER_STATES: Array<{ executed: string[]; skipped: string[]; errors: Array<[string, unknown]> }> = []; diff --git a/conformance/runners/typescript/runner.ts b/conformance/runners/typescript/runner.ts index 5e7c7eb..0640840 100644 --- a/conformance/runners/typescript/runner.ts +++ b/conformance/runners/typescript/runner.ts @@ -19,7 +19,7 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import { pathToFileURL } from "node:url"; -import { Payload, Pipeline } from "../../../packages/codeupipe-ts/src/index.ts"; +import { Payload, Pipeline, State } from "../../../packages/codeupipe-ts/src/index.ts"; import type { Filter, Predicate, Tap } from "../../../packages/codeupipe-ts/src/index.ts"; import { AddFilter } from "./cup-ops/add-filter.ts"; import { AppendFilter } from "./cup-ops/append-filter.ts"; @@ -36,11 +36,12 @@ import { RecordHook } from "./cup-ops/record-hook.ts"; import { RecordTap } from "./cup-ops/record-tap.ts"; import { RemoveFilter } from "./cup-ops/remove-filter.ts"; import { SetFilter } from "./cup-ops/set-filter.ts"; -import { TAPS } from "./cup-ops/sink.ts"; +import { RunAgainFilter } from "./cup-ops/run-again-filter.ts"; +import { INNER_STATES, TAPS } from "./cup-ops/sink.ts"; import { TruthyPredicate } from "./cup-ops/truthy-predicate.ts"; type Spec = Record; -type Case = { id: string; hooks?: boolean; steps: Spec[]; input: Record }; +type Case = { id: string; hooks?: boolean; given_state?: boolean; steps: Spec[]; input: Record }; class Unsupported extends Error {} @@ -55,6 +56,7 @@ function makeFilter(spec: Spec): Filter { case "has": return new HasFilter(spec.key, spec.to); case "mutate": return new MutateFilter(spec.key, spec.value); case "fail": return new FailFilter(spec.message); + case "run_again": return new RunAgainFilter(spec.input); default: throw new Unsupported(`unknown filter op '${spec.op}'`); } } @@ -79,7 +81,9 @@ function makePredicate(when: Spec): Predicate { function build(steps: Spec[], pipeline: Pipeline): Pipeline { for (const step of steps) { if (step.kind === "filter") { - pipeline.addFilter(makeFilter(step)); + const filter = makeFilter(step); + if (filter instanceof RunAgainFilter) filter.pipeline = pipeline; + pipeline.addFilter(filter); } else if (step.kind === "tap") { pipeline.addTap(makeTap(step)); } else if (step.kind === "valve") { @@ -113,17 +117,21 @@ async function runCase(spec: Case): Promise> { const data = structuredClone(spec.input); const result: Record = { id: spec.id }; + // With "given_state", the caller gives the State of the run. If not, the run makes one. + const given = spec.given_state ? new State() : undefined; try { - result.output = (await pipeline.run(new Payload(data))).toDict(); + result.output = (await (given === undefined ? pipeline.run(new Payload(data)) : pipeline.run(new Payload(data), given))).toDict(); } catch (error) { result.error = error instanceof Error ? `${error.name}: ${error.message}` : String(error); } - const state = pipeline.state; - result.state = { + const report = (state: { executed: string[]; skipped: string[]; errors: Array<[string, unknown]> }) => ({ executed: state.executed.map(named), skipped: state.skipped.map(named), errors: state.errors.map(([label]) => named(label)), - }; + }); + result.state = report(given ?? pipeline.state); + result.pipeline_state = report(pipeline.state); + result.inner_states = INNER_STATES.map(report); result.taps = [...TAPS]; if (spec.hooks) { result.hooks = events.map((event) => { @@ -190,6 +198,9 @@ function sourceOf(spec: Case): string { type Compile = (file: string, className: string, options: { outFile: string }) => Promise; async function runStatic(spec: Case, compile: Compile, directory: string, index: number): Promise> { + if (spec.steps.some((step) => step.op === "run_again")) { + throw new Unsupported("the compiled form is one function: it has no Pipeline that a step can run again"); + } const sourceFile = join(directory, `case-${index}-pipeline.ts`); const outFile = join(directory, `case-${index}-compiled.ts`); writeFileSync(sourceFile, sourceOf(spec)); @@ -231,6 +242,7 @@ const cases: Case[] = JSON.parse(readFileSync(0, "utf8")); const results: Record[] = []; for (const [index, spec] of cases.entries()) { TAPS.length = 0; + INNER_STATES.length = 0; try { results.push(compile === undefined ? await runCase(spec) : await runStatic(spec, compile, directory, index)); } catch (error) { diff --git a/docs/compile.md b/docs/compile.md index e2b6dbf..6550ed2 100644 --- a/docs/compile.md +++ b/docs/compile.md @@ -74,7 +74,7 @@ generated `run()` (scorecard findings F3–F8). | Type hints on a step (`Payload[Reads] -> Payload[Writes]`, hinted locals) | erased: compiled code carries no hints and never imports their names | `test_export_typed_steps` | | Payload ops `get insert remove merge has to_dict copy` | dict operations; a write is an in-place statement (`data[k] = v`) | `test_rewrite_payload_ops` | -Evidence: all 27 conformance cases on both layouts (`python:static-steps`, +Evidence: each conformance case that a compiled form can express (28 of 30) on both layouts (`python:static-steps`, `python:static-inline`), and the parity spike at 2400/2400 inputs. ## What is refused @@ -116,7 +116,7 @@ state["executed"], state["skipped"], state["errors"], state["timings"], state["t ``` It reproduces the dynamic run's observable behaviour exactly, and the -`python:static-observe` conformance target checks it on all 27 cases: +`python:static-observe` conformance target checks it on the 28 cases that a compiled form can express: - **State** is written into the `state` dict you pass (reset each run): `executed`, `skipped`, `errors` as `(name, error)`, `timings` (only with diff --git a/packages/codeupipe-ts/README.md b/packages/codeupipe-ts/README.md index 90ade7a..e2a47a3 100644 --- a/packages/codeupipe-ts/README.md +++ b/packages/codeupipe-ts/README.md @@ -5,7 +5,7 @@ This is the TypeScript runtime. It has no dependencies. The built JavaScript runs in a browser and in Node. It implements the same contract as the Python -package, and the same 27 conformance cases judge it: +package, and the same 30 conformance cases judge it: [conformance/API.md](../../conformance/API.md). The version is below 1.0: the package is not stable yet, and a new version @@ -140,7 +140,9 @@ override the methods that you need. - **A function is not a step.** Every JavaScript function has a `call` method, so `addFilter` refuses a function by its type. - **Two runs of one pipeline at the same time** each write to their own - State. `pipeline.state` is the State of the run that started last. + State. A caller with more than one run at a time gives each run its + State: `pipeline.run(payload, state)`. `pipeline.state` is the State of + the run that started last. - **The linter and the compiler are separate packages**, because they need a parser: [codeupipe-lint-ts](../codeupipe-lint-ts/) and [codeupipe-compile-ts](../codeupipe-compile-ts/). In Python the compiler is diff --git a/packages/codeupipe-ts/src/pipeline.ts b/packages/codeupipe-ts/src/pipeline.ts index 1436487..c6833c4 100644 --- a/packages/codeupipe-ts/src/pipeline.ts +++ b/packages/codeupipe-ts/src/pipeline.ts @@ -124,11 +124,14 @@ export class Pipeline { * A step can be sync or async. A Filter failure stops the run and is * thrown again after the Hooks have seen it. A Tap failure is recorded in * State and the run continues. + * + * The run writes into `state` when the caller gives one, and into a new + * State when not. A State belongs to one run: a caller with more than one + * run of this Pipeline at a time gives each run its own. */ - async run(initial: Payload): Promise { - // Each run writes to its own State, so two runs at the same time do not mix. - const state = new State(); - this.#state = state; + async run(initial: Payload, state: State = new State()): Promise { + // The State is a local of this call, so two runs at the same time do not mix. + this.#state = state; // the State of the run that started last const hooks = this.#hooks; const timed = this.#timing; let payload = initial.copy(); // the caller's Payload is not changed diff --git a/packages/codeupipe-ts/tests/pipeline.test.ts b/packages/codeupipe-ts/tests/pipeline.test.ts index 9cef7d3..68881dd 100644 --- a/packages/codeupipe-ts/tests/pipeline.test.ts +++ b/packages/codeupipe-ts/tests/pipeline.test.ts @@ -1,7 +1,7 @@ import assert from "node:assert/strict"; import { test } from "node:test"; -import { Hook, Payload, Pipeline } from "../src/index.ts"; +import { Hook, Payload, Pipeline, State } from "../src/index.ts"; import type { Filter, Predicate, Tap } from "../src/index.ts"; class AddFilter implements Filter { @@ -252,3 +252,46 @@ test("two runs at the same time do not write into the same State", async () => { assert.equal(b.get("n"), 2); assert.deepEqual(pipeline.state.executed, ["AddLaterFilter", "AddLaterFilter#2"]); }); + +// -- a State belongs to one run ------------------------------------------- + +class WaitFilter implements Filter { + async call(payload: Payload): Promise { + await new Promise((resolve) => setTimeout(resolve, payload.get("wait") ?? 0)); + return payload.insert("done", payload.get("name")); + } +} + +class FailForBFilter implements Filter { + call(payload: Payload): Payload { + if (payload.get("name") === "B") throw new Error("B fails"); + return payload; + } +} + +test("each of two runs at the same time writes into the State that its caller gave", async () => { + const pipeline = new Pipeline(); + pipeline.addFilter(new WaitFilter()); + pipeline.addFilter(new FailForBFilter()); + const [a, b] = [new State(), new State()]; + // A starts first and ends last. B starts second, fails, and ends first. + await Promise.all([ + pipeline.run(new Payload({ name: "A", wait: 40 }), a), + pipeline.run(new Payload({ name: "B", wait: 5 }), b).catch(() => null), + ]); + assert.deepEqual(a.executed, ["WaitFilter", "FailForBFilter"]); + assert.deepEqual(b.executed, ["WaitFilter"]); + assert.equal(pipeline.state, b); // the State of the run that started last +}); + +test("with no State from the caller, a run makes a new one, and a later run does not write into a State that was given", async () => { + const pipeline = new Pipeline(); + pipeline.addFilter(new FailForBFilter()); + const given = new State(); + await pipeline.run(new Payload({ name: "A" }), given); + assert.equal(pipeline.state, given); + await pipeline.run(new Payload({ name: "A" })); + assert.notEqual(pipeline.state, given); + assert.deepEqual(given.executed, ["FailForBFilter"]); + assert.deepEqual(pipeline.state.executed, ["FailForBFilter"]); +}); diff --git a/packages/codeupipe/codeupipe/core/README.md b/packages/codeupipe/codeupipe/core/README.md index aa22183..07a40be 100644 --- a/packages/codeupipe/codeupipe/core/README.md +++ b/packages/codeupipe/codeupipe/core/README.md @@ -34,8 +34,9 @@ library only. - `tap.py` — `Tap` / `AsyncTap`: read-only observation `Protocol`s (`observe(payload) -> None`); must not modify the payload. - `state.py` — `State`: pipeline execution metadata — executed/skipped - filters, timing, errors, arbitrary metadata. Available as `pipeline.state` - after a run. + filters, timing, errors, arbitrary metadata. A State belongs to one run. + `pipeline.state` is the State of the run that started last. A caller with + more than one run at a time gives each run its own: `run(payload, state)`. - `hook.py` — `Hook`: lifecycle ABC with `before`/`after`/`on_error` callbacks around filter and pipeline execution. - `async_hook.py` — `AsyncHook`: a Hook whose methods are coroutines. diff --git a/packages/codeupipe/codeupipe/core/pipeline.py b/packages/codeupipe/codeupipe/core/pipeline.py index 767ee24..b1a1824 100644 --- a/packages/codeupipe/codeupipe/core/pipeline.py +++ b/packages/codeupipe/codeupipe/core/pipeline.py @@ -47,7 +47,11 @@ def __init__(self): @property def state(self) -> State: - """Access pipeline execution state after run().""" + """The State of the run that started last. + + A State belongs to one run. A caller with more than one run of this + Pipeline at a time gives each run its own: ``run(payload, state)``. + """ return self._state @property @@ -91,7 +95,7 @@ def add_valve(self, predicate: Any, filter: Union[Filter[TInput, TOutput], Async self._require(predicate, "test", "A Predicate") self._require(filter, "call", "A Filter") valve = Valve(filter, predicate) - self._steps.append((self._label(valve), valve, "async_filter")) + self._steps.append((self._label(valve), valve, "valve")) def add_tap(self, tap: Union[Tap, AsyncTap]) -> None: """Add a Tap. A tap whose ``observe`` is a coroutine needs ``run_async()``.""" @@ -132,34 +136,38 @@ def _can_run_fast(self) -> bool: """Whether the common case can skip hooks and instrumentation.""" return not self._hooks and not self._observe_timing and not self._observe_lineage - async def _run_fast(self, initial_payload: Payload[TInput]) -> Payload[TOutput]: + async def _run_fast(self, initial_payload: Payload[TInput], state: State) -> Payload[TOutput]: """Hot path for simple pipelines: no hooks, no timing, no lineage.""" - self._state = State() payload = initial_payload.copy() # the caller's Payload is not changed for name, step, step_type in self._steps: if step_type == "tap": - self._observe(name, step, payload) - self._state.mark_executed(name) + self._observe(name, step, payload, state) + state.mark_executed(name) continue if step_type == "async_tap": - await self._observe_async(name, step, payload) - self._state.mark_executed(name) + await self._observe_async(name, step, payload, state) + state.mark_executed(name) continue if step_type == "pipeline": payload = await step.run_async(payload) - self._state.mark_executed(name) + state.mark_executed(name) continue - if step_type == "async_filter": + if step_type == "valve": + # The Pipeline asks the Predicate itself. A flag on the Valve + # would be shared by each run that is in progress. + if not step._predicate(payload): + state.mark_skipped(name) + continue + result = step._inner.call(payload) + payload = await result if inspect.isawaitable(result) else result + elif step_type == "async_filter": payload = await step.call(payload) # type: ignore[attr-defined] else: payload = step.call(payload) # type: ignore[attr-defined] - if hasattr(step, "_last_skipped") and step._last_skipped: - self._state.mark_skipped(name) - else: - self._state.mark_executed(name) + state.mark_executed(name) return payload # type: ignore[return-value] @@ -171,12 +179,19 @@ async def _invoke(fn, *args): result = await result return result - async def run_async(self, initial_payload: Payload[TInput]) -> Payload[TOutput]: - """Execute with async steps allowed (opt-in). ``run()`` is the default.""" + async def run_async(self, initial_payload: Payload[TInput], state: Optional[State] = None) -> Payload[TOutput]: + """Execute with async steps allowed (opt-in). ``run()`` is the default. + + The run writes into ``state`` when the caller gives one, and into a + new State when not. It is a local of this call, so two runs that are + in progress at the same time never write into the same State. + """ + if state is None: + state = State() + self._state = state # the State of the run that started last if self._can_run_fast(): - return await self._run_fast(initial_payload) + return await self._run_fast(initial_payload, state) - self._state = State() payload = initial_payload.copy() # the caller's Payload is not changed _step_name: Optional[str] = None _step_t0: Optional[float] = None @@ -192,13 +207,13 @@ async def run_async(self, initial_payload: Payload[TInput]) -> Payload[TOutput]: _failing = None if step_type == "tap": - self._observe(name, step, payload) - self._state.mark_executed(name) + self._observe(name, step, payload, state) + state.mark_executed(name) continue if step_type == "async_tap": - await self._observe_async(name, step, payload) - self._state.mark_executed(name) + await self._observe_async(name, step, payload, state) + state.mark_executed(name) continue _step_t0 = time.monotonic() @@ -209,30 +224,32 @@ async def run_async(self, initial_payload: Payload[TInput]) -> Payload[TOutput]: if step_type == "pipeline": payload = await step.run_async(payload) - self._state.mark_executed(name) + state.mark_executed(name) + elif step_type == "valve": + if step._predicate(payload): + result = step._inner.call(payload) + payload = await result if inspect.isawaitable(result) else result + state.mark_executed(name) + else: + state.mark_skipped(name) else: - # filter or valve if step_type == "async_filter": payload = await step.call(payload) # type: ignore[attr-defined] else: payload = step.call(payload) # type: ignore[attr-defined] - - if hasattr(step, '_last_skipped') and step._last_skipped: - self._state.mark_skipped(name) - else: - self._state.mark_executed(name) + state.mark_executed(name) for hook in self._hooks: await self._invoke(hook.after, name, payload) if self._observe_timing: - self._state.record_timing(name, time.monotonic() - _step_t0) + state.record_timing(name, time.monotonic() - _step_t0) if self._observe_lineage: payload = payload._stamp(name) except Exception as e: if _step_t0 is not None and _step_name is not None and self._observe_timing: - self._state.record_timing(_step_name, time.monotonic() - _step_t0) + state.record_timing(_step_name, time.monotonic() - _step_t0) if _failing is not None: # The failing step first, then the pipeline: on_error mirrors # before/after, which fire at both levels. @@ -247,7 +264,7 @@ async def run_async(self, initial_payload: Payload[TInput]) -> Payload[TOutput]: return payload # type: ignore - def _observe(self, name: str, tap: Any, payload: Payload) -> None: + def _observe(self, name: str, tap: Any, payload: Payload, state: State) -> None: """Run a sync Tap on Plane B: its failure never halts Plane A. The ``Exception`` is recorded in ``state.errors`` under the Tap's @@ -258,24 +275,27 @@ def _observe(self, name: str, tap: Any, payload: Payload) -> None: try: tap.observe(payload) except Exception as error: - self._state.record_error(name, error) + state.record_error(name, error) - async def _observe_async(self, name: str, tap: Any, payload: Payload) -> None: + async def _observe_async(self, name: str, tap: Any, payload: Payload, state: State) -> None: """``_observe`` for a Tap whose ``observe`` may be a coroutine.""" try: result = tap.observe(payload) if inspect.isawaitable(result): await result except Exception as error: - self._state.record_error(name, error) + state.record_error(name, error) # ------------------------------------------------------------------ # Sync execution: the default. Never starts an event loop. # ------------------------------------------------------------------ - def run(self, initial_payload: Payload[TInput]) -> Payload[TOutput]: + def run(self, initial_payload: Payload[TInput], state: Optional[State] = None) -> Payload[TOutput]: """Execute synchronously (the default). + The run writes into ``state`` when the caller gives one, and into a + new State when not (see ``run_async``). + Never starts an event loop: if the pipeline contains async work (an async filter, tap or hook) this raises ``AsyncStepInSyncRun``; use ``run_async()`` for those pipelines. Entering the async path costs @@ -287,7 +307,7 @@ def run(self, initial_payload: Payload[TInput]) -> Payload[TOutput]: f"cannot run async step '{offender}' inside run(). Use " f"run_async() from async code, or replace it with a sync step." ) - return self._run_sync(initial_payload) + return self._run_sync(initial_payload, state) def _preflight(self) -> Optional[str]: """The first async offender in this pipeline, or None. Cached. @@ -318,7 +338,7 @@ def _first_async_offender(self, nested: bool = True) -> Optional[str]: for name, step, step_type in self._steps: if step_type == "async_tap": return name - if step_type == "async_filter" and not self._has_sync_path(step): + if step_type == "async_filter" or (step_type == "valve" and not self._has_sync_path(step)): return name if step_type == "pipeline" and nested: inner = step._first_async_offender() @@ -329,9 +349,8 @@ def _first_async_offender(self, nested: bool = True) -> Optional[str]: def _has_sync_path(self, step: Any) -> bool: """True for an async-declared step that can run with no event loop. - ``Valve.call`` is async because its inner filter *may* be, so every - Valve registers as ``async_filter``. A Valve over a sync inner offers - ``call_sync``, which never touches a loop. A Valve over an async (or + ``Valve.call`` is async because its inner filter *may* be. A Valve + over a sync inner can run with no loop. A Valve over an async (or nested Valve) inner is async work. """ if not callable(getattr(step, "call_sync", None)): @@ -339,8 +358,10 @@ def _has_sync_path(self, step: Any) -> bool: inner = getattr(step, "_inner", None) return inner is None or not self._is_async_method(inner, "call") - def _run_sync(self, initial_payload: Payload[TInput]) -> Payload[TOutput]: - self._state = State() + def _run_sync(self, initial_payload: Payload[TInput], state: Optional[State]) -> Payload[TOutput]: + if state is None: + state = State() + self._state = state # the State of the run that started last payload = initial_payload.copy() # the caller's Payload is not changed # Clock reads cost ~0.1 µs each; only take them when someone reads them. @@ -361,8 +382,8 @@ def _run_sync(self, initial_payload: Payload[TInput]) -> Payload[TOutput]: _failing = None if step_type == "tap": - self._observe(name, step, payload) - self._state.mark_executed(name) + self._observe(name, step, payload, state) + state.mark_executed(name) continue if timed: @@ -373,27 +394,29 @@ def _run_sync(self, initial_payload: Payload[TInput]) -> Payload[TOutput]: hook.before(name, payload) if step_type == "pipeline": payload = step.run(payload) - self._state.mark_executed(name) - else: - if step_type == "async_filter": - payload = step.call_sync(payload) # admitted by _has_sync_path + state.mark_executed(name) + elif step_type == "valve": + # The Pipeline asks the Predicate itself: a flag on the + # Valve would be shared by each run that is in progress. + if step._predicate(payload): + payload = step._sync_inner(payload) # admitted by _has_sync_path + state.mark_executed(name) else: - payload = step.call(payload) # type: ignore[attr-defined] - if getattr(step, "_last_skipped", False): - self._state.mark_skipped(name) - else: - self._state.mark_executed(name) + state.mark_skipped(name) + else: + payload = step.call(payload) # type: ignore[attr-defined] + state.mark_executed(name) for hook in hooks: hook.after(name, payload) if timed: - self._state.record_timing(name, time.monotonic() - _step_t0) + state.record_timing(name, time.monotonic() - _step_t0) if self._observe_lineage: payload = payload._stamp(name) except Exception as e: if timed and _step_t0 is not None and _step_name is not None: - self._state.record_timing(_step_name, time.monotonic() - _step_t0) + state.record_timing(_step_name, time.monotonic() - _step_t0) if _failing is not None: # The failing step first, then the pipeline: on_error mirrors # before/after, which fire at both levels. diff --git a/packages/codeupipe/codeupipe/core/valve.py b/packages/codeupipe/codeupipe/core/valve.py index c3f4c5a..5817642 100644 --- a/packages/codeupipe/codeupipe/core/valve.py +++ b/packages/codeupipe/codeupipe/core/valve.py @@ -69,6 +69,10 @@ def __init__( self._inner_sync_call = inner_sync if callable(inner_sync) else None self._inner_is_sync = self._inner_sync_call is not None or not inspect.iscoroutinefunction( getattr(inner, "call", None)) + # What a synchronous run calls when the Predicate holds. The Pipeline + # asks the Predicate and calls this itself, and reads no flag from + # the Valve: a flag would be shared by each run that is in progress. + self._sync_inner = self._inner_sync_call if self._inner_sync_call is not None else inner.call async def call(self, payload: Payload[TInput]) -> Payload[TOutput]: """Execute the inner filter only if the predicate passes.""" diff --git a/tests/conformance/test_python_targets.py b/tests/conformance/test_python_targets.py index ba82b20..5983e07 100644 --- a/tests/conformance/test_python_targets.py +++ b/tests/conformance/test_python_targets.py @@ -16,6 +16,10 @@ PYTHON_TARGETS = ["python:run", "python:run_async", "python:static-steps", "python:static-inline", "python:static-observe"] +# The cases with a step that starts a second run of its own Pipeline. +RUNS_AGAIN = ["state/a-run-that-starts-inside-another-run-has-its-own-state", + "state/with-no-state-from-the-caller-the-pipeline-has-the-state-of-the-run-that-started-last"] + @pytest.fixture(scope="module") def report(tmp_path_factory): @@ -32,6 +36,11 @@ def test_target_passes_every_case_it_can_express(report, target): def test_observable_export_is_judged_on_state_and_hooks_too(report): - """No partial passes: static-observe exposes everything the dynamic run does.""" + """No partial passes: static-observe exposes everything the dynamic run does. + + The compiled form is one function, so it cannot express a step that runs + its Pipeline again. Those cases are the only ones that it leaves out. + """ summary = report["targets"]["python:static-observe"] - assert summary["pass"] == report["cases"], summary + assert sorted(summary["unsupported_cases"]) == RUNS_AGAIN, summary + assert summary["pass"] == report["cases"] - len(RUNS_AGAIN), summary diff --git a/tests/conformance/test_typescript_target.py b/tests/conformance/test_typescript_target.py index 47c1850..0ed4991 100644 --- a/tests/conformance/test_typescript_target.py +++ b/tests/conformance/test_typescript_target.py @@ -37,7 +37,10 @@ def test_the_compiled_pipeline_passes_every_case_on_what_it_can_show(): "typescript:static", conformance_run.TARGETS["typescript:static"], cases) assert error is None, error verdicts = {case["id"]: conformance_run.judge(case, results.get(case["id"])) for case in cases} - wrong = {case_id: verdict for case_id, verdict in verdicts.items() if verdict[0] not in ("pass", "partial")} + # The compiled form is one function: it cannot express a step that runs its Pipeline again. + runs_again = sorted(case["id"] for case in cases if any(step.get("op") == "run_again" for step in case["steps"])) + assert sorted(case_id for case_id, verdict in verdicts.items() if verdict[0] == "unsupported") == runs_again + wrong = {case_id: verdict for case_id, verdict in verdicts.items() if verdict[0] not in ("pass", "partial", "unsupported")} assert wrong == {}, json.dumps(wrong, indent=2) assert sum(1 for verdict in verdicts.values() if verdict[0] == "pass") == 11 diff --git a/tests/core/test_state_per_run.py b/tests/core/test_state_per_run.py new file mode 100644 index 0000000..dcf9b5f --- /dev/null +++ b/tests/core/test_state_per_run.py @@ -0,0 +1,91 @@ +"""A State belongs to one run (conformance/API.md, "State"). + +Two runs of one Pipeline that are in progress at the same time each have +their own State. The conformance cases check a run that starts inside +another run. These tests check runs that really overlap, with asyncio. +""" + +import asyncio + +from codeupipe import Payload, Pipeline, State + + +class SlowFilter: + async def call(self, payload): + await asyncio.sleep(payload.get("wait")) + return payload.insert("done", payload.get("name")) + + +class FailForBFilter: + def call(self, payload): + if payload.get("name") == "B": + raise ValueError("B fails") + return payload + + +class IsAPredicate: + def test(self, payload): + return payload.get("name") == "A" + + +class FailTap: + def observe(self, payload): + if payload.get("name") == "B": + raise RuntimeError("the sink is down") + + +def _overlap(pipeline, give_state): + """Run A (starts first, ends last) and B (starts second, ends first) at the same time.""" + async def one(name, wait): + state = State() if give_state else None + try: + await pipeline.run_async(Payload({"name": name, "wait": wait}), state) + except ValueError: + pass + return state + + async def both(): + return await asyncio.gather(one("A", 0.05), one("B", 0.01)) + + return asyncio.run(both()) + + +def test_each_of_two_runs_at_the_same_time_writes_into_the_state_that_its_caller_gave(): + pipeline = Pipeline() + pipeline.add_filter(SlowFilter()) + pipeline.add_tap(FailTap()) + pipeline.add_filter(FailForBFilter()) + state_a, state_b = _overlap(pipeline, give_state=True) + assert state_a.executed == ["SlowFilter", "FailTap", "FailForBFilter"] + assert state_a.errors == [] + assert state_b.executed == ["SlowFilter", "FailTap"] + assert [label for label, _error in state_b.errors] == ["FailTap"] + + +def test_with_no_state_from_the_caller_the_steps_of_two_runs_do_not_mix(): + pipeline = Pipeline() + pipeline.add_filter(SlowFilter()) + pipeline.add_filter(FailForBFilter()) + _overlap(pipeline, give_state=False) + # B started last, so the Pipeline has the State of B: the one step that B did, and no step of A. + assert pipeline.state.executed == ["SlowFilter"] + + +def test_a_valve_with_an_async_filter_is_skipped_or_run_for_its_own_run(): + pipeline = Pipeline() + pipeline.add_valve(IsAPredicate(), SlowFilter()) + state_a, state_b = _overlap(pipeline, give_state=True) + assert (state_a.executed, state_a.skipped) == (["SlowFilter"], []) + assert (state_b.executed, state_b.skipped) == ([], ["SlowFilter"]) + + +def test_the_pipeline_has_the_state_that_the_caller_gave(): + pipeline = Pipeline() + pipeline.add_filter(FailForBFilter()) + state = State() + pipeline.run(Payload({"name": "A"}), state) + assert pipeline.state is state + assert state.executed == ["FailForBFilter"] + pipeline.run(Payload({"name": "A"})) + assert pipeline.state is not state + assert state.executed == ["FailForBFilter"] # a later run does not write into it From 286088ffc44db9e5456bc96698d6b4e2bdecd86a Mon Sep 17 00:00:00 2001 From: Joshua Wink Date: Tue, 13 Oct 2026 08:28:24 -0500 Subject: [PATCH 2/2] test(conformance): the compiled forms run the cases with a second run The reference step no longer holds a Pipeline. The runner gives it the way to run the pipeline again: a method of the Pipeline, or for a compiled form a second call of its function. Both compilers lower the step, so no case is unsupported on any target. Co-Authored-By: Claude Opus 5.5 --- conformance/MATRIX.md | 26 +++++++------- conformance/README.md | 6 ++-- conformance/cases/state.json | 25 +------------ conformance/run.py | 2 +- .../runners/python/conformance_sink.py | 3 ++ .../python/cup_ops/async_run_again_filter.py | 9 ++--- .../python/cup_ops/run_again_filter.py | 14 +++----- conformance/runners/python/runner.py | 36 ++++++++++++++----- .../typescript/cup-ops/run-again-filter.ts | 19 +++++----- .../runners/typescript/cup-ops/sink.ts | 5 +++ conformance/runners/typescript/runner.ts | 20 ++++++----- docs/compile.md | 4 +-- tests/conformance/test_python_targets.py | 17 ++++----- tests/conformance/test_typescript_target.py | 5 +-- 14 files changed, 89 insertions(+), 102 deletions(-) diff --git a/conformance/MATRIX.md b/conformance/MATRIX.md index 9a2c172..14481c7 100644 --- a/conformance/MATRIX.md +++ b/conformance/MATRIX.md @@ -4,18 +4,16 @@ A snapshot of `python conformance/run.py`. Regenerate with that command; the live copy is `results/latest.md` (not committed). The command exits non-zero if any target has a ❌. -**Every target passes every case that it can express.** The targets are the -Python dynamic run, async run and three compiled forms, and the TypeScript -run and its compiled form. - -Two cases are ➖ for the compiled forms. In them a step starts a second run -of its own Pipeline, and a compiled form is one function with no Pipeline to -run again. +**Every target passes every case, and nothing is unsupported.** The targets +are the Python dynamic run, async run and three compiled forms, and the +TypeScript run and its compiled form. The fast modes (`python:static-steps`, `python:static-inline`, `typescript:static`) expose no State or Hooks, so they pass those cases as ✅\*. `static-observe` (compile with -`observe=True`) records both and passes every case outright. See +`observe=True`) records both and passes every case outright, but for one: +the case that asks for `state` on the Pipeline object. A compiled form is a +function and has no such object. See [docs/compile.md](../docs/compile.md). | case | python:run | python:run_async | python:static-steps | python:static-inline | python:static-observe | typescript:run | typescript:static | @@ -39,8 +37,8 @@ The fast modes (`python:static-steps`, `python:static-inline`, | `pipeline/filter-error-halts-the-run` | ✅ | ✅ | ✅* | ✅* | ✅ | ✅ | ✅* | | `pipeline/nested-pipeline-is-a-step` | ✅ | ✅ | ✅* | ✅* | ✅ | ✅ | ✅* | | `state/the-caller-gives-the-state-of-a-run` | ✅ | ✅ | ✅* | ✅* | ✅ | ✅ | ✅* | -| `state/a-run-that-starts-inside-another-run-has-its-own-state` | ✅ | ✅ | ➖ | ➖ | ➖ | ✅ | ➖ | -| `state/with-no-state-from-the-caller-the-pipeline-has-the-state-of-the-run-that-started-last` | ✅ | ✅ | ➖ | ➖ | ➖ | ✅ | ➖ | +| `state/a-run-that-starts-inside-another-run-has-its-own-state` | ✅ | ✅ | ✅* | ✅* | ✅ | ✅ | ✅* | +| `state/with-no-state-from-the-caller-the-pipeline-has-the-state-of-the-run-that-started-last` | ✅ | ✅ | ✅* | ✅* | ✅* | ✅ | ✅* | | `tap/observes-without-changing-output` | ✅ | ✅ | ✅* | ✅* | ✅ | ✅ | ✅* | | `tap/failure-never-halts-the-pipeline` | ✅ | ✅ | ✅* | ✅* | ✅ | ✅ | ✅* | | `tap/failure-does-not-stop-later-taps` | ✅ | ✅ | ✅* | ✅* | ✅ | ✅ | ✅* | @@ -53,10 +51,10 @@ The fast modes (`python:static-steps`, `python:static-inline`, - **python:run**: 30 pass, 0 pass*, 0 fail, 0 unsupported - **python:run_async**: 30 pass, 0 pass*, 0 fail, 0 unsupported -- **python:static-steps**: 11 pass, 17 pass*, 0 fail, 2 unsupported -- **python:static-inline**: 11 pass, 17 pass*, 0 fail, 2 unsupported -- **python:static-observe**: 28 pass, 0 pass*, 0 fail, 2 unsupported +- **python:static-steps**: 11 pass, 19 pass*, 0 fail, 0 unsupported +- **python:static-inline**: 11 pass, 19 pass*, 0 fail, 0 unsupported +- **python:static-observe**: 29 pass, 1 pass*, 0 fail, 0 unsupported - **typescript:run**: 30 pass, 0 pass*, 0 fail, 0 unsupported -- **typescript:static**: 11 pass, 17 pass*, 0 fail, 2 unsupported +- **typescript:static**: 11 pass, 19 pass*, 0 fail, 0 unsupported ✅* passed everything the target can observe; it does not expose State or Hooks (the fast mode). ➖ the target cannot express the case. diff --git a/conformance/README.md b/conformance/README.md index 44ad888..2871a48 100644 --- a/conformance/README.md +++ b/conformance/README.md @@ -77,7 +77,9 @@ Three more keys are for the cases about State and runs that overlap: (`run(p, state)`), and reports that State as `state`. - `expect.inner_states`: the State of each run that a `run_again` step started, in order. -- `expect.pipeline_state`: `state` on the Pipeline after the run ends. +- `expect.pipeline_state`: `state` on the Pipeline after the run ends. A + compiled form is a function and has no Pipeline object, so it cannot + show this. Every case also checks, implicitly, that **the caller's input was not mutated**. Numbers compare as JSON numbers (`1 == 1.0`), but `true` is never `1`. @@ -95,7 +97,7 @@ Numbers compare as JSON numbers (`1 == 1.0`), but `true` is never `1`. | filter | `merge` value | `merge(Payload(value))` | | filter | `has` key, to | insert `to = has(key)` | | filter | `mutate` key, value | `insert(key, value)` as a statement of its own, then return the payload | -| filter | `run_again` input | when `get("inner")` is not true: run the pipeline that this step is in one more time, with `input` as a new Payload and a new State that the step gives, and keep that State for the report. Then return the payload with no change. This is a second run that starts and ends while the first one is in progress | +| filter | `run_again` input | when `get("inner")` is not true: run the pipeline that this step is in one more time, with `input` and a new State, and keep that State for the report. Then return the payload with no change. This is a second run that starts and ends while the first one is in progress. How a pipeline is run again depends on the target (a method of the Pipeline, or for a compiled form a second call of its function), so the runner gives the step that as a function | | tap | `record` key | append `get(key)` to the runner's tap log | | tap | `fail` message | raise | | valve `when` | `equals` key, value | `get(key) == value` | diff --git a/conformance/cases/state.json b/conformance/cases/state.json index bdd5219..6bf7701 100644 --- a/conformance/cases/state.json +++ b/conformance/cases/state.json @@ -56,19 +56,6 @@ "audit" ] }, - "pipeline_state": { - "executed": [ - "first", - "audit", - "last" - ], - "skipped": [ - "gate" - ], - "errors": [ - "audit" - ] - }, "inner_states": [] } }, @@ -140,17 +127,7 @@ "skipped": [], "errors": [] } - ], - "pipeline_state": { - "executed": [ - "first", - "again", - "gate", - "last" - ], - "skipped": [], - "errors": [] - } + ] } }, { diff --git a/conformance/run.py b/conformance/run.py index d2fabac..8fab413 100644 --- a/conformance/run.py +++ b/conformance/run.py @@ -173,7 +173,7 @@ def judge(case, result): if problems: return "FAIL", problems - unchecked = [k for k in ("state", "hooks") if k in expect and k not in result] + 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", [] diff --git a/conformance/runners/python/conformance_sink.py b/conformance/runners/python/conformance_sink.py index fcc42ec..f9c6073 100644 --- a/conformance/runners/python/conformance_sink.py +++ b/conformance/runners/python/conformance_sink.py @@ -11,9 +11,12 @@ HOOKS = [] # The State of each run that a run_again step started, in order. INNER_STATES = [] +# How the target runs its pipeline one more time: one function, that the runner sets for each case. +RUN_AGAIN = [None] def reset(): del TAPS[:] del HOOKS[:] del INNER_STATES[:] + RUN_AGAIN[0] = None diff --git a/conformance/runners/python/cup_ops/async_run_again_filter.py b/conformance/runners/python/cup_ops/async_run_again_filter.py index 4620e17..cfc7e98 100644 --- a/conformance/runners/python/cup_ops/async_run_again_filter.py +++ b/conformance/runners/python/cup_ops/async_run_again_filter.py @@ -1,8 +1,6 @@ """AsyncRunAgainFilter: the reference `run_again` step for ``run_async`` (conformance/README.md).""" -from codeupipe import Payload, State - -from conformance_sink import INNER_STATES +from conformance_sink import RUN_AGAIN class AsyncRunAgainFilter: @@ -10,11 +8,8 @@ class AsyncRunAgainFilter: def __init__(self, data): self.data = data - self.pipeline = None async def call(self, payload): if payload.get("inner") is not True: - state = State() - INNER_STATES.append(state) - await self.pipeline.run_async(Payload(dict(self.data)), state) + await RUN_AGAIN[0](dict(self.data)) return payload diff --git a/conformance/runners/python/cup_ops/run_again_filter.py b/conformance/runners/python/cup_ops/run_again_filter.py index d655f3c..7408b24 100644 --- a/conformance/runners/python/cup_ops/run_again_filter.py +++ b/conformance/runners/python/cup_ops/run_again_filter.py @@ -1,24 +1,20 @@ """RunAgainFilter: the reference `run_again` step (conformance/README.md).""" -from codeupipe import Payload, State - -from conformance_sink import INNER_STATES +from conformance_sink import RUN_AGAIN class RunAgainFilter: """Starts a second run of the pipeline that it is in, inside the first one. - The runner sets ``pipeline`` after it builds the pipeline. The second run - has ``inner`` in its input, so it does not start a third one. + How a pipeline is run again depends on the target, so the runner gives + that as a function, in ``RUN_AGAIN``. The second run has ``inner`` in its + input, so it does not start a third one. """ def __init__(self, data): self.data = data - self.pipeline = None def call(self, payload): if payload.get("inner") is not True: - state = State() - INNER_STATES.append(state) - self.pipeline.run(Payload(dict(self.data)), state) + RUN_AGAIN[0](dict(self.data)) return payload diff --git a/conformance/runners/python/runner.py b/conformance/runners/python/runner.py index 279e316..52039ec 100644 --- a/conformance/runners/python/runner.py +++ b/conformance/runners/python/runner.py @@ -113,8 +113,6 @@ def build(case, cls, predicates, names, run_async=False): kind, name = step["kind"], step["name"] if kind == "filter": obj = _filter(step, run_async) - if step["op"] == "run_again": - obj.pipeline = p p.add_filter(obj) elif kind == "tap": obj = _tap(step) @@ -167,6 +165,13 @@ def _dynamic(case, predicates, run_async=False): # 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)) @@ -214,8 +219,6 @@ def _load_export(p, lowering, observe, workdir): def _static(case, predicates, lowering, workdir, observe=False): # The compiled form always gets its State from the caller, so "given_state" needs nothing more. - if any(step.get("op") == "run_again" for step in case["steps"]): - raise Unsupported("the compiled form is one function: it has no Pipeline that a step can run again") names = {} p = build(case, Pipeline, predicates, names) label = labels(p, names) @@ -226,6 +229,17 @@ def _static(case, predicates, lowering, workdir, observe=False): 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) @@ -234,11 +248,15 @@ def _static(case, predicates, lowering, workdir, observe=False): except Exception as error: # noqa: BLE001 result["error"] = f"{type(error).__name__}: {error}" if observe: - result["state"] = { - "executed": _named(state["executed"], label), - "skipped": _named(state["skipped"], label), - "errors": _named([step for step, _ in state["errors"]], label), - } + 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 diff --git a/conformance/runners/typescript/cup-ops/run-again-filter.ts b/conformance/runners/typescript/cup-ops/run-again-filter.ts index 554144d..08f0ea2 100644 --- a/conformance/runners/typescript/cup-ops/run-again-filter.ts +++ b/conformance/runners/typescript/cup-ops/run-again-filter.ts @@ -1,27 +1,24 @@ -import { Payload, State } from "../../../../packages/codeupipe-ts/src/index.ts"; -import type { Filter, Pipeline } from "../../../../packages/codeupipe-ts/src/index.ts"; +import type { Filter, Payload } from "../../../../packages/codeupipe-ts/src/index.ts"; -import { INNER_STATES } from "./sink.ts"; +import { RUN_AGAIN } from "./sink.ts"; /** * The reference `run_again` step (conformance/README.md): it starts a second - * run of the pipeline that it is in, inside the first one. The runner sets - * `pipeline` after it builds the pipeline. The second run has `inner` in its - * input, so it does not start a third one. + * run of the pipeline that it is in, inside the first one. How a pipeline is + * run again depends on the target, so the runner gives that as a function, + * in `RUN_AGAIN`. The second run has `inner` in its input, so it does not + * start a third one. */ export class RunAgainFilter implements Filter { readonly data: Record; - pipeline: Pipeline | null = null; constructor(data: Record) { this.data = data; } async call(payload: Payload): Promise { - if (payload.get("inner") !== true && this.pipeline !== null) { - const state = new State(); - INNER_STATES.push(state); - await this.pipeline.run(new Payload({ ...this.data }), state); + if (payload.get("inner") !== true) { + await RUN_AGAIN.run({ ...this.data }); } return payload; } diff --git a/conformance/runners/typescript/cup-ops/sink.ts b/conformance/runners/typescript/cup-ops/sink.ts index c4dce20..f017898 100644 --- a/conformance/runners/typescript/cup-ops/sink.ts +++ b/conformance/runners/typescript/cup-ops/sink.ts @@ -3,3 +3,8 @@ export const TAPS: unknown[] = []; /** The State of each run that a `run_again` step started, in order. */ export const INNER_STATES: Array<{ executed: string[]; skipped: string[]; errors: Array<[string, unknown]> }> = []; + +/** How the target runs its pipeline one more time. The runner sets `run` for each case. */ +export const RUN_AGAIN: { run: (input: Record) => Promise } = { + run: async () => { throw new Error("the runner did not say how to run the pipeline again"); }, +}; diff --git a/conformance/runners/typescript/runner.ts b/conformance/runners/typescript/runner.ts index 0640840..d833f9e 100644 --- a/conformance/runners/typescript/runner.ts +++ b/conformance/runners/typescript/runner.ts @@ -37,7 +37,7 @@ import { RecordTap } from "./cup-ops/record-tap.ts"; import { RemoveFilter } from "./cup-ops/remove-filter.ts"; import { SetFilter } from "./cup-ops/set-filter.ts"; import { RunAgainFilter } from "./cup-ops/run-again-filter.ts"; -import { INNER_STATES, TAPS } from "./cup-ops/sink.ts"; +import { INNER_STATES, RUN_AGAIN, TAPS } from "./cup-ops/sink.ts"; import { TruthyPredicate } from "./cup-ops/truthy-predicate.ts"; type Spec = Record; @@ -81,9 +81,7 @@ function makePredicate(when: Spec): Predicate { function build(steps: Spec[], pipeline: Pipeline): Pipeline { for (const step of steps) { if (step.kind === "filter") { - const filter = makeFilter(step); - if (filter instanceof RunAgainFilter) filter.pipeline = pipeline; - pipeline.addFilter(filter); + pipeline.addFilter(makeFilter(step)); } else if (step.kind === "tap") { pipeline.addTap(makeTap(step)); } else if (step.kind === "valve") { @@ -119,6 +117,11 @@ async function runCase(spec: Case): Promise> { const result: Record = { id: spec.id }; // With "given_state", the caller gives the State of the run. If not, the run makes one. const given = spec.given_state ? new State() : undefined; + RUN_AGAIN.run = (input) => { + const state = new State(); + INNER_STATES.push(state); + return pipeline.run(new Payload(input), state); + }; try { result.output = (await (given === undefined ? pipeline.run(new Payload(data)) : pipeline.run(new Payload(data), given))).toDict(); } catch (error) { @@ -150,12 +153,12 @@ const CLASS_OF: Record = { set: ["SetFilter", "set-filter"], copy: ["CopyFilter", "copy-filter"], add: ["AddFilter", "add-filter"], append: ["AppendFilter", "append-filter"], remove: ["RemoveFilter", "remove-filter"], merge: ["MergeFilter", "merge-filter"], has: ["HasFilter", "has-filter"], mutate: ["MutateFilter", "mutate-filter"], - fail: ["FailFilter", "fail-filter"], record: ["RecordTap", "record-tap"], "tap:fail": ["FailTap", "fail-tap"], + fail: ["FailFilter", "fail-filter"], run_again: ["RunAgainFilter", "run-again-filter"], record: ["RecordTap", "record-tap"], "tap:fail": ["FailTap", "fail-tap"], noop: ["NoopTap", "noop-tap"], equals: ["EqualsPredicate", "equals-predicate"], truthy: ["TruthyPredicate", "truthy-predicate"], }; const ARGUMENTS: Record = { set: ["key", "value"], copy: ["from", "to"], add: ["key", "by"], append: ["key", "value"], remove: ["key"], - merge: ["value"], has: ["key", "to"], mutate: ["key", "value"], fail: ["message"], record: ["key"], + merge: ["value"], has: ["key", "to"], mutate: ["key", "value"], fail: ["message"], run_again: ["input"], record: ["key"], "tap:fail": ["message"], noop: [], equals: ["key", "value"], truthy: ["key"], }; @@ -198,9 +201,6 @@ function sourceOf(spec: Case): string { type Compile = (file: string, className: string, options: { outFile: string }) => Promise; async function runStatic(spec: Case, compile: Compile, directory: string, index: number): Promise> { - if (spec.steps.some((step) => step.op === "run_again")) { - throw new Unsupported("the compiled form is one function: it has no Pipeline that a step can run again"); - } const sourceFile = join(directory, `case-${index}-pipeline.ts`); const outFile = join(directory, `case-${index}-compiled.ts`); writeFileSync(sourceFile, sourceOf(spec)); @@ -212,6 +212,8 @@ async function runStatic(spec: Case, compile: Compile, directory: string, index: } const compiled: { run(input: Record): Promise> } = await import(pathToFileURL(outFile).href); + // The compiled form is a function. To run the pipeline again is to call that function again. + RUN_AGAIN.run = (input) => compiled.run(input); const data = structuredClone(spec.input); const result: Record = { id: spec.id }; try { diff --git a/docs/compile.md b/docs/compile.md index 6550ed2..a572878 100644 --- a/docs/compile.md +++ b/docs/compile.md @@ -74,7 +74,7 @@ generated `run()` (scorecard findings F3–F8). | Type hints on a step (`Payload[Reads] -> Payload[Writes]`, hinted locals) | erased: compiled code carries no hints and never imports their names | `test_export_typed_steps` | | Payload ops `get insert remove merge has to_dict copy` | dict operations; a write is an in-place statement (`data[k] = v`) | `test_rewrite_payload_ops` | -Evidence: each conformance case that a compiled form can express (28 of 30) on both layouts (`python:static-steps`, +Evidence: all 30 conformance cases on both layouts (`python:static-steps`, `python:static-inline`), and the parity spike at 2400/2400 inputs. ## What is refused @@ -116,7 +116,7 @@ state["executed"], state["skipped"], state["errors"], state["timings"], state["t ``` It reproduces the dynamic run's observable behaviour exactly, and the -`python:static-observe` conformance target checks it on the 28 cases that a compiled form can express: +`python:static-observe` conformance target checks it on all 30 cases: - **State** is written into the `state` dict you pass (reset each run): `executed`, `skipped`, `errors` as `(name, error)`, `timings` (only with diff --git a/tests/conformance/test_python_targets.py b/tests/conformance/test_python_targets.py index 5983e07..2aaa2ac 100644 --- a/tests/conformance/test_python_targets.py +++ b/tests/conformance/test_python_targets.py @@ -16,9 +16,9 @@ PYTHON_TARGETS = ["python:run", "python:run_async", "python:static-steps", "python:static-inline", "python:static-observe"] -# The cases with a step that starts a second run of its own Pipeline. -RUNS_AGAIN = ["state/a-run-that-starts-inside-another-run-has-its-own-state", - "state/with-no-state-from-the-caller-the-pipeline-has-the-state-of-the-run-that-started-last"] +# `state` on the Pipeline is a part of the Pipeline object. A compiled form is a function and has no +# such object, so it cannot show what this one case asks for. +NEEDS_A_PIPELINE_OBJECT = ["state/with-no-state-from-the-caller-the-pipeline-has-the-state-of-the-run-that-started-last"] @pytest.fixture(scope="module") @@ -36,11 +36,8 @@ def test_target_passes_every_case_it_can_express(report, target): def test_observable_export_is_judged_on_state_and_hooks_too(report): - """No partial passes: static-observe exposes everything the dynamic run does. - - The compiled form is one function, so it cannot express a step that runs - its Pipeline again. Those cases are the only ones that it leaves out. - """ + """static-observe exposes the State and the Hooks that the dynamic run does, so it passes + each case outright, but for the one that asks for the State on the Pipeline object.""" summary = report["targets"]["python:static-observe"] - assert sorted(summary["unsupported_cases"]) == RUNS_AGAIN, summary - assert summary["pass"] == report["cases"] - len(RUNS_AGAIN), summary + assert summary["unsupported"] == 0, summary + assert summary["pass"] == report["cases"] - len(NEEDS_A_PIPELINE_OBJECT), summary diff --git a/tests/conformance/test_typescript_target.py b/tests/conformance/test_typescript_target.py index 0ed4991..47c1850 100644 --- a/tests/conformance/test_typescript_target.py +++ b/tests/conformance/test_typescript_target.py @@ -37,10 +37,7 @@ def test_the_compiled_pipeline_passes_every_case_on_what_it_can_show(): "typescript:static", conformance_run.TARGETS["typescript:static"], cases) assert error is None, error verdicts = {case["id"]: conformance_run.judge(case, results.get(case["id"])) for case in cases} - # The compiled form is one function: it cannot express a step that runs its Pipeline again. - runs_again = sorted(case["id"] for case in cases if any(step.get("op") == "run_again" for step in case["steps"])) - assert sorted(case_id for case_id, verdict in verdicts.items() if verdict[0] == "unsupported") == runs_again - wrong = {case_id: verdict for case_id, verdict in verdicts.items() if verdict[0] not in ("pass", "partial", "unsupported")} + wrong = {case_id: verdict for case_id, verdict in verdicts.items() if verdict[0] not in ("pass", "partial")} assert wrong == {}, json.dumps(wrong, indent=2) assert sum(1 for verdict in verdicts.values() if verdict[0] == "pass") == 11