Fix a terminal graph items bug

cylc/flow/graph_parser.py

@@ -21,6 +21,7 @@
 from typing import (
     Set,
     Dict,
+    Iterable,
     List,
     Tuple,
     Optional,
@@ -168,6 +169,7 @@ class GraphParser:
     _RE_OFFSET = r'\[[\w\-\+\^:]+\]'
     _RE_QUAL = QUALIFIER + r'[\w\-]+'  # task or fam trigger
     _RE_OPT = r'\??'  # optional output indicator
+    _RE_ANDOR = re.compile(r'\s*[&|]\s*')
 
     REC_QUAL = re.compile(_RE_QUAL)
 
@@ -470,13 +472,33 @@ def parse_graph(self, graph_string: str) -> None:
                 pairs.add((chain[i], chain[i + 1]))
 
         # Get a set of RH nodes which are not at the LH of another pair:
-        self.terminals = {p[1] for p in pairs}.difference(
-            {p[0] for p in pairs}
-        )
+        self.terminals = self.get_graph_terminals(pairs)
 
         for pair in sorted(pairs, key=lambda p: str(p[0])):
             self._proc_dep_pair(pair, self.terminals)
 
+    @staticmethod
+    def get_graph_terminals(pairs: Iterable[Tuple[str]]) -> set[str]:
+        """Get terminating ends of graphs.
+
+        For example in `foo => bar => baz` only `baz` terminates the chain
+        of dependencies.
+
+        Examples:
+            >>> this = GraphParser.get_graph_terminals
+            >>> this({('foo', 'bar')})
+            {'bar'}
+        """
+        lefts = []
+        for left, _ in pairs:
+            if left and ('&' in left or '|' in left):
+                # RE used because don't want to have to worry about
+                # mutiple and/or and cleaning whitespace.
+                lefts += GraphParser._RE_ANDOR.split(left)
+            else:
+                lefts.append(left)
+        return {p[1] for p in pairs}.difference(set(lefts))
+
     @classmethod
     def _report_invalid_lines(cls, lines: List[str]) -> None:
         """Raise GraphParseError in a consistent format when there are

tests/unit/test_graph_parser.py

@@ -27,7 +27,6 @@
 from cylc.flow.graph_parser import GraphParser
 from cylc.flow.task_outputs import (
     TASK_OUTPUT_SUBMITTED,
-    TASK_OUTPUT_SUBMIT_FAILED,
     TASK_OUTPUT_STARTED,
     TASK_OUTPUT_SUCCEEDED,
     TASK_OUTPUT_FAILED
@@ -810,7 +809,6 @@ def test_cannot_be_required():
         gp.parse_graph('a:submit-failed => b')
 
 
-
 @pytest.mark.parametrize(
     'graph, error',
     [
@@ -987,3 +985,42 @@ def test_proc_dep_pair(args, err):
             gp._proc_dep_pair(*args)
     else:
         assert gp._proc_dep_pair(*args) is None
+
+
+@pytest.mark.parametrize(
+    'pairs, terminals',
+    (
+        param(
+            {
+                (None, 'foo[-P1D]:restart1'),
+                (None, 'bar?'),
+                ('foo[-P1D]:restart1|bar?', 'foo')
+            },
+            {'foo'},
+            id='|'
+        ),
+        param(
+            {
+                (None, 'foo[-P1D]:restart1'),
+                (None, 'bar?'),
+                ('foo[-P1D]:restart1 & bar?', 'foo')
+            },
+            {'foo'},
+            id='&'
+        ),
+        param(
+            {
+                (None, 'foo[-P1D]:restart1'),
+                (None, 'bar?'),
+                (None, 'qux'),
+                ('foo[-P1D]:restart1 &bar? |qux', 'foo')
+            },
+            {'foo'},
+            id='&-and-|'
+        ),
+    )
+)
+def test_get_graph_terminals(pairs, terminals):
+    """It identifies all graph terminals, and no non terminals.
+    """
+    assert GraphParser.get_graph_terminals(pairs) == terminals