services: validate-reinstall

* Closes #526
* Add basic support for the `cylc vr` command.

Author: oliver-sanders

cylc/uiserver/resolvers.py

@@ -216,31 +216,34 @@ class Services:
     # log file stream lag
     CAT_LOG_SLEEP = 1
 
+    #  command timeout for commands which start schedulers
+    START_TIMEOUT = 120
+
     @staticmethod
     def _error(message: Union[Exception, str]):
         """Format error case response."""
-        return [
+        return (
             False,
             str(message)
-        ]
+        )
 
     @staticmethod
     def _return(message: str):
         """Format success case response."""
-        return [
+        return (
             True,
             message
-        ]
+        )
 
     @classmethod
     async def clean(
         cls,
+        workflows_mgr: 'WorkflowsManager',
         workflows: Iterable['Tokens'],
         args: dict,
-        workflows_mgr: 'WorkflowsManager',
         executor: 'Executor',
         log: 'Logger'
-    ):
+    ) -> tuple[bool, str]:
         """Calls `cylc clean`"""
         # Convert Schema options → cylc.flow.workflow_files.init_clean opts:
         opts = _schema_opts_to_api_opts(args, schema=CleanOptions)
@@ -273,25 +276,50 @@ async def scan(
         cls,
         args: dict,
         workflows_mgr: 'WorkflowsManager',
-    ):
+    ) -> tuple[bool, str]:
         await workflows_mgr.scan()
         return cls._return("Scan requested")
 
     @classmethod
-    async def play(
+    async def run_command(
         cls,
+        command: Iterable[str],
         workflows: Iterable[Tokens],
         args: Dict[str, Any],
-        workflows_mgr: 'WorkflowsManager',
         log: 'Logger',
-    ) -> List[Union[bool, str]]:
-        """Calls `cylc play`."""
+        timeout: int,
+        success_msg: str = 'Command succeeded',
+        fail_msg: str = 'Command failed',
+    ) -> tuple[bool, str]:
+        """Calls the specified Cylc command.
+
+        Args:
+            command:
+                The Cylc subcommand to run.
+                e.g ["play"] or ["cat-log", "-m", "p"].
+            workflows:
+                The workflows to run this command against.
+            args:
+                CLI arguments to be provided to this command.
+                e.g {'color': 'never'} would result in "--color=never".
+            log:
+                The application log, used to record this command invocation.
+            timeout:
+                Length of time to wait for the command to complete.
+            success_msg:
+                Message to be used in the response if the command succeeds.
+            fail_msg:
+                Message to be used in the response if the command fails.
+
+        Returns:
+
+        """
         cylc_version = args.pop('cylc_version', None)
         results: Dict[str, str] = {}
         failed = False
         for tokens in workflows:
             try:
