ensure that xtrigger or fails

Author: wxtim

cylc/flow/graph_parser.py

@@ -186,6 +186,7 @@ class GraphParser:
 
     # Match @xtriggers.
     REC_XTRIG = re.compile(r'@[\w\-+%]+')
+    REC_XTRIG_OR = re.compile(r'@.*\s*\|\s*@.*')
 
     # Match fully qualified parameterized single nodes.
     REC_NODE_FULL = re.compile(
@@ -577,11 +578,14 @@ def _proc_dep_pair(
         if '' in lefts or left and not all(lefts):
             raise GraphParseError(
                 f"Null task name in graph: {left} => {right}")
-
         _rights.update(*([rights] or []))
 
         for left in lefts:
             # Extract information about all nodes on the left.
+            if left and self.REC_XTRIG_OR.match(left):
+                raise GraphParseError(
+                    "'|' operator is not supported between xtriggers.")
+
             if left:
                 info = self.__class__.REC_NODES.findall(left)
                 expr = left

tests/integration/test_config.py

@@ -30,6 +30,7 @@
     WorkflowConfigError,
     XtriggerConfigError,
 )
+from cylc.flow.graph_parser import GraphParseError
 from cylc.flow.parsec.exceptions import ListValueError
 from cylc.flow.pathutil import get_workflow_run_pub_db_path
 
@@ -488,6 +489,32 @@ def test_xtrig_signature_validation(
         validate(id_)
 
 
+def test_xtrig_or_fails_validation(
+    flow: "Fixture",
+    validate: "Fixture",
+):
+    """Xtriggers cannot be chained with the 'or'
+
+    https://github.com/cylc/cylc-flow/issues/6771
+    https://github.com/cylc/cylc-flow/issues/2712
+    """
+    id_ = flow(
+        {
+            "scheduling": {
+                "initial cycle point": "2024",
+                "xtriggers": {
+                    "xrandom": "xrandom(100)",
+                    "echo": "echo(succeed=True)"
+                },
+                "graph": {"R1": "@xrandom | @echo => foo"},
+            }
+        }
+    )
+    expected_msg = "'|' operator is not supported between xtriggers."
+    with pytest.raises(GraphParseError, match=expected_msg):
+        validate(id_)
+
+
 def test_special_task_non_word_names(flow: Fixture, validate: Fixture):
     """Test validation of special tasks names with non-word characters"""
     wid = flow({