Response to reviews

Update changes.d/6583.fix.md

Co-authored-by: Hilary James Oliver <[email protected]>

Update cylc/flow/config.py

Co-authored-by: Hilary James Oliver <[email protected]>

respond to Hilary's review: Don't check all outputs only terminals.

use regexs to sort out issues with qualifiers, suicide triggers &c

mypy

Author: wxtim

changes.d/6583.fix.md

@@ -1 +1 @@
-Fix bug where graph items with undefined outputs were missed at validation if the graph item was not an upstream dependency of another graph item.
+Fix bug where undefined outputs were missed by validation if no tasks trigger off of them.

cylc/flow/config.py

@@ -99,6 +99,7 @@
     get_trigger_completion_variable_maps,
     trigger_to_completion_variable,
 )
+from cylc.flow.task_qualifiers import TASK_QUALIFIERS
 from cylc.flow.run_modes import RunMode
 from cylc.flow.task_trigger import TaskTrigger, Dependency
 from cylc.flow.taskdef import TaskDef
@@ -2266,17 +2267,10 @@ def load_graph(self):
             self.workflow_polling_tasks.update(
                 parser.workflow_state_polling_tasks)
             self._proc_triggers(parser, seq, task_triggers)
-            self.check_outputs(
-                [
-                    task_output
-                    for task_output in parser.task_output_opt
-                    if task_output[0]
-                    in [
-                        task_output.split(':')[0]
-                        for task_output in parser.terminals
-                    ]
-                ]
-            )
+
+            # Checking for undefined outputs for terminal tasks. Tasks with
+            # dependencies are checked in generate_triggers:
+            self.check_outputs(parser.terminals)
 
         self.set_required_outputs(task_output_opt)
 
@@ -2290,22 +2284,19 @@ def load_graph(self):
         for tdef in self.taskdefs.values():
             tdef.tweak_outputs()
 
-    def check_outputs(
-        self, tasks_and_outputs: Iterable[Tuple[str, str]]
-    ) -> None:
+    def check_outputs(self, terminals: Iterable[str]) -> None:
         """Check that task outputs have been registered with tasks.
 
-        Args: tasks_and_outputs: ((task, output), ...)
-
-        Raises: WorkflowConfigError is a user has defined a task with a
-        custom output, but has not registered a custom output.
+        Raises: WorkflowConfigError if a custom output is not defined.
         """
-        for task, output in tasks_and_outputs:
-            registered_outputs = self.cfg['runtime'][task]['outputs']
-            if (
-                not TaskOutputs.is_valid_std_name(output)
-                and output not in registered_outputs
-            ):
+        terminal_outputs = [
+            (a[0].strip("!"), b)
+            for a in (t.split(':') for t in terminals if ":" in t)
+            if (b := a[1].strip("?")) not in TASK_QUALIFIERS
+        ]
+
+        for task, output in terminal_outputs:
+            if output not in self.cfg['runtime'][task]['outputs']:
                 raise WorkflowConfigError(
                     f"Undefined custom output: {task}:{output}"
                 )

tests/integration/reftests/test_pre_initial.py

@@ -94,7 +94,7 @@ async def test_drop(flow, scheduler, reftest):
     }
 
 
-async def test_over_bracketed(flow, scheduler, reftest):
+async def test_over_bracketed(flow, scheduler, reftest, validate):
     """Test nested conditional simplification for pre-initial cycling."""
     wid = flow({
         'scheduling': {
@@ -108,6 +108,7 @@ async def test_over_bracketed(flow, scheduler, reftest):
             },
         },
     })
+    validate(wid)
     schd = scheduler(wid, paused_start=False)
 
     assert await reftest(schd) == {

tests/unit/test_config.py

@@ -15,12 +15,12 @@
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
 import os
-import sys
 from optparse import Values
 from typing import (
     TYPE_CHECKING, Any, Callable, Dict, List, Optional, Tuple, Type)
 import pytest
 import logging
+from textwrap import dedent
 from types import SimpleNamespace
 from contextlib import suppress
 
@@ -47,6 +47,10 @@
 
 from cylc.flow.cycling.iso8601 import ISO8601Point
 
+
+param = pytest.param
+
+
 if TYPE_CHECKING:
     from pathlib import Path
     Fixture = Any
@@ -1703,7 +1707,6 @@ def test_cylc_env_at_parsing(
 
 def test_force_workflow_compat_mode(tmp_path):
     fpath = (tmp_path / 'flow.cylc')
-    from textwrap import dedent
     fpath.write_text(dedent("""
         [scheduler]
             allow implicit tasks = true
@@ -1716,3 +1719,31 @@ def test_force_workflow_compat_mode(tmp_path):
         WorkflowConfig('foo', str(fpath), {})
     # It succeeds with compat mode:
     WorkflowConfig('foo', str(fpath), {}, force_compat_mode=True)
+
+
+@pytest.mark.parametrize(
+    'registered_outputs, tasks_and_outputs, fails',
+    (
+        param([], ['foo:x'], True, id='output-unregistered'),
+        param([], ['foo:x?'], True, id='optional-output-unregistered'),
+        param([], ['foo'], False, id='no-modifier-unregistered'),
+        param(['x'], ['foo:x'], False, id='output-registered'),
+        param([], ['foo:succeed'], False, id='alt-default-ok'),
+        param([], ['foo:failed'], False, id='default-ok'),
+    )
+)
+def test_check_outputs(tmp_path, registered_outputs, tasks_and_outputs, fails):
+    (tmp_path / 'flow.cylc').write_text(dedent("""
+        [scheduler]
+            allow implicit tasks = true
+        [scheduling]
+            [[graph]]
+                R1 = foo
+    """))
+    cfg = WorkflowConfig('', tmp_path / 'flow.cylc', '')
+    cfg.cfg['runtime']['foo']['outputs'] = registered_outputs
+    if fails:
+        with pytest.raises(WorkflowConfigError, match='Undefined custom output'):
+            cfg.check_outputs(tasks_and_outputs)
+    else:
+        assert cfg.check_outputs(tasks_and_outputs) is None