-                cmd = _build_cmd(['cylc', 'play', '--color=never'], args)
+                cmd = _build_cmd(['cylc', *command, '--color=never'], args)
 
                 if tokens['user'] and tokens['user'] != getuser():
                     return cls._error(
@@ -322,10 +350,10 @@ async def play(
                     stderr=PIPE,
                     text=True
                 )
-                ret_code = proc.wait(timeout=120)
+                ret_code = proc.wait(timeout=timeout)
 
                 if ret_code:
-                    msg = f"Command failed ({ret_code}): {cmd_repr}"
+                    msg = f"{fail_msg} ({ret_code}): {cmd_repr}"
                     out, err = proc.communicate()
                     results[wflow] = err.strip() or out.strip() or msg
                     log.error(
@@ -335,26 +363,65 @@ async def play(
                     )
                     failed = True
                 else:
-                    results[wflow] = 'started'
+                    results[wflow] = success_msg
 
             except Exception as exc:  # unexpected error
                 log.exception(exc)
                 return cls._error(exc)
 
         if failed:
             if len(results) == 1:
+                # all commands failed
                 return cls._error(results.popitem()[1])
-            # else log each workflow result on separate lines
+
+            # some commands failed
             return cls._error(
+                # log each workflow result on separate lines
                 "\n\n" + "\n\n".join(
                     f"{wflow}: {msg}" for wflow, msg in results.items()
                 )
             )
 
+        # all commands succeeded
+        return cls._return(f'Workflow(s) {success_msg}')
+
+    @classmethod
+    async def play(
+        cls, workflows_mgr: 'WorkflowsManager', *args, **kwargs
+    ) -> tuple[bool, str]:
+        """Calls `cylc play`."""
+        ret = await cls.run_command(
+            ('play',),
+            *args,
+            **kwargs,
+            timeout=cls.START_TIMEOUT,
+            success_msg='started',
+        )
+
+        # trigger a re-scan
+        await workflows_mgr.scan()
+
+        # return results
+        return ret
+
+    @classmethod
+    async def validate_reinstall(
+        cls, workflows_mgr: 'WorkflowsManager', *args, **kwargs
+    ) -> tuple[bool, str]:
+        """Calls `cylc validate-reinstall`."""
+        ret = await cls.run_command(
+            ('validate-reinstall', '--yes'),
+            *args,
+            **kwargs,
+            timeout=cls.START_TIMEOUT,
+            success_msg='reinstalled',
+        )
+
         # trigger a re-scan
         await workflows_mgr.scan()
-        # send a success message
-        return cls._return('Workflow(s) started')
+
+        # return results
+        return ret
 
     @staticmethod
     async def enqueue(stream, queue):
@@ -581,8 +648,7 @@ async def service(
         command: str,
         workflows: Iterable['Tokens'],
         kwargs: Dict[str, Any],
-    ) -> List[Union[bool, str]]:
-
+    ) -> tuple[bool, str]:
         # GraphQL v3 includes all variables that are set, even if set to null.
         kwargs = {
             k: v
@@ -592,19 +658,26 @@ async def service(
 
         if command == 'clean':  # noqa: SIM116
             return await Services.clean(
+                self.workflows_mgr,
                 workflows,
                 kwargs,
-                self.workflows_mgr,
                 log=self.log,
                 executor=self.executor
             )
         elif command == 'play':
             return await Services.play(
+                self.workflows_mgr,
                 workflows,
                 kwargs,
-                self.workflows_mgr,
                 log=self.log
             )
+        elif command == 'validate_reinstall':
+            return await Services.validate_reinstall(
+                self.workflows_mgr,
+                workflows,
+                kwargs,
+                log=self.log,
+            )
         elif command == 'scan':
             return await Services.scan(
                 kwargs,

cylc/uiserver/schema.py

@@ -241,6 +241,44 @@ class Arguments:
     result = GenericScalar()
 
 
+class ValidateReinstall(graphene.Mutation):
+    class Meta:
+        description = sstrip('''
+            Validate, reinstall, then reload or restart as appropriate.
+
+            This command updates a workflow to reflect any new changes made in
+            the workflow source directory since it was installed.
+
+            The workflow will be reinstalled, then either:
+            * Reloaded (if the workflow is running),
+            * or restarted (if it is stopped).
+        ''')
+        # Note we have the "resume" mutation for paused workflows.
+        resolver = partial(mutator, command='validate_reinstall')
+
+    class Arguments:
+        workflows = graphene.List(WorkflowID, required=True)
+        cylc_version = CylcVersion(
+            description=sstrip('''
+                Set the Cylc version that the workflow starts with.
+            ''')
+        )
+        set = graphene.List(  # noqa: A003 (graphql field name)
+            graphene.String,
+            description=sstrip('''
+                Set the value of a Jinja2 template variable in the workflow
+                definition. Values should be valid Python literals so strings
+                must be quoted e.g. `STR="string"`, `INT=43`, `BOOL=True`.
+                This option can be used multiple  times on the command line.
+                NOTE: these settings persist across workflow restarts, but can
+                be set again on the `cylc play` command line if they need to be
+                overridden.
+            ''')
+        )
+
+    result = GenericScalar()
+
+
 class Clean(graphene.Mutation):
     class Meta:
         description = sstrip('''
@@ -894,6 +932,7 @@ class Logs(graphene.ObjectType):
 
 class UISMutations(Mutations):
     play = _mut_field(Play)
+    validate_reinstall = _mut_field(ValidateReinstall)
     clean = _mut_field(Clean)
     scan = _mut_field(Scan)
 

cylc/uiserver/tests/test_resolvers.py

@@ -64,8 +64,8 @@ def test__schema_opts_to_api_opts(schema_opts, schema, expect):
 @pytest.mark.parametrize(
     'func, message, expect',
     [
-        (services._return, 'Hello.', [True, 'Hello.']),
-        (services._error, 'Goodbye.', [False, 'Goodbye.'])
+        (services._return, 'Hello.', (True, 'Hello.')),
+        (services._error, 'Goodbye.', (False, 'Goodbye.'))
     ]
 )
 def test_Services_anciliary_methods(func, message, expect):
@@ -81,31 +81,31 @@ def test_Services_anciliary_methods(func, message, expect):
             [Tokens('wflow1'), Tokens('~murray/wflow2')],
             {},
             {},
-            [True, "Workflow(s) started"],
+            (True, "Workflow(s) started"),
             {},
             id="multiple"
         ),
         pytest.param(
             [Tokens('~feynman/wflow1')],
             {},
             {},
-            [False, "Cannot start workflows for other users."],
+            (False, "Cannot start workflows for other users."),
             {},
             id="other user's wflow"
         ),
         pytest.param(
             [Tokens('wflow1')],
             {'cylc_version': 'top'},
             {'CYLC_VERSION': 'bottom', 'CYLC_ENV_NAME': 'quark'},
-            [True, "Workflow(s) started"],
+            (True, "Workflow(s) started"),
             {'CYLC_VERSION': 'top'},
             id="cylc version overrides env"
         ),
         pytest.param(
             [Tokens('wflow1')],
             {},
             {'CYLC_VERSION': 'charm', 'CYLC_ENV_NAME': 'quark'},
-            [True, "Workflow(s) started"],
+            (True, "Workflow(s) started"),
             {'CYLC_VERSION': 'charm', 'CYLC_ENV_NAME': 'quark'},
             id="cylc env not overriden if no version specified"
         ),
@@ -144,9 +144,9 @@ async def test_play(
     monkeypatch.setattr('cylc.uiserver.resolvers.Popen', mock_popen)
 
     ret = await Services.play(
+        Mock(spec=WorkflowsManager),
         workflows,
         {'some': 'opt', **args},
-        workflows_mgr=Mock(spec=WorkflowsManager),
         log=Mock(),
     )
 
@@ -228,9 +228,9 @@ async def test_play_fail(
     caplog.set_level(logging.ERROR)
 
     status, message = await Services.play(
+        Mock(spec=WorkflowsManager),
         workflows,
         {},
-        workflows_mgr=Mock(spec=WorkflowsManager),
         log=logging.root,
     )
     assert status is False
@@ -252,9 +252,9 @@ def wait(timeout):
     monkeypatch.setattr('cylc.uiserver.resolvers.Popen', mock_popen)
 
     ret = await Services.play(
+        Mock(spec=WorkflowsManager),
         [Tokens('wflow1')],
         {},
-        workflows_mgr=Mock(spec=WorkflowsManager),
         log=Mock(),
     )
     assert ret == [
@@ -401,12 +401,12 @@ def bad_clean(*a, **k):
     caplog.set_level(logging.ERROR)
 
     ret = Services.clean(
+        Mock(spec=WorkflowsManager),
         [Tokens('wflow1')],
         {},
-        workflows_mgr=Mock(spec=WorkflowsManager),
         executor=ThreadPoolExecutor(1),
         log=logging.root,
     )
     err_msg = "CylcError: bad things!!"
-    assert (await ret) == [False, err_msg]
+    assert (await ret) == (False, err_msg)
     assert err_msg in caplog.text