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..14481c7 100644 --- a/conformance/MATRIX.md +++ b/conformance/MATRIX.md @@ -11,7 +11,9 @@ 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 | @@ -34,6 +36,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 +49,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, 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, 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 a489f06..2871a48 100644 --- a/conformance/README.md +++ b/conformance/README.md @@ -71,6 +71,16 @@ 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. 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`. @@ -87,6 +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` 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 new file mode 100644 index 0000000..6bf7701 --- /dev/null +++ b/conformance/cases/state.json @@ -0,0 +1,202 @@ +[ + { + "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" + ] + }, + "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": [] + } + ] + } + }, + { + "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..8fab413 100644 --- a/conformance/run.py +++ b/conformance/run.py @@ -164,12 +164,16 @@ 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}") 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 89b2a56..f9c6073 100644 --- a/conformance/runners/python/conformance_sink.py +++ b/conformance/runners/python/conformance_sink.py @@ -9,8 +9,14 @@ TAPS = [] 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/__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..cfc7e98 --- /dev/null +++ b/conformance/runners/python/cup_ops/async_run_again_filter.py @@ -0,0 +1,15 @@ +"""AsyncRunAgainFilter: the reference `run_again` step for ``run_async`` (conformance/README.md).""" + +from conformance_sink import RUN_AGAIN + + +class AsyncRunAgainFilter: + """``RunAgainFilter`` for a pipeline that runs with ``run_async``.""" + + def __init__(self, data): + self.data = data + + async def call(self, payload): + if payload.get("inner") is not True: + 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 new file mode 100644 index 0000000..7408b24 --- /dev/null +++ b/conformance/runners/python/cup_ops/run_again_filter.py @@ -0,0 +1,20 @@ +"""RunAgainFilter: the reference `run_again` step (conformance/README.md).""" + +from conformance_sink import RUN_AGAIN + + +class RunAgainFilter: + """Starts a second 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. + """ + + def __init__(self, data): + self.data = data + + def call(self, payload): + if payload.get("inner") is not True: + RUN_AGAIN[0](dict(self.data)) + return payload diff --git a/conformance/runners/python/runner.py b/conformance/runners/python/runner.py index 37b5859..52039ec 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,12 @@ 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) p.add_filter(obj) elif kind == "tap": obj = _tap(step) @@ -155,26 +157,40 @@ 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 = {} + + 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))) + 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 +218,7 @@ 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. names = {} p = build(case, Pipeline, predicates, names) label = labels(p, names) @@ -212,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) @@ -220,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 new file mode 100644 index 0000000..08f0ea2 --- /dev/null +++ b/conformance/runners/typescript/cup-ops/run-again-filter.ts @@ -0,0 +1,25 @@ +import type { Filter, Payload } from "../../../../packages/codeupipe-ts/src/index.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. 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; + + constructor(data: Record) { + this.data = data; + } + + async call(payload: Payload): Promise { + 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 05246a9..f017898 100644 --- a/conformance/runners/typescript/cup-ops/sink.ts +++ b/conformance/runners/typescript/cup-ops/sink.ts @@ -1,2 +1,10 @@ /** 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]> }> = []; + +/** 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 5e7c7eb..d833f9e 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, RUN_AGAIN, 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}'`); } } @@ -113,17 +115,26 @@ 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; + RUN_AGAIN.run = (input) => { + const state = new State(); + INNER_STATES.push(state); + return pipeline.run(new Payload(input), state); + }; 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) => { @@ -142,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"], }; @@ -201,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 { @@ -231,6 +244,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..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: all 27 conformance cases 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 all 27 cases: +`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/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..2aaa2ac 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"] +# `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") def report(tmp_path_factory): @@ -32,6 +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.""" + """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 summary["pass"] == report["cases"], summary + assert summary["unsupported"] == 0, summary + assert summary["pass"] == report["cases"] - len(NEEDS_A_PIPELINE_OBJECT), summary 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