Implement scheduler command method arg validation.

cylc/flow/command_validation.py

@@ -0,0 +1,77 @@
+# THIS FILE IS PART OF THE CYLC WORKFLOW ENGINE.
+# Copyright (C) NIWA & British Crown (Met Office) & Contributors.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+"""Cylc command argument validation funcions."""
+
+
+from typing import (
+    Callable,
+    List,
+    Optional,
+)
+
+from cylc.flow.exceptions import InputError
+
+
+def print_command_response(multi_results):
+    """Print server mutation response, for the CLI.
+
+    """
+    for multi_result in multi_results:
+        for _cmd, results in multi_result.items():
+            for result in results.values():
+                for wf_res in result:
+                    wf_id = wf_res["id"]
+                    response = wf_res["response"]
+                    if not response[0]:
+                        # Validation failure
+                        print(
+                            f"{wf_id}: command validation failed:\n"
+                            f"  {response[1]}"
+                        )
+                    else:
+                        print(f"{wf_id}: command queued (ID {response[1]})")
+
+
+def validate(func: Callable):
+    """Decorate scheduler commands with a callable .validate attribute.
+
+    """
+    # TODO: properly handle "Callable has no attribute validate"?
+    func.validate = globals()[  # type: ignore
+        func.__name__.replace("command", "validate")
+    ]
+    return func
+
+
+def validate_set(
+    tasks: List[str],
+    flow: List[str],
+    outputs: Optional[List[str]] = None,
+    prerequisites: Optional[List[str]] = None,
+    flow_wait: bool = False,
+    flow_descr: Optional[str] = None
+) -> None:
+    """Validate args of the scheduler "command_set" method.
+
+    Raise InputError if validation fails.
+
+    """
+    # TODO: validate the actual args
+
+    # For initial testing...
+    if False:
+        raise InputError(" ...args wrong innit...")

cylc/flow/network/multi.py

@@ -107,4 +107,4 @@ def _report_single(report, workflow, result):
 
 
 def _report(_):
-    print('Command submitted; the scheduler will log any problems.')
+    pass

cylc/flow/network/resolvers.py

@@ -42,6 +42,7 @@
     EDGES, FAMILY_PROXIES, TASK_PROXIES, WORKFLOW,
     DELTA_ADDED, create_delta_store
 )
+from cylc.flow.exceptions import InputError
 from cylc.flow.id import Tokens
 from cylc.flow.network.schema import (
     DEF_TYPES,
@@ -740,10 +741,19 @@ async def _mutation_mapper(
             return method(**kwargs)
 
         try:
-            self.schd.get_command_method(command)
+            meth = self.schd.get_command_method(command)
         except AttributeError:
             raise ValueError(f"Command '{command}' not found")
 
+        # If meth has a command validation function, call it.
+        try:
+            # TODO: properly handle "Callable has no attribute validate"?
+            meth.validate(**kwargs)  # type: ignore
+        except AttributeError:
+            LOG.debug(f"No command validation for {command}")
+        except InputError as exc:
+            return (False, str(exc))
+
         # Queue the command to the scheduler, with a unique command ID
         cmd_uuid = str(uuid4())
         LOG.info(f"{log1} ID={cmd_uuid}\n{log2}")

cylc/flow/scheduler.py

@@ -53,6 +53,7 @@
 )
 from cylc.flow.broadcast_mgr import BroadcastMgr
 from cylc.flow.cfgspec.glbl_cfg import glbl_cfg
+from cylc.flow import command_validation
 from cylc.flow.config import WorkflowConfig
 from cylc.flow.data_store_mgr import DataStoreMgr
 from cylc.flow.id import Tokens
@@ -2146,6 +2147,7 @@ def command_force_trigger_tasks(
         return self.pool.force_trigger_tasks(
             tasks, flow, flow_wait, flow_descr)
 
+    @command_validation.validate
     def command_set(
         self,
         tasks: List[str],

cylc/flow/scripts/set.py

@@ -91,6 +91,7 @@
 from functools import partial
 from typing import TYPE_CHECKING, List, Optional
 
+from cylc.flow.command_validation import print_command_response
 from cylc.flow.exceptions import InputError
 from cylc.flow.network.client_factory import get_client
 from cylc.flow.network.multi import call_multi
@@ -411,14 +412,15 @@ async def run(
         }
     }
 
-    await pclient.async_request('graphql', mutation_kwargs)
+    return await pclient.async_request('graphql', mutation_kwargs)
 
 
 @cli_function(get_option_parser)
 def main(parser: COP, options: 'Values', *ids) -> None:
+
     validate_opts(options.outputs, options.prerequisites)
     validate_flow_opts(options)
-    call_multi(
-        partial(run, options),
-        *ids,
+
+    print_command_response(
+        call_multi(partial(run, options), *ids)
     )