Merge pull request #6844 from oliver-sanders/6537

trigger: support ^/$ syntax for referencing the initial/final cycle points

Author: wxtim

changes.d/6844.feat.md

@@ -0,0 +1 @@
+Cylc commands including `cylc trigger` now support the `^` and `$` syntax for referencing the initial and final cycle points respectively.

cylc/flow/id_match.py

@@ -21,6 +21,7 @@
     Dict,
     Iterable,
     List,
+    Optional,
     TYPE_CHECKING,
     # Tuple,
     # Union,
@@ -78,6 +79,8 @@
 def filter_ids(
     pool: 'Pool',
     ids: 'Iterable[str]',
+    icp: 'PointBase',
+    fcp: 'Optional[PointBase]',
     *,
     warn: 'bool' = True,
     out: 'IDTokens' = IDTokens.Task,
@@ -112,6 +115,7 @@ def filter_ids(
     _cycles: 'List[PointBase]' = []
     _tasks: 'List[TaskProxy]' = []
     _not_matched: 'List[str]' = []
+    _invalid: 'List[str]' = []
 
     # enable / disable pattern matching
     match: Callable[[Any, Any], bool]
@@ -138,7 +142,7 @@ def filter_ids(
         try:
             id_tokens_map[id_] = Tokens(id_, relative=True)
         except ValueError:
-            _not_matched.append(id_)
+            _invalid.append(id_)
             LOG.warning(f'Invalid ID: {id_}')
 
     for id_, tokens in id_tokens_map.items():
@@ -174,6 +178,19 @@ def filter_ids(
             task = tokens[IDTokens.Task.value]
             task_sel_raw = tokens.get(IDTokens.Task.value + '_sel')
             task_sel = task_sel_raw or '*'
+
+            # support ^/$ derefencing of the initial and final cycle points
+            if cycle == '^':
+                cycle = str(icp)
+            elif cycle == '$':
+                if not fcp:
+                    _invalid.append(id_)
+                    LOG.warning(
+                        'ID references final cycle point, but none is set:'
+                        f' {id_}'
+                    )
+                cycle = str(fcp)
+
             for icycle, itasks in pool.items():
                 if not point_match(icycle, cycle, pattern_match):
                     continue
@@ -208,7 +225,9 @@ def filter_ids(
             raise NotImplementedError
 
         if not (cycles or tasks):
-            _not_matched.append(id_)
+            _not_matched.append(
+                id_.replace('^', str(icp)).replace('$', str(fcp))
+            )
             if warn:
                 LOG.warning(f"No active tasks matching: {id_}")
         else:
@@ -227,7 +246,7 @@ def filter_ids(
             if icycle in pool:
                 _tasks.extend(pool[icycle].values())
         ret = _tasks
-    return ret, _not_matched
+    return ret, _not_matched, _invalid
 
 
 def point_match(

cylc/flow/task_pool.py

@@ -2496,9 +2496,11 @@ def filter_task_proxies(
             (matched, inactive_matched, unmatched)
 
         """
-        matched, unmatched = filter_ids(
+        matched, unmatched, invalid = filter_ids(
             self.active_tasks,
             ids,
+            self.config.initial_point,
+            self.config.final_point,
             warn=warn_no_active,
         )
         inactive_matched: 'Set[Tuple[TaskDef, PointBase]]' = set()
@@ -2507,7 +2509,7 @@ def filter_task_proxies(
                 unmatched
             )
 
-        return matched, inactive_matched, unmatched
+        return matched, inactive_matched, unmatched + invalid
 
     def match_inactive_tasks(
         self,

tests/functional/cylc-trigger/00-compat/flow.cylc

@@ -3,6 +3,6 @@
         R1 = foo => bar
 [runtime]
     [[foo]]
-        script = cylc trigger "${CYLC_WORKFLOW_ID}//1/bar"
+        script = cylc trigger "${CYLC_WORKFLOW_ID}//^/bar"
     [[bar]]
         script = true

tests/integration/test_task_pool.py

@@ -1133,6 +1133,57 @@ async def test_detect_incomplete_tasks(
             assert itask in schd.pool.get_tasks()
 
 
+async def test_trigger_icp_fcp_syntax(
+    flow,
+    scheduler,
+    start,
+    log_filter,
+):
+    """It should support the ^/$ syntax for referencing the initial/final cp.
+
+    See https://github.com/cylc/cylc-flow/issues/6537
+    """
+    cfg = {
+        'scheduling': {
+            'cycling mode': 'integer',
+            'initial cycle point': 1,
+            'final cycle point': 2,
+            'graph': {
+                'R1/^': 'start',
+                'R1/$': 'end',
+            },
+        },
+    }
+
+    # trigger tasks at both the initial and final cycle points
+    id_ = flow(cfg)
+    schd = scheduler(id_)
+    async with start(schd) as log:
+        await commands.run_cmd(
+            commands.force_trigger_tasks(schd, ['^/start', '$/end'], ['1'])
+        )
+        assert log_filter(contains='[1/start:waiting(queued)] => waiting')
+        assert log_filter(contains='[2/end:waiting(queued)] => waiting')
+        log.clear()
+
+    # clear the final cycle point
+    del cfg['scheduling']['final cycle point']
+    del cfg['scheduling']['graph']['R1/$']
+
+    # try triggering a task at the (non existent) final cycle point
+    id_ = flow(cfg)
+    schd = scheduler(id_)
+    async with start(schd):
+        await commands.run_cmd(
+            commands.force_trigger_tasks(schd, ['^/start', '$/end'], ['1'])
+        )
+        assert log_filter(contains='[1/start:waiting(queued)] => waiting')
+        assert not log_filter(contains='[2/end:waiting(queued)] => waiting')
+        assert log_filter(
+            contains='ID references final cycle point, but none is set: $/end'
+        )
+
+
 async def test_future_trigger_final_point(
     flow,
     scheduler,

tests/unit/test_id_match.py

@@ -128,9 +128,15 @@ def test_filter_ids_task_mode(task_pool, ids, matched, not_matched):
         {}
     )
 
-    _matched, _not_matched = filter_ids(pool, ids)
+    _matched, _not_matched, _invalid = filter_ids(
+        pool,
+        ids,
+        IntegerPoint('1'),
+        IntegerPoint('1'),
+    )
     assert [get_task_id(itask) for itask in _matched] == matched
     assert _not_matched == not_matched
+    assert not _invalid
 
 
 @pytest.mark.parametrize(
@@ -189,22 +195,43 @@ def test_filter_ids_cycle_mode(task_pool, ids, matched, not_matched):
         {}
     )
 
-    _matched, _not_matched = filter_ids(pool, ids, out=IDTokens.Cycle)
+    _matched, _not_matched, _invalid = filter_ids(
+        pool,
+        ids,
+        IntegerPoint('1'),
+        IntegerPoint('1'),
+        out=IDTokens.Cycle,
+    )
     assert _matched == [IntegerPoint(i) for i in matched]
     assert _not_matched == not_matched
+    assert not _invalid
 
 
 def test_filter_ids_invalid(caplog):
     """Ensure invalid IDs are handled elegantly."""
-    matched, not_matched = filter_ids({}, ['#'])
-    assert matched == []
-    assert not_matched == ['#']
+    _matched, _not_matched, _invalid = filter_ids(
+        {},
+        ['#'],
+        IntegerPoint('1'),
+        IntegerPoint('1'),
+    )
+    assert _matched == []
+    assert _not_matched == ['#']
+    assert not _invalid
     assert caplog.record_tuples == [
         ('cylc', 30, 'No active tasks matching: #'),
     ]
+
     caplog.clear()
-    matched, not_matched = filter_ids({}, ['#'], warn=False)
+    _matched, _not_matched, _invalid = filter_ids(
+        {},
+        ['#'],
+        IntegerPoint('1'),
+        IntegerPoint('1'),
+        warn=False,
+    )
     assert caplog.record_tuples == []
+    assert not _invalid
 
 
 def test_filter_ids_pattern_match_off(task_pool):
@@ -216,14 +243,17 @@ def test_filter_ids_pattern_match_off(task_pool):
         {}
     )
 
-    _matched, _not_matched = filter_ids(
+    _matched, _not_matched, _invalid = filter_ids(
         pool,
         ['1/a'],
+        IntegerPoint('1'),
+        IntegerPoint('1'),
         out=IDTokens.Task,
         pattern_match=False,
     )
     assert [get_task_id(itask) for itask in _matched] == ['1/a:x']
     assert _not_matched == []
+    assert not _invalid
 
 
 def test_filter_ids_toggle_pattern_matching(task_pool, caplog):
@@ -238,25 +268,31 @@ def test_filter_ids_toggle_pattern_matching(task_pool, caplog):
     ids = ['*/*']
 
     # ensure pattern matching works
-    _matched, _not_matched = filter_ids(
+    _matched, _not_matched, _invalid = filter_ids(
         pool,
         ids,
+        IntegerPoint('1'),
+        IntegerPoint('1'),
         out=IDTokens.Task,
         pattern_match=True,
     )
     assert [get_task_id(itask) for itask in _matched] == ['1/a:x']
     assert _not_matched == []
+    assert not _invalid
 
     # ensure pattern matching can be disabled
     caplog.clear()
-    _matched, _not_matched = filter_ids(
+    _matched, _not_matched, _invalid = filter_ids(
         pool,
         ids,
+        IntegerPoint('1'),
+        IntegerPoint('1'),
         out=IDTokens.Task,
         pattern_match=False,
     )
     assert [get_task_id(itask) for itask in _matched] == []
     assert _not_matched == ['*/*']
+    assert not _invalid
 
     # ensure the ID is logged
     assert len(caplog.record_tuples) == 1
@@ -285,36 +321,56 @@ def test_filter_ids_namespace_hierarchy(task_pool, ids, matched, not_matched):
         },
     )
 
-    _matched, _not_matched = filter_ids(
+    _matched, _not_matched, _invalid = filter_ids(
         pool,
         ids,
+        IntegerPoint('1'),
+        IntegerPoint('1'),
         pattern_match=False,
     )
 
     assert [get_task_id(itask) for itask in _matched] == matched
     assert _not_matched == not_matched
+    assert not _invalid
 
 
 def test_filter_ids_out_format():
-    filter_ids({}, [], out=IDTokens.Cycle)
+    filter_ids(
+        {},
+        [],
+        IntegerPoint('1'),
+        IntegerPoint('1'),
+        out=IDTokens.Cycle,
+    )
     with pytest.raises(ValueError):
-        filter_ids({}, [], out=IDTokens.Job)
+        filter_ids(
+            {},
+            [],
+            IntegerPoint('1'),
+            IntegerPoint('1'),
+            out=IDTokens.Job,
+        )
 
 
 def test_filter_ids_log_errors(caplog):
-    _, _not_matched = filter_ids({}, ['/////'])
-    assert _not_matched == ['/////']
+    *_, _invalid = filter_ids(
+        {},
+        ['/////'],
+        IntegerPoint('1'),
+        IntegerPoint('1'),
+    )
+    assert _invalid == ['/////']
     assert caplog.record_tuples == [('cylc', 30, 'Invalid ID: /////')]
 
 
 @pytest.mark.parametrize(
     'point, value, pattern_match, expected',
     [
-        (IntegerPoint(23), '23', True, True),
-        (IntegerPoint(23), '23', False, True),
-        (IntegerPoint(23), '2*', True, True),
-        (IntegerPoint(23), '2*', False, False),
-        (IntegerPoint(23), '2a', True, False),
+        (IntegerPoint('23'), '23', True, True),
+        (IntegerPoint('23'), '23', False, True),
+        (IntegerPoint('23'), '2*', True, True),
+        (IntegerPoint('23'), '2*', False, False),
+        (IntegerPoint('23'), '2a', True, False),
         (ISO8601Point('2049-01-01T00:00Z'), '2049', True, True),
         (ISO8601Point('2049-01-01T00:00Z'), '2049', False, True),
         (ISO8601Point('2049-03-01T00:00Z'), '2049', True, False),