From 5751871b1fa1bd7bda5da4d691209d265aba11a1 Mon Sep 17 00:00:00 2001
From: Olof Svensson <svensson@esrf.fr>
Date: Wed, 21 Oct 2026 18:23:29 +0200
Subject: [PATCH] test_ppf_workflow28: reproduce merge node triggered twice
 below a conditional link (#23)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 src/ewoksppf/tests/test_ppf_workflow28.py | 88 +++++++++++++++++++++++
 1 file changed, 88 insertions(+)
 create mode 100644 src/ewoksppf/tests/test_ppf_workflow28.py

diff --git a/src/ewoksppf/tests/test_ppf_workflow28.py b/src/ewoksppf/tests/test_ppf_workflow28.py
new file mode 100644
index 0000000..9fe2701
--- /dev/null
+++ b/src/ewoksppf/tests/test_ppf_workflow28.py
@@ -0,0 +1,88 @@
+import threading
+from collections import Counter
+
+import pytest
+from ewokscore.graph import analysis
+from ewokscore.graph import load_graph
+from ewoksutils.import_utils import qualname
+
+from ewoksppf import execute_graph
+
+_EXECUTED = Counter()
+_LOCK = threading.Lock()
+
+
+def _executed(name: str, a=None, b=None) -> dict:
+    with _LOCK:
+        _EXECUTED[name] += 1
+    return {"a": (a or 0) + 1, "b": (b or 0) + 1}
+
+
+def start0(**kw):
+    return _executed("start0", **kw)
+
+
+def start(**kw):
+    return _executed("start", **kw)
+
+
+def up(**kw):
+    return _executed("up", **kw)
+
+
+def down(**kw):
+    return _executed("down", **kw)
+
+
+def end(**kw):
+    return _executed("end", **kw)
+
+
+def workflow28():
+    nodes = [
+        {"id": "start0", "task_type": "ppfmethod", "task_identifier": qualname(start0)},
+        {"id": "start", "task_type": "ppfmethod", "task_identifier": qualname(start)},
+        {"id": "up", "task_type": "ppfmethod", "task_identifier": qualname(up)},
+        {"id": "down", "task_type": "ppfmethod", "task_identifier": qualname(down)},
+        {"id": "end", "task_type": "ppfmethod", "task_identifier": qualname(end)},
+    ]
+    links = [
+        {
+            "source": "start0",
+            "target": "start",
+            "map_all_data": True,
+            "conditions": [{"source_output": "a", "value": 1}],
+        },
+        {"source": "start", "target": "up", "map_all_data": True},
+        {"source": "start", "target": "down", "map_all_data": True},
+        {"source": "up", "target": "end", "map_all_data": True},
+        {"source": "down", "target": "end", "map_all_data": True},
+    ]
+    return {"graph": {"id": "workflow28"}, "links": links, "nodes": nodes}
+
+
+def test_workflow28_required_links():
+    """A conditional link upstream should not make the links of the
+    unconditional branches downstream optional (issue #23)."""
+    taskgraph = load_graph(workflow28())
+    required = {
+        (source, target): analysis.link_is_required(taskgraph.graph, source, target)
+        for source, target in taskgraph.graph.edges
+    }
+    expected = {
+        ("start0", "start"): False,
+        ("start", "up"): True,
+        ("start", "down"): True,
+        ("up", "end"): True,
+        ("down", "end"): True,
+    }
+    assert required == expected
+
+
+@pytest.mark.parametrize("repeat", range(10))
+def test_workflow28(ppf_log_config, repeat):
+    """Each task should be executed exactly once (issue #23)."""
+    _EXECUTED.clear()
+    result = execute_graph(workflow28(), pool_type="thread")
+    assert _EXECUTED == {"start0": 1, "start": 1, "up": 1, "down": 1, "end": 1}
+    assert result == {"_ppfdict": {"a": 4, "b": 4}}