From 8db37bd972f430a365dd3fcda514a44e343da19f Mon Sep 17 00:00:00 2001 From: Olof Svensson Date: Wed, 28 Oct 2026 10:50:17 +0100 Subject: [PATCH] test_ppf_workflow30/31: self-triggering loop node next to an independent start node (#16) - test_ppf_workflow30: the example from the issue, where "save" has only optional inputs. Runs with both force_start_node and the config -> loop link, repeated to catch "the pool is closed" errors. - test_ppf_workflow31: "save" with required inputs. Passes when the loop -> save link is explicitly required. Strict xfail tests record that ewokscore makes the link optional by default because of the conditional self-link on "loop". --- src/ewoksppf/tests/test_ppf_workflow30.py | 106 ++++++++++++++++++++ src/ewoksppf/tests/test_ppf_workflow31.py | 116 ++++++++++++++++++++++ 2 files changed, 222 insertions(+) create mode 100644 src/ewoksppf/tests/test_ppf_workflow30.py create mode 100644 src/ewoksppf/tests/test_ppf_workflow31.py diff --git a/src/ewoksppf/tests/test_ppf_workflow30.py b/src/ewoksppf/tests/test_ppf_workflow30.py new file mode 100644 index 0000000..408ea96 --- /dev/null +++ b/src/ewoksppf/tests/test_ppf_workflow30.py @@ -0,0 +1,106 @@ +import threading + +import pytest +from ewokscore.task import Task +from ewoksutils.import_utils import qualname + +from ewoksppf import execute_graph + +_SAVED = list() +_LOCK = threading.Lock() + + +class ConfigTask(Task, output_names=["config"]): + def run(self): + self.outputs.config = {} + + +class LoopTask( + Task, + input_names=["i", "n"], + optional_input_names=["config"], + output_names=["i", "keep_looping"], +): + def run(self): + self.outputs.i = self.inputs.i + 1 + self.outputs.keep_looping = self.outputs.i < self.inputs.n + + +class SaveTask(Task, optional_input_names=["i", "config"], output_names=["result"]): + def run(self): + if self.missing_inputs.i: + self.outputs.result = False + else: + with _LOCK: + _SAVED.append(self.inputs.i) + self.outputs.result = True + + +def workflow30(force_start_node: bool, n: int = 10): + """Self-triggering "loop" node next to an independent "config" start node (issue #16). + + The "loop" node is started either by marking it explicitly as a start node + or by adding a link from "config" to "loop". + """ + loop_node = { + "id": "loop", + "task_type": "class", + "task_identifier": qualname(LoopTask), + "default_inputs": [{"name": "i", "value": 0}, {"name": "n", "value": n}], + } + if force_start_node: + loop_node["force_start_node"] = True + nodes = [ + { + "id": "config", + "task_type": "class", + "task_identifier": qualname(ConfigTask), + }, + loop_node, + { + "id": "save", + "task_type": "class", + "task_identifier": qualname(SaveTask), + }, + ] + links = [ + { + "source": "loop", + "target": "loop", + "data_mapping": [{"source_output": "i", "target_input": "i"}], + "conditions": [{"source_output": "keep_looping", "value": True}], + }, + { + "source": "loop", + "target": "save", + "data_mapping": [{"source_output": "i", "target_input": "i"}], + }, + { + "source": "config", + "target": "save", + "data_mapping": [{"source_output": "config", "target_input": "config"}], + }, + ] + if not force_start_node: + links.append( + { + "source": "config", + "target": "loop", + "data_mapping": [{"source_output": "config", "target_input": "config"}], + } + ) + return {"graph": {"id": "workflow30"}, "nodes": nodes, "links": links} + + +@pytest.mark.parametrize("repeat", range(10)) +@pytest.mark.parametrize("force_start_node", [False, True]) +def test_workflow30(ppf_log_config, force_start_node, repeat): + """Every iteration of the self-triggering "loop" node should be saved + without "the pool is closed" errors (issue #16).""" + _SAVED.clear() + execute_graph( + workflow30(force_start_node), + pool_type="thread", + scaling_workers=False, + ) + assert sorted(_SAVED) == list(range(1, 11)) diff --git a/src/ewoksppf/tests/test_ppf_workflow31.py b/src/ewoksppf/tests/test_ppf_workflow31.py new file mode 100644 index 0000000..06df3f4 --- /dev/null +++ b/src/ewoksppf/tests/test_ppf_workflow31.py @@ -0,0 +1,116 @@ +import threading + +import pytest +from ewokscore.graph import analysis +from ewokscore.graph import load_graph +from ewokscore.task import Task +from ewoksutils.import_utils import qualname + +from ewoksppf import execute_graph + +_SAVED = list() +_LOCK = threading.Lock() + + +class ConfigTask(Task, output_names=["config"]): + def run(self): + self.outputs.config = {} + + +class LoopTask( + Task, + input_names=["i", "n"], + output_names=["i", "keep_looping"], +): + def run(self): + self.outputs.i = self.inputs.i + 1 + self.outputs.keep_looping = self.outputs.i < self.inputs.n + + +class SaveTask(Task, input_names=["i", "config"], output_names=["result"]): + def run(self): + with _LOCK: + _SAVED.append(self.inputs.i) + self.outputs.result = True + + +def workflow31(required_loop_link: bool, n: int = 10): + """Self-triggering "loop" node and an independent "config" node which + both provide required inputs to "save" (issue #16).""" + loop_to_save = { + "source": "loop", + "target": "save", + "data_mapping": [{"source_output": "i", "target_input": "i"}], + } + if required_loop_link: + loop_to_save["required"] = True + nodes = [ + { + "id": "config", + "task_type": "class", + "task_identifier": qualname(ConfigTask), + }, + { + "id": "loop", + "force_start_node": True, + "task_type": "class", + "task_identifier": qualname(LoopTask), + "default_inputs": [{"name": "i", "value": 0}, {"name": "n", "value": n}], + }, + { + "id": "save", + "task_type": "class", + "task_identifier": qualname(SaveTask), + }, + ] + links = [ + { + "source": "loop", + "target": "loop", + "data_mapping": [{"source_output": "i", "target_input": "i"}], + "conditions": [{"source_output": "keep_looping", "value": True}], + }, + loop_to_save, + { + "source": "config", + "target": "save", + "data_mapping": [{"source_output": "config", "target_input": "config"}], + }, + ] + return {"graph": {"id": "workflow31"}, "nodes": nodes, "links": links} + + +@pytest.mark.parametrize("repeat", range(5)) +def test_workflow31_required_link(ppf_log_config, repeat): + """With an explicitly required link from "loop", "save" should only be + triggered when both "loop" and "config" have provided their inputs.""" + _SAVED.clear() + execute_graph( + workflow31(required_loop_link=True), + pool_type="thread", + scaling_workers=False, + ) + assert sorted(_SAVED) == list(range(1, 11)) + + +@pytest.mark.xfail( + strict=True, + reason="the conditional self-link of 'loop' makes 'loop -> save' optional by default (issue #16)", +) +def test_workflow31_default_link_is_required(): + taskgraph = load_graph(workflow31(required_loop_link=False)) + assert analysis.link_is_required(taskgraph.graph, "loop", "save") + + +@pytest.mark.xfail( + strict=True, + reason="'save' is triggered by 'config' before 'loop' provides 'i' (issue #16)", +) +def test_workflow31_default_link(ppf_log_config): + _SAVED.clear() + execute_graph( + workflow31(required_loop_link=False), + pool_type="thread", + scaling_workers=False, + ) + assert sorted(_SAVED) == list(range(1, 11))