From cbb34b7243fa17aba2061d2e4b3ee9a0feaf4c4f Mon Sep 17 00:00:00 2001 From: alvaro Date: Fri, 28 Aug 2026 11:50:40 +0200 Subject: [PATCH] every IndexedStep can now store its own variables --- CHANGELOG.md | 17 +++ docs/source/yaml_format.rst | 9 ++ src/pypts/steps.py | 79 +++++++------ tests/unit_tests/test_steps.py | 200 +++++++++++++++++++++++++++++++++ 4 files changed, 272 insertions(+), 33 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 2fa3205..5a3aace 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,23 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Fixed + +- Indexed steps can now store values. `local` and `global` output mappings on a + step with an `indexed: true` input are no longer discarded: the wrapper saves + the aggregated list of per-iteration values (one entry per iteration) under the + configured `local_name` / `global_name`. `passfail`, `equals`, `range` and + `image` outputs continue to be evaluated and attached per iteration. +- An indexed step whose indexed lists are empty now stores empty lists instead of + erroring, and an iteration that errors no longer masks its own failure with a + `KeyError` while the aggregate is stored. + +### Removed + +- Handling of `indexed: true` on an *output* mapping in `IndexedStep`. Every + output model is `extra="forbid"`, so the field was rejected by validation and + the branch was unreachable from any valid recipe. + ## [0.6.1] - 2026-08-27 - Add CHANGELOG.md diff --git a/docs/source/yaml_format.rst b/docs/source/yaml_format.rst index dca8d42..32ea332 100644 --- a/docs/source/yaml_format.rst +++ b/docs/source/yaml_format.rst @@ -72,6 +72,15 @@ verdict mapping on a step. ``local`` and ``global`` store values, while :ref:`recipe-v2-output-passfail` through :ref:`recipe-v2-output-range` definitions for exact fields. +On a step with an indexed input, output mappings are split between the iterations +and the wrapper. ``passfail``, ``equals``, ``range`` and ``image`` are evaluated +(or attached to the report) once per iteration, so each iteration gets its own +verdict row and its own images. ``local`` and ``global`` are stored once, by the +wrapper, and receive the **list** of per-iteration values — one entry per +iteration, in iteration order. An indexed step whose indexed lists are empty +skips and stores empty lists. ``indexed`` is an input-only field; it is rejected +on an output mapping. + Runtime-specific behavior ------------------------- diff --git a/src/pypts/steps.py b/src/pypts/steps.py index 4a3b5e8..c8e6e7a 100644 --- a/src/pypts/steps.py +++ b/src/pypts/steps.py @@ -20,6 +20,12 @@ logger = logging.getLogger(__name__) +# Output types evaluated per iteration by each generated sub-step. +_PER_ITERATION_RESULT_TYPES = ("passthrough", "passfail", "equals", "range") +# Output types the IndexedStep wrapper owns: values are aggregated across all +# iterations and stored once, so the iterations must not write them N times. +_STORAGE_OUTPUT_TYPES = ("local", "global") + class IndexedStep(Step): """ @@ -40,11 +46,20 @@ def __init__(self, step: Step, **kwargs): raise TypeError(f"IndexedStep requires a valid Step instance, got {type(step)}") self.template_step: Step = step self.steps: List[Step] = [] # Stores the generated step instances for each run - # Override output_mapping for the wrapper step itself to capture the aggregate result - self.output_mapping = {"__result": {"type": "passthrough"}} - # Note: Any output mappings defined in the original step_data for the wrapper - # (like saving the aggregated result to a variable) should be added here or - # processed by the base Step.process_outputs method using self.output_mapping. + # The wrapper owns the storage (local/global) mappings: their values are + # aggregated over every iteration and saved once, from process_outputs. + # Verdict and image mappings stay on the template step so they are evaluated + # (or attached to the report) per iteration. + # A new dict is built on purpose: build_step hands the *same* output_mapping + # object to both the template step and this wrapper, so mutating in place would + # corrupt the template and break the aggregation below. + self.output_mapping = { + name: config + for name, config in self.output_mapping.items() + if isinstance(config, dict) and config.get("type") in _STORAGE_OUTPUT_TYPES + } + # The wrapper's own verdict is the aggregate of every iteration's result. + self.output_mapping["__result"] = {"type": "passthrough"} def check_indexing(self): """ @@ -79,6 +94,17 @@ def _step(self, runtime: Runtime, input: dict, parent_step_result_uuid: uuid.UUI indexed_list_names = list(indexed_input_configs.keys()) + # Pre-seed the aggregation with an empty list per non-verdict output of the + # template step. Every key the wrapper's output_mapping may reference then + # always exists in the returned dict, even when an iteration errors (which + # leaves StepResult.outputs empty) or when there is nothing to run at all. + aggregated_outputs = { + name: [] + for name, config in self.template_step.output_mapping.items() + if isinstance(config, dict) + and config.get("type") not in _PER_ITERATION_RESULT_TYPES + } + # Get the actual input *values* provided to the IndexedStep wrapper wrapper_inputs = input @@ -98,7 +124,9 @@ def _step(self, runtime: Runtime, input: dict, parent_step_result_uuid: uuid.UUI num_runs = min(len(wrapper_inputs[name]) for name in indexed_list_names) if num_runs == 0: logger.warning(f"IndexedStep '{self.name}' has empty lists for indexed inputs. Skipping execution.") - return {"__result": ResultType.SKIP} # Or DONE? SKIP seems appropriate. + # Storage outputs still resolve — to empty lists — so the wrapper's + # process_outputs does not fail on a missing key. + return {**aggregated_outputs, "__result": ResultType.SKIP} # Or DONE? SKIP seems appropriate. except ValueError: # Handle case where indexed_list_names is empty after checks logger.warning(f"IndexedStep '{self.name}' inconsistency: Indexed inputs found, but failed to determine run count. Running once.") num_runs = 1 @@ -132,22 +160,15 @@ def _step(self, runtime: Runtime, input: dict, parent_step_result_uuid: uuid.UUI copied_step.input_mapping = iteration_input_mapping # Remove local/global variable *saving* definitions from the copied step's *output* mapping. - # We want to aggregate these values in the wrapper, not have each iteration save potentially overwriting variables. - # Pass/Fail/Range/Equals checks on outputs should still happen per iteration. - output_mapping_keys = list(copied_step.output_mapping.keys()) - for key in output_mapping_keys: - output_conf = copied_step.output_mapping[key] - if isinstance(output_conf, dict): - if output_conf.get("indexed", False): - values = output_conf.get("value", []) - if i < len(values): - copied_step.output_mapping[key]["value"] = values[i] - else: - logger.warning(f"No indexed output value for iteration {i} in '{self.name}'") - copied_step.output_mapping[key]["value"] = None #could also be a specified standard value. - elif output_conf.get("type") in ["local", "global"]: - logger.debug(f"Removing output mapping '{key}' (type: {output_conf.get('type')}) from iteration {i} of '{self.name}'") - del copied_step.output_mapping[key] + # The wrapper aggregates these values and saves them once, instead of each + # iteration overwriting the same variable. + # Pass/Fail/Range/Equals checks and image outputs still happen per iteration. + copied_step.output_mapping = { + name: config + for name, config in copied_step.output_mapping.items() + if not (isinstance(config, dict) + and config.get("type") in _STORAGE_OUTPUT_TYPES) + } # Modify the name for clarity in logs and results. # If step_name contains {input_name} placeholders, substitute @@ -174,20 +195,12 @@ def _step(self, runtime: Runtime, input: dict, parent_step_result_uuid: uuid.UUI # Aggregate outputs from the individual step results. # We only aggregate outputs that were *not* used for pass/fail/range/equals checks # within the iterations (i.e., likely intended as data outputs). - aggregated_outputs = {} - # These types imply the output was used for per-iteration result calculation. - per_iteration_result_types = ["passthrough", "passfail", "equals", "range"] - + # aggregated_outputs was pre-seeded above with the template's non-verdict + # outputs, so only those names are collected here. for result in step_results: # Iterate through the outputs recorded in the StepResult for this iteration for output_name, output_value in result.outputs.items(): - # Check the *template* step's original output mapping config for this output name. - template_output_conf = self.template_step.output_mapping.get(output_name) - - # If the output exists in the template config AND its type was NOT a per-iteration check, aggregate it. - if template_output_conf and isinstance(template_output_conf, dict) and template_output_conf.get("type") not in per_iteration_result_types: - if output_name not in aggregated_outputs: - aggregated_outputs[output_name] = [] + if output_name in aggregated_outputs: aggregated_outputs[output_name].append(output_value) # Determine the overall result of the IndexedStep based on all iteration results. diff --git a/tests/unit_tests/test_steps.py b/tests/unit_tests/test_steps.py index d64a181..99b7eba 100644 --- a/tests/unit_tests/test_steps.py +++ b/tests/unit_tests/test_steps.py @@ -51,6 +51,20 @@ def _step(self, runtime, input, parent_step): return input +class EchoWithVerdictStep(Step): + """Echoes the indexed input 'a' as a data output plus a passing verdict.""" + def _step(self, runtime, input, parent_step): + return {"value": input["a"], "passed": True} + + +class FlakyEchoStep(Step): + """Echoes the indexed input 'a', but raises when it is 20.""" + def _step(self, runtime, input, parent_step): + if input["a"] == 20: + raise ValueError("deliberate failure") + return {"value": input["a"]} + + @pytest.fixture def mock_runtime(): """A lightweight mock Runtime that satisfies Step.run's contract.""" @@ -423,6 +437,34 @@ def test_build_indexed_step_wraps_when_indexed(self): assert isinstance(step, IndexedStep) assert isinstance(step.template_step, WaitStep) + def test_build_indexed_step_keeps_storage_output_mapping(self): + """Storage outputs survive on the wrapper; verdicts stay on the template.""" + from pypts.recipe_language import WaitStep as WaitDefinition + + definition = WaitDefinition( + steptype="WaitStep", + step_name="IndexedWait", + description="Wait twice and store the waits.", + input_mapping={ + "wait_time": { + "type": "direct", "value": [0.01, 0.02], "indexed": True + } + }, + output_mapping={ + "waited": {"type": "local", "local_name": "waited"}, + "passed": {"type": "passfail"}, + }, + ) + step = Step.build_step(definition) + assert isinstance(step, IndexedStep) + assert step.output_mapping["waited"] == { + "type": "local", "local_name": "waited" + } + assert step.output_mapping["__result"] == {"type": "passthrough"} + assert "passed" not in step.output_mapping + # The template keeps the full authored mapping (shared-dict aliasing guard). + assert set(step.template_step.output_mapping) == {"waited", "passed"} + def test_rejects_unvalidated_dictionary(self): with pytest.raises(TypeError, match="validated step definition"): Step.build_step({"steptype": "WaitStep"}) @@ -656,6 +698,164 @@ def test_empty_indexed_list_skips(self, mock_runtime): result = indexed.run(mock_runtime, {}) assert result.result == ResultType.SKIP + # --- Output storage ------------------------------------------------- + + @staticmethod + def _prepare(runtime): + """Give a real Runtime the scopes and metadata Step.run needs.""" + runtime.stop_event = Event() + runtime.push_locals({}) + runtime.set_globals({}) + runtime.recipe_name = "T" + runtime.recipe_file_name = "t.yaml" + runtime.serial_number = "SN" + runtime.current_sequence_name = "Main" + + def test_local_output_stores_aggregated_list(self, real_runtime): + self._prepare(real_runtime) + # build_step passes one and the same mapping object to both, so do that here. + mapping = {"a": {"type": "local", "local_name": "measured"}} + template = PassFailStep( + step_name="Inner", + input_mapping={"a": {"type": "direct", "indexed": True}}, + output_mapping=mapping, + ) + indexed = IndexedStep( + template, + step_name="Outer", + input_mapping={ + "a": {"type": "direct", "value": [10, 20, 30], "indexed": True} + }, + output_mapping=mapping, + ) + + result = indexed.run(real_runtime, {}) + assert real_runtime.get_local("measured") == [10, 20, 30] + assert result.outputs["a"] == [10, 20, 30] + + def test_global_output_stores_aggregated_list(self, real_runtime): + self._prepare(real_runtime) + mapping = {"a": {"type": "global", "global_name": "saved"}} + template = PassFailStep( + step_name="Inner", + input_mapping={"a": {"type": "direct", "indexed": True}}, + output_mapping=mapping, + ) + indexed = IndexedStep( + template, + step_name="Outer", + input_mapping={ + "a": {"type": "direct", "value": [1, 2], "indexed": True} + }, + output_mapping=mapping, + ) + + indexed.run(real_runtime, {}) + assert real_runtime.get_global("saved") == [1, 2] + + def test_verdict_and_storage_coexist(self, real_runtime): + """A verdict output next to a storage output must not trip the + passthrough-exclusivity rule, and both must still take effect.""" + self._prepare(real_runtime) + mapping = { + "passed": {"type": "passfail"}, + "value": {"type": "local", "local_name": "measured"}, + } + template = EchoWithVerdictStep( + step_name="Inner", + input_mapping={"a": {"type": "direct", "indexed": True}}, + output_mapping=mapping, + ) + indexed = IndexedStep( + template, + step_name="Outer", + input_mapping={ + "a": {"type": "direct", "value": [10, 20], "indexed": True} + }, + output_mapping=mapping, + ) + + result = indexed.run(real_runtime, {}) + assert result.error_info == "" + assert result.result == ResultType.PASS + assert real_runtime.get_local("measured") == [10, 20] + # The verdict was evaluated once per iteration, on the sub-steps. + assert [sub.result for sub in result.subresults] == [ + ResultType.PASS, ResultType.PASS + ] + + def test_wrapper_and_template_mappings_are_split(self, mock_runtime): + mapping = { + "passed": {"type": "passfail"}, + "value": {"type": "local", "local_name": "measured"}, + "chart": {"type": "image"}, + } + template = EchoWithVerdictStep( + step_name="Inner", + input_mapping={"a": {"type": "direct", "indexed": True}}, + output_mapping=mapping, + ) + indexed = IndexedStep( + template, + step_name="Outer", + input_mapping={ + "a": {"type": "direct", "value": [10, 20], "indexed": True} + }, + output_mapping=mapping, + ) + + # Wrapper keeps only storage mappings, plus its own aggregate verdict. + assert set(indexed.output_mapping) == {"value", "__result"} + # The shared authored mapping is left untouched on the template. + assert set(template.output_mapping) == {"passed", "value", "chart"} + + indexed.run(mock_runtime, {}) + for sub in indexed.steps: + assert set(sub.output_mapping) == {"passed", "chart"} + + def test_empty_indexed_list_stores_empty_list(self, real_runtime): + self._prepare(real_runtime) + mapping = {"a": {"type": "local", "local_name": "measured"}} + template = PassFailStep( + step_name="Inner", + input_mapping={"a": {"type": "direct", "indexed": True}}, + output_mapping=mapping, + ) + indexed = IndexedStep( + template, + step_name="Outer", + input_mapping={"a": {"type": "direct", "value": [], "indexed": True}}, + output_mapping=mapping, + ) + + result = indexed.run(real_runtime, {}) + assert result.result == ResultType.SKIP + assert real_runtime.get_local("measured") == [] + + def test_errored_iteration_still_stores_list(self, real_runtime): + """A failing iteration must surface as ERROR, not be masked by a + KeyError while storing the aggregate.""" + self._prepare(real_runtime) + mapping = {"value": {"type": "local", "local_name": "measured"}} + template = FlakyEchoStep( + step_name="Inner", + input_mapping={"a": {"type": "direct", "indexed": True}}, + output_mapping=mapping, + ) + indexed = IndexedStep( + template, + step_name="Outer", + input_mapping={ + "a": {"type": "direct", "value": [10, 20, 30], "indexed": True} + }, + output_mapping=mapping, + ) + + result = indexed.run(real_runtime, {}) + assert result.result == ResultType.ERROR + assert result.error_info == "" # the wrapper itself did not raise + assert real_runtime.get_local("measured") == [10] + # ============================================================ # PythonModuleStep