diff --git a/CHANGELOG.md b/CHANGELOG.md index e36b762..b21ce56 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,9 +7,21 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Support the `outputs` and `merge_outputs` arguments of `execute_graph`. + +### Changed + +- `execute_graph` returns the requested task outputs instead of the inputs and + outputs of the task that finished last. +- `execute_graph` with `raise_on_error=False` returns no outputs when the + workflow fails, like the other Ewoks engines. + ### Fixed - `InputMergeActor`: possible deadlock for trigger loopback from a downstream node. +- Task errors are no longer discarded when another execution of that task succeeded. ## [3.0.0] - 2026-07-01 diff --git a/src/ewoksppf/bindings.py b/src/ewoksppf/bindings.py index 4e9e6cd..be42cda 100644 --- a/src/ewoksppf/bindings.py +++ b/src/ewoksppf/bindings.py @@ -2,11 +2,14 @@ import threading import warnings from contextlib import contextmanager +from typing import Any from typing import Dict from typing import Generator from typing import List +from typing import Mapping from typing import Optional from typing import Sequence +from typing import Union from ewokscore import events from ewokscore import execute_graph_decorator @@ -14,8 +17,10 @@ from ewokscore import ppftasks from ewokscore.graph import TaskGraph from ewokscore.graph import analysis +from ewokscore.graph import graph_io from ewokscore.inittask import task_executable from ewokscore.inittask import task_executable_info +from ewokscore.missing_data import MISSING_DATA from ewokscore.node import NodeIdType from ewokscore.node import get_node_label from ewokscore.node import get_varinfo @@ -30,9 +35,21 @@ from pypushflow.StopActor import StopActor from pypushflow.ThreadCounter import ThreadCounter from pypushflow.Workflow import Workflow +from pypushflow.WorkflowResults import WORKFLOW_EXCEPTION_INSTANCE_KEY +from pypushflow.WorkflowResults import OutputSelection from . import ppfrunscript +WorkflowOutputsType = Union[Dict[str, Any], Dict[NodeIdType, Dict[str, Any]]] +"""The requested outputs of all tasks merged in a single dictionary or the +requested outputs of each task. +""" + +_NEW_WORKFLOW_EXCEPTION_KEY = "_NewWorkflowException" +"""Marks the error data of `WORKFLOW_EXCEPTION_INSTANCE_KEY` as not yet +propagated to a task that handles it. +""" + def ppfname(node_id: NodeIdType) -> str: return node_id_as_string(node_id, sep="/") @@ -156,6 +173,7 @@ def _conditions_fulfilled(self, inData: dict) -> bool: def _execute(self, inData: dict, _scope_id: Optional[str] = None) -> None: trigger = self._conditions_fulfilled(inData) if trigger: + self._store_result(inData) for actor in self.listDownStreamActor: actor.trigger(inData) @@ -188,15 +206,15 @@ def connect(self, actor): actor.register_input_actor(self) def _execute(self, inData: dict, _scope_id: Optional[str] = None) -> None: - is_error = "WorkflowExceptionInstance" in inData and inData.get( - "_NewWorkflowException" + is_error = WORKFLOW_EXCEPTION_INSTANCE_KEY in inData and inData.get( + _NEW_WORKFLOW_EXCEPTION_KEY ) if is_error and not self.trigger_on_error: return try: if is_error: inData = dict(inData) - inData["_NewWorkflowException"] = False + inData[_NEW_WORKFLOW_EXCEPTION_KEY] = False # Map output names of this task to input # names of the downstream task newInData = dict() @@ -206,6 +224,7 @@ def _execute(self, inData: dict, _scope_id: Optional[str] = None) -> None: newInData[input_name] = inData[output_name] newInData[ppfrunscript.INFOKEY] = dict(inData[ppfrunscript.INFOKEY]) + self._store_result(newInData) for actor in self.listDownStreamActor: if isinstance(actor, InputMergeActor): actor.trigger(newInData, source=self) @@ -338,6 +357,7 @@ def _has_all_required_triggers(self) -> bool: def _trigger_downstream(self, retained_inputs: Optional[dict]): merged_inputs = self._downstream_inputs(retained_inputs) + self._store_result(merged_inputs) for actor in self.listDownStreamActor: actor.trigger(merged_inputs) @@ -398,11 +418,11 @@ def _clean_workflow(self): self._threadcounter = ThreadCounter(parent=self) - self._start_actor = StartActor(name="Start", **self._actor_arguments) - self._stop_actor = StopActor(name="Stop", **self._actor_arguments) + self.startActor = StartActor(name="Start", **self._actor_arguments) + self.stopActor = StopActor(name="Stop", **self._actor_arguments) self._error_actor = ErrorHandler(name="Stop on error", **self._actor_arguments) - self._connect_actors(self._error_actor, self._stop_actor) + self._connect_actors(self._error_actor, self.stopActor) @property def _actor_arguments(self): @@ -620,7 +640,7 @@ def _connect_start_actor(self, taskgraph: TaskGraph): taskactors = self._taskactors # target_id -> EwoksPythonActor or InputMergeActor targetactors = self._targetactors - start_actor = self._start_actor + start_actor = self.startActor has_start_node = False for target_id in analysis.start_nodes(taskgraph.graph): has_start_node = True @@ -634,7 +654,7 @@ def _connect_start_actor(self, taskgraph: TaskGraph): def _connect_stop_actor(self, taskgraph: TaskGraph): # task_name -> EwoksPythonActor taskactors = self._taskactors - stop_actor = self._stop_actor + stop_actor = self.stopActor has_end_node = False for source_id in analysis.end_nodes(taskgraph.graph): has_end_node = True @@ -644,28 +664,21 @@ def _connect_stop_actor(self, taskgraph: TaskGraph): raise RuntimeError(f"{taskgraph} has no end node") @contextmanager - def _run_context( + def _ewoks_run_context( self, varinfo: Optional[dict] = None, execinfo: Optional[dict] = None, task_options: Optional[dict] = None, - max_workers: Optional[int] = None, - scaling_workers: bool = True, - pool_type: Optional[str] = None, - **pool_options, ) -> Generator[None, None, None]: + """Provide the tasks with the ewoks execution options and send the ewoks + workflow events. + """ self.startargs[ppfrunscript.INFOKEY]["varinfo"] = varinfo self.startargs[ppfrunscript.INFOKEY]["task_options"] = task_options graph = self.__ewoksgraph.graph with events.workflow_context(execinfo, workflow=graph) as execinfo: self.startargs[ppfrunscript.INFOKEY]["execinfo"] = execinfo - with super()._run_context( - max_workers=max_workers, - scaling_workers=scaling_workers, - pool_type=pool_type, - **pool_options, - ): - yield + yield def run( self, @@ -681,48 +694,92 @@ def run( scaling_workers: bool = True, pool_type: Optional[str] = None, **pool_options, - ) -> dict: - if outputs is None: - outputs = [{"all": False}] - # TODO: pypushflow returns the values of the last task that was - # executed, not all end nodes as is expected here - if outputs and (outputs != [{"all": False}] or not merge_outputs): - raise ValueError( - "the Pypushflow engine can only return the merged results of end tasks" - ) - self._stop_actor.reset() - with self._run_context( - varinfo=varinfo, - execinfo=execinfo, - task_options=task_options, - max_workers=max_workers, - scaling_workers=scaling_workers, - pool_type=pool_type, - **pool_options, + ) -> WorkflowOutputsType: + r"""Execute the workflow and return the requested task outputs. + + :param startargs: Extra input data for the start actor, merged with the + graph start arguments. Not part of the Ewoks SPEC. + :param raise_on_error: Raise the exception in which the workflow ended. + When `False` no outputs are returned in that case. + :param outputs: The task outputs to be returned. All outputs of all end + tasks by default. See `ewokscore.graph.graph_io.parse_outputs`. + :param merge_outputs: Merge the outputs of all tasks in a single + dictionary. When `False` the outputs are grouped + per node id. + :param timeout: Maximum time in seconds to wait for the workflow to + finish. The outputs of unfinished tasks are missing. + :param varinfo: Data persistence configuration of the task outputs. + :param execinfo: Ewoks event handling configuration. + :param task_options: Extra options for all tasks. + :param max_workers: Maximum number of workers in the execution pool. + :param scaling_workers: Add workers to the execution pool when needed. + :param pool_type: The type of execution pool. + :param \**pool_options: Extra options for the execution pool. + :returns: The requested task outputs, merged in a single dictionary or + grouped per node id depending on `merge_outputs`. Tasks that + did not finish successfully are absent when grouped per node id. + Empty when the workflow ended in an error state and + `raise_on_error` is `False`. + """ + merge_outputs = bool(merge_outputs) + with self._ewoks_run_context( + varinfo=varinfo, execinfo=execinfo, task_options=task_options ): - startindata = dict(self.startargs) + inData = dict(self.startargs) if startargs: - startindata.update(startargs) - - self._start_actor.trigger(startindata) - self._stop_actor.join(timeout=timeout) - result = self._stop_actor.outData - if result is None: - return dict() - result = self.__parse_result(result) - ex = result.get("WorkflowExceptionInstance") - if ex is not None and raise_on_error: - raise ex - if outputs: - return result - return dict() - - def __parse_result(self, result) -> dict: + inData.update(startargs) + + result = super().run( + inData, + timeout=timeout, + max_workers=max_workers, + scaling_workers=scaling_workers, + pool_type=pool_type, + actor_outputs=self._actor_outputs(outputs), + merge_outputs=merge_outputs, + missing_value=MISSING_DATA, + raise_on_error=raise_on_error, + **pool_options, + ) + return self.__parse_result(result, merge_outputs) + + def _actor_outputs( + self, outputs: Optional[List[dict]] + ) -> Dict[EwoksPythonActor, List[OutputSelection]]: + """Tell pypushflow which actor results need to be stored and how.""" + actor_outputs: Dict[EwoksPythonActor, List[OutputSelection]] = dict() + for output_item in graph_io.parse_outputs(self.__ewoksgraph.graph, outputs): + actor = self._taskactors.get(output_item["id"]) + if actor is None: + # The output item refers to a node that is not in the graph + continue + selections = actor_outputs.setdefault(actor, list()) + selections.append( + OutputSelection( + name=output_item.get("name"), new_name=output_item.get("new_name") + ) + ) + return actor_outputs + + def __parse_result( + self, result: Mapping, merge_outputs: bool + ) -> WorkflowOutputsType: + """Resolve the values of the pypushflow result and identify the actors + by their node id. + """ + if merge_outputs: + return self.__parse_values(result) + node_ids = {actor: node_id for node_id, actor in self._taskactors.items()} + return { + node_ids[actor]: self.__parse_values(values) + for actor, values in result.items() + } + + def __parse_values(self, values: Mapping) -> Dict[str, Any]: varinfo = varinfo_from_indata(self.startargs) return { name: value_from_transfer(value, varinfo=varinfo) - for name, value in result.items() - if name is not ppfrunscript.INFOKEY + for name, value in values.items() } @@ -749,7 +806,7 @@ def execute_graph( pool_type: Optional[str] = None, pool_options: Optional[dict] = None, **deprecated_pool_options, -) -> dict: +) -> WorkflowOutputsType: if load_options is None: load_options = dict() ewoksgraph = load_graph(graph, inputs=inputs, **load_options) diff --git a/src/ewoksppf/engine.py b/src/ewoksppf/engine.py index 6c656ff..28583cb 100644 --- a/src/ewoksppf/engine.py +++ b/src/ewoksppf/engine.py @@ -35,7 +35,7 @@ def execute_graph( max_workers: Optional[int] = None, scaling_workers: bool = True, **deprecated_pool_options, - ) -> dict: + ) -> bindings.WorkflowOutputsType: return bindings.execute_graph( graph, inputs=inputs, diff --git a/src/ewoksppf/tests/test_on_error_meta.py b/src/ewoksppf/tests/test_on_error_meta.py new file mode 100644 index 0000000..9c88839 --- /dev/null +++ b/src/ewoksppf/tests/test_on_error_meta.py @@ -0,0 +1,157 @@ +from time import sleep +from typing import List + +import pytest +from ewokscore.task import Task +from ewoksutils.exceptions import TaskExecutionError +from pypushflow.WorkflowResults import WORKFLOW_EXCEPTION_INSTANCE_KEY +from pypushflow.WorkflowResults import WORKFLOW_EXCEPTION_KEY + +from ..bindings import execute_graph + + +@pytest.fixture +def global_error_node() -> dict: + return { + "graph": {"id": "test"}, + "nodes": [ + { + "id": "failing_node", + "task_type": "class", + "task_identifier": f"{__name__}.MyTask", + }, + { + "id": "meta_node", + "task_type": "class", + "task_identifier": f"{__name__}.ErrorMetadataTask", + "default_error_attributes": {"cache_if_optional": True}, + }, + { + "id": "error_handler", + "task_type": "class", + "task_identifier": f"{__name__}.ErrorHandler", + "default_error_node": True, + }, + ], + "links": [ + { + "source": "meta_node", + "target": "error_handler", + "map_all_data": True, + "required": True, + }, + ], + } + + +@pytest.fixture +def explicit_error_node() -> dict: + return { + "graph": {"id": "test"}, + "nodes": [ + { + "id": "failing_node", + "task_type": "class", + "task_identifier": f"{__name__}.MyTask", + }, + { + "id": "meta_node", + "task_type": "class", + "task_identifier": f"{__name__}.ErrorMetadataTask", + }, + { + "id": "error_handler", + "task_type": "class", + "task_identifier": f"{__name__}.ErrorHandler", + "default_error_node": False, + }, + ], + "links": [ + { + "source": "meta_node", + "target": "error_handler", + "map_all_data": True, + "required": True, + }, + { + "source": "failing_node", + "target": "error_handler", + "on_error": True, + "map_all_data": True, + "cache_if_optional": True, + }, + ], + } + + +def test_default_error_node_metadata_error_first(global_error_node): + inputs = [ + {"id": "failing_node", "name": "sleep", "value": 0.0}, + {"id": "meta_node", "name": "sleep", "value": 0.1}, + ] + _assert_failure(global_error_node, inputs) + + +def test_default_error_node_metadata_meta_first(global_error_node): + inputs = [ + {"id": "failing_node", "name": "sleep", "value": 0.1}, + {"id": "meta_node", "name": "sleep", "value": 0.0}, + ] + _assert_failure(global_error_node, inputs) + + +def test_on_error_link_metadata_error_first(explicit_error_node): + inputs = [ + {"id": "failing_node", "name": "sleep", "value": 0.0}, + {"id": "meta_node", "name": "sleep", "value": 0.1}, + ] + _assert_failure(explicit_error_node, inputs) + + +def test_on_error_link_metadata_meta_first(explicit_error_node): + inputs = [ + {"id": "failing_node", "name": "sleep", "value": 0.1}, + {"id": "meta_node", "name": "sleep", "value": 0.0}, + ] + _assert_failure(explicit_error_node, inputs) + + +class CustomError(Exception): + pass + + +class MyTask(Task, input_names=["sleep"]): + def run(self): + sleep(self.inputs.sleep) + raise CustomError("original error message") + + +class ErrorMetadataTask(Task, input_names=["sleep"], output_names=["metadata"]): + def run(self): + sleep(self.inputs.sleep) + self.outputs.metadata = True + + +class ErrorHandler( + Task, + input_names=["metadata"], + optional_input_names=[WORKFLOW_EXCEPTION_KEY, WORKFLOW_EXCEPTION_INSTANCE_KEY], +): + def run(self): + exception = getattr(self.inputs, WORKFLOW_EXCEPTION_INSTANCE_KEY) + if exception: + assert self.inputs.metadata + raise exception + + +def _assert_failure(workflow: dict, inputs: List[dict]) -> None: + with pytest.raises(TaskExecutionError, match="original error message") as exc_info: + execute_graph(workflow, inputs=inputs, pool_type="thread") + + expected = f"Execution failed for ewoks task 'failing_node' (id: 'failing_node', task: '{__name__}.MyTask'): original error message" + assert isinstance(exc_info.value.__cause__, TaskExecutionError) + assert str(exc_info.value.__cause__) == expected + + expected = "original error message" + assert isinstance(exc_info.value.__cause__.__cause__, CustomError) + assert str(exc_info.value.__cause__.__cause__) == expected diff --git a/src/ewoksppf/tests/test_ppf_workflow2.py b/src/ewoksppf/tests/test_ppf_workflow2.py index b948332..f30cfe0 100644 --- a/src/ewoksppf/tests/test_ppf_workflow2.py +++ b/src/ewoksppf/tests/test_ppf_workflow2.py @@ -1,3 +1,4 @@ +import pytest from ewokscore.tests.utils.results import assert_execute_graph_default_result from ewoksppf import execute_graph @@ -30,7 +31,12 @@ def workflow2(): def test_workflow2(ppf_log_config, tmpdir): varinfo = {"root_uri": str(tmpdir)} graph, expected = workflow2() + err_msg = "Intentional error in pythonErrorHandlerTest!" + + with pytest.raises(RuntimeError, match=err_msg): + execute_graph(graph, varinfo=varinfo) + + # The workflow failed so there are no results result = execute_graph(graph, varinfo=varinfo, raise_on_error=False) + assert result == dict() assert_execute_graph_default_result(graph, result, expected, varinfo=varinfo) - err_msg = "Intentional error in pythonErrorHandlerTest!" - assert err_msg in str(result["WorkflowExceptionInstance"]) diff --git a/src/ewoksppf/tests/test_ppf_workflow24.py b/src/ewoksppf/tests/test_ppf_workflow24.py index 3ab0a0b..c528487 100644 --- a/src/ewoksppf/tests/test_ppf_workflow24.py +++ b/src/ewoksppf/tests/test_ppf_workflow24.py @@ -1,4 +1,5 @@ from ewoksutils.import_utils import qualname +from pypushflow.WorkflowResults import WORKFLOW_EXCEPTION_INSTANCE_KEY from ewoksppf import execute_graph @@ -163,7 +164,7 @@ def test_ppf_workflow24(ppf_log_config): result = execute_graph(workflow(), inputs=inputs, raise_on_error=False) succeeded = "task1", "task2", "task3" assert result["_ppfdict"]["succeeded"] == succeeded - assert "WorkflowExceptionInstance" not in result["_ppfdict"] + assert WORKFLOW_EXCEPTION_INSTANCE_KEY not in result["_ppfdict"] inputs = [ {"name": "succeeded", "value": tuple()}, @@ -172,7 +173,7 @@ def test_ppf_workflow24(ppf_log_config): result = execute_graph(workflow(), inputs=inputs, raise_on_error=False) succeeded = "task1", "task2", "subtask1", "subtask2", "subtask3" assert result["_ppfdict"]["succeeded"] == succeeded - err_msg = str(result["_ppfdict"]["WorkflowExceptionInstance"]) + err_msg = str(result["_ppfdict"][WORKFLOW_EXCEPTION_INSTANCE_KEY]) assert "raise on name: task3" in err_msg inputs = [ @@ -189,7 +190,7 @@ def test_ppf_workflow24(ppf_log_config): "subsubtask3", ) assert result["_ppfdict"]["succeeded"] == succeeded - err_msg = str(result["_ppfdict"]["WorkflowExceptionInstance"]) + err_msg = str(result["_ppfdict"][WORKFLOW_EXCEPTION_INSTANCE_KEY]) assert "raise on name: task3" in err_msg inputs = [ @@ -199,5 +200,5 @@ def test_ppf_workflow24(ppf_log_config): result = execute_graph(workflow(), inputs=inputs, raise_on_error=False) succeeded = "task1", "task2", "subtask1", "subtask2", "subsub_handler" assert result["_ppfdict"]["succeeded"] == succeeded - err_msg = str(result["_ppfdict"]["WorkflowExceptionInstance"]) + err_msg = str(result["_ppfdict"][WORKFLOW_EXCEPTION_INSTANCE_KEY]) assert "raise on name: task3" in err_msg diff --git a/src/ewoksppf/tests/test_workflow_outputs.py b/src/ewoksppf/tests/test_workflow_outputs.py new file mode 100644 index 0000000..76c9676 --- /dev/null +++ b/src/ewoksppf/tests/test_workflow_outputs.py @@ -0,0 +1,363 @@ +from typing import Optional + +import pytest +from ewokscore.bindings import execute_graph as execute_graph_sequential +from ewokscore.missing_data import MISSING_DATA +from ewokscore.task import Task +from ewoksutils.import_utils import qualname + +from ..bindings import execute_graph + +OUTPUT_CONFIGURATIONS = [ + None, + [], + [{"all": False}], + [{"all": True}], + [{"id": "task5"}], + [{"label": "task5"}], + [{"id": "task1", "name": "inputs"}, {"id": "task4", "name": "result"}], + [{"id": "task1", "name": "inputs", "new_name": "a"}, {"id": "task4"}], + [{"id": "task1", "name": "not_an_output"}], + [{"id": "not_a_node"}], +] + + +@pytest.mark.parametrize("outputs", OUTPUT_CONFIGURATIONS) +@pytest.mark.parametrize("merge_outputs", [True, False], ids=["merge", "separate"]) +def test_outputs_like_sequential(outputs, merge_outputs, ppf_log_config): + """The Pypushflow engine returns the same outputs as the sequential engine.""" + expected = execute_graph_sequential( + _create_graph(), outputs=outputs, merge_outputs=merge_outputs + ) + result = execute_graph( + _create_graph(), outputs=outputs, merge_outputs=merge_outputs + ) + assert result == expected + + +def test_default_outputs(ppf_log_config): + """The default is the merged outputs of the end nodes. The only end node + of this graph has no outputs. + """ + assert execute_graph(_create_graph()) == dict() + + +def test_all_outputs_merged(ppf_log_config): + """'task6' is the last task with outputs so it takes precedence.""" + result = execute_graph(_create_graph(), outputs=[{"all": True}]) + assert result == {"inputs": {"a": 10, "b": 6}, "result": 16, "label": "task6"} + + +def test_all_outputs_per_task(ppf_log_config): + result = execute_graph( + _create_graph(), outputs=[{"all": True}], merge_outputs=False + ) + assert result == { + "task1": {"inputs": {"a": 1}, "result": 1, "label": "task1"}, + "task2": {"inputs": {"a": 2}, "result": 2, "label": "task2"}, + "task3": {"inputs": {"a": 1, "b": 3}, "result": 4, "label": "task3"}, + "task4": {"inputs": {"a": 2, "b": 4}, "result": 6, "label": "task4"}, + "task5": {"inputs": {"a": 4, "b": 6}, "result": 10, "label": "task5"}, + "task6": {"inputs": {"a": 10, "b": 6}, "result": 16, "label": "task6"}, + "task7": {}, + } + + +def test_selected_outputs_merged(ppf_log_config): + result = execute_graph( + _create_graph(), + outputs=[ + {"id": "task1", "name": "inputs", "new_name": "a"}, + {"id": "task4", "name": "result"}, + ], + ) + assert result == {"a": {"a": 1}, "result": 6} + + +def test_selected_outputs_per_task(ppf_log_config): + result = execute_graph( + _create_graph(), + outputs=[ + {"id": "task1", "name": "inputs", "new_name": "a"}, + {"id": "task4", "name": "result"}, + ], + merge_outputs=False, + ) + assert result == {"task1": {"a": {"a": 1}}, "task4": {"result": 6}} + + +def test_missing_output(ppf_log_config): + result = execute_graph( + _create_graph(), outputs=[{"id": "task1", "name": "not_an_output"}] + ) + assert result == {"not_an_output": MISSING_DATA} + + +def test_persistent_outputs(ppf_log_config, tmp_path): + """Task outputs are passed around as URI's which need to be resolved.""" + varinfo = {"root_uri": str(tmp_path)} + result = execute_graph( + _create_graph(), outputs=[{"id": "task5"}], varinfo=varinfo, merge_outputs=False + ) + assert result == { + "task5": {"inputs": {"a": 4, "b": 6}, "result": 10, "label": "task5"} + } + + +def test_sub_graph_outputs(ppf_log_config): + """Nodes of sub-graphs are identified by a tuple of node id's.""" + result = execute_graph( + _graph_with_sub_graph(), + inputs=[{"name": "value", "value": 0}], + outputs=[{"all": True}], + merge_outputs=False, + ) + assert result == { + "task": {"return_value": 1}, + ("sub", "subtask1"): {"return_value": 2}, + ("sub", "subtask2"): {"return_value": 3}, + } + + result = execute_graph( + _graph_with_sub_graph(), inputs=[{"name": "value", "value": 0}] + ) + assert result == {"return_value": 3} + + +def _failing_graph() -> dict: + return { + "graph": {"id": "failing_graph"}, + "nodes": [ + {"id": "task", "task_type": "method", "task_identifier": qualname(add)}, + {"id": "failing", "task_type": "class", "task_identifier": qualname(Fail)}, + ], + "links": [{"source": "task", "target": "failing", "map_all_data": True}], + } + + +@pytest.mark.parametrize("merge_outputs", [True, False]) +def test_error_without_raising(merge_outputs, ppf_log_config): + """No outputs are returned when the workflow fails, even though the first + task did finish successfully. + """ + result = execute_graph( + _failing_graph(), + inputs=[{"name": "value", "value": 0}], + outputs=[{"all": True}], + merge_outputs=merge_outputs, + raise_on_error=False, + ) + assert result == dict() + + +@pytest.mark.parametrize("outputs", [None, [], [{"all": True}]]) +@pytest.mark.parametrize("merge_outputs", [True, False]) +def test_error_raised(outputs, merge_outputs, ppf_log_config): + """The exception is raised whatever the requested outputs are.""" + with pytest.raises(RuntimeError, match="Intentional failure"): + execute_graph( + _failing_graph(), + inputs=[{"name": "value", "value": 0}], + outputs=outputs, + merge_outputs=merge_outputs, + ) + + +@pytest.mark.parametrize("merge_outputs", [True, False]) +def test_cyclic_graph_outputs(merge_outputs, ppf_log_config): + """Only the result of the last execution of a task is returned.""" + result = execute_graph( + _cyclic_graph(limit=5), outputs=[{"all": True}], merge_outputs=merge_outputs + ) + expected = {"_ppfdict": {"value": 5, "limit": 5, "repeat": False}} + if not merge_outputs: + expected = {"loop": expected} + assert result == expected + + +def add(value: Optional[int] = None, amount: int = 1) -> int: + return value + amount + + +def add_until_limit(value: int = 0, limit: int = 1, **_) -> dict: + value += 1 + return {"value": value, "repeat": value < limit} + + +class Fail(Task, optional_input_names=["return_value"]): + def run(self): + raise RuntimeError("Intentional failure") + + +class SumTask( + Task, + input_names=["a"], + optional_input_names=["b"], + output_names=["result", "inputs", "label"], +): + def run(self): + result = self.inputs.a + if self.inputs.b: + result += self.inputs.b + self.outputs.result = result + self.outputs.inputs = {k: v for k, v in self.get_input_values().items() if v} + self.outputs.label = self.label + + +def _create_graph(): + task = qualname(SumTask) + graph = {"id": "testgraph", "schema_version": "1.1"} + nodes = [ + { + "id": "task1", + "default_inputs": [{"name": "a", "value": 1}], + "task_type": "class", + "task_identifier": task, + }, + { + "id": "task2", + "default_inputs": [{"name": "a", "value": 2}], + "task_type": "class", + "task_identifier": task, + }, + { + "id": "task3", + "default_inputs": [{"name": "b", "value": 3}], + "task_type": "class", + "task_identifier": task, + }, + { + "id": "task4", + "default_inputs": [{"name": "b", "value": 4}], + "task_type": "class", + "task_identifier": task, + }, + { + "id": "task5", + "default_inputs": [{"name": "b", "value": 5}], + "task_type": "class", + "task_identifier": task, + }, + { + "id": "task6", + "default_inputs": [{"name": "b", "value": 6}], + "task_type": "class", + "task_identifier": task, + }, + { + "id": "task7", + "task_type": "class", + "task_identifier": "ewokscore.tests.examples.tasks.nooutputtask.NoOutputTask", + }, + ] + + links = [ + { + "source": "task1", + "target": "task3", + "data_mapping": [{"source_output": "result", "target_input": "a"}], + }, + { + "source": "task2", + "target": "task4", + "data_mapping": [{"source_output": "result", "target_input": "a"}], + }, + { + "source": "task3", + "target": "task5", + "data_mapping": [{"source_output": "result", "target_input": "a"}], + }, + { + "source": "task4", + "target": "task5", + "data_mapping": [{"source_output": "result", "target_input": "b"}], + }, + { + "source": "task5", + "target": "task6", + "data_mapping": [{"source_output": "result", "target_input": "a"}], + }, + { + "source": "task6", + "target": "task7", + }, + ] + + return {"graph": graph, "links": links, "nodes": nodes} + + +def _cyclic_graph(limit: int) -> dict: + return { + "graph": {"id": "cyclic_graph"}, + "nodes": [ + { + "id": "loop", + "task_type": "ppfmethod", + "task_identifier": qualname(add_until_limit), + "default_inputs": [ + {"name": "value", "value": 0}, + {"name": "limit", "value": limit}, + ], + "force_start_node": True, + } + ], + "links": [ + { + "source": "loop", + "target": "loop", + "map_all_data": True, + "conditions": [{"source_output": "repeat", "value": True}], + } + ], + } + + +def _graph_with_sub_graph() -> dict: + return { + "graph": {"id": "graph_with_subgraph"}, + "nodes": [ + {"id": "task", "task_type": "method", "task_identifier": qualname(add)}, + {"id": "sub", "task_type": "graph", "task_identifier": _sub_graph()}, + ], + "links": [ + { + "source": "task", + "target": "sub", + "sub_target": "subtask1", + "data_mapping": [ + {"source_output": "return_value", "target_input": "value"} + ], + } + ], + } + + +def _sub_graph() -> dict: + return { + "graph": { + "id": "subgraph", + "input_nodes": [{"id": "in", "node": "subtask1"}], + "output_nodes": [{"id": "out", "node": "subtask2"}], + }, + "nodes": [ + { + "id": "subtask1", + "task_type": "method", + "task_identifier": qualname(add), + }, + { + "id": "subtask2", + "task_type": "method", + "task_identifier": qualname(add), + }, + ], + "links": [ + { + "source": "subtask1", + "target": "subtask2", + "data_mapping": [ + {"source_output": "return_value", "target_input": "value"} + ], + } + ], + }