Merge pull request #6845 from mvdbeek/fix_slurm_cli_on_secure_shell

[18.05] Fix slurm cli on secure shell

lib/galaxy/jobs/runners/util/cli/shell/local.py

@@ -44,7 +44,7 @@ class LocalShell(BaseShellExec):
 
     def execute(self, cmd, persist=False, timeout=DEFAULT_TIMEOUT, timeout_check_interval=DEFAULT_TIMEOUT_CHECK_INTERVAL, **kwds):
         outf = TemporaryFile()
-        p = Popen(cmd, shell=True, stdin=None, stdout=outf, stderr=PIPE)
+        p = Popen(cmd, stdin=None, stdout=outf, stderr=PIPE)
         # poll until timeout
 
         for i in range(int(timeout / timeout_check_interval)):

lib/galaxy/jobs/runners/util/cli/shell/rsh.py

@@ -15,20 +15,23 @@ __all__ = ('RemoteShell', 'SecureShell', 'GlobusSecureShell', 'ParamikoShell')
 
 class RemoteShell(LocalShell):
 
-    def __init__(self, rsh='rsh', rcp='rcp', hostname='localhost', username=None, **kwargs):
+    def __init__(self, rsh='rsh', rcp='rcp', hostname='localhost', username=None, options=None, **kwargs):
         super(RemoteShell, self).__init__(**kwargs)
         self.rsh = rsh
         self.rcp = rcp
         self.hostname = hostname
         self.username = username
+        self.options = options
         self.sessions = {}
 
     def execute(self, cmd, persist=False, timeout=60):
         # TODO: implement persistence
-        if self.username is None:
-            fullcmd = '%s %s %s' % (self.rsh, self.hostname, cmd)
-        else:
-            fullcmd = '%s -l %s %s %s' % (self.rsh, self.username, self.hostname, cmd)
+        fullcmd = [self.rsh]
+        if self.options:
+            fullcmd.extend(self.options)
+        if self.username:
+            fullcmd.extend(["-l", self.username])
+        fullcmd.extend([self.hostname, cmd])
         return super(RemoteShell, self).execute(fullcmd, persist, timeout)
 
 
@@ -37,15 +40,13 @@ class SecureShell(RemoteShell):
 
     def __init__(self, rsh='ssh', rcp='scp', private_key=None, port=None, strict_host_key_checking=True, **kwargs):
         strict_host_key_checking = "yes" if strict_host_key_checking else "no"
-        rsh += " -oStrictHostKeyChecking=%s -oConnectTimeout=60" % strict_host_key_checking
-        rcp += " -oStrictHostKeyChecking=%s -oConnectTimeout=60" % strict_host_key_checking
+        options = ["-o", "StrictHostKeyChecking=%s" % strict_host_key_checking]
+        options.extend(["-o", "ConnectTimeout=60"])
         if private_key:
-            rsh += " -i %s" % private_key
-            rcp += " -i %s" % private_key
+            options.extend(['-i', private_key])
         if port:
-            rsh += " -p %s" % port
-            rcp += " -p %s" % port
-        super(SecureShell, self).__init__(rsh=rsh, rcp=rcp, **kwargs)
+            options.extend(['-p', str(port)])
+        super(SecureShell, self).__init__(rsh=rsh, rcp=rcp, options=options, **kwargs)
 
 
 class ParamikoShell(object):

test/functional/tools/job_environment_default.xml

@@ -8,6 +8,7 @@
       echo `pwd` > '$pwd';
       echo "\$HOME" > '$home';
       echo "\$TMP"  > '$tmp';
+      echo "\$SOME_ENV_VAR" > '$some_env_var';
     ]]></command>
     <inputs>
     </inputs>
@@ -17,6 +18,7 @@
         <data name="pwd" format="txt" label="pwd" />
         <data name="home" format="txt" label="home" />
         <data name="tmp" format="txt" label="tmp" />
+        <data name="some_env_var" format="txt" label="env_var" />
     </outputs>
     <help>
     </help>

test/functional/tools/job_environment_default_legacy.xml

@@ -8,6 +8,7 @@
       echo `pwd` > '$pwd';
       echo "\$HOME" > '$home';
       echo "\$TMP"  > '$tmp';
+      echo "\$SOME_ENV_VAR" > '$some_env_var';
     ]]></command>
     <inputs>
     </inputs>
@@ -17,6 +18,7 @@
         <data name="pwd" format="txt" label="pwd" />
         <data name="home" format="txt" label="home" />
         <data name="tmp" format="txt" label="tmp" />
+        <data name="some_env_var" format="txt" label="env_var" />
     </outputs>
     <help>
     </help>

test/functional/tools/job_environment_explicit_shared_home.xml

@@ -8,6 +8,7 @@
       echo `pwd` > '$pwd';
       echo "\$HOME" > '$home';
       echo "\$TMP"  > '$tmp';
+      echo "\$SOME_ENV_VAR" > '$some_env_var';
     ]]></command>
     <inputs>
     </inputs>
@@ -17,6 +18,7 @@
         <data name="pwd" format="txt" label="pwd" />
         <data name="home" format="txt" label="home" />
         <data name="tmp" format="txt" label="tmp" />
+        <data name="some_env_var" format="txt" label="env_var" />
     </outputs>
     <help>
     </help>

test/integration/test_cli_runners.py

+"""Integration tests for the CLI shell plugins and runners."""
+import collections
+import os
+import string
+import subprocess
+import tempfile
+import unittest
+
+from Crypto.PublicKey import RSA
+
+from base import integration_util  # noqa: I100,I202
+from base.populators import skip_without_tool
+from .test_job_environments import BaseJobEnvironmentIntegrationTestCase  # noqa: I201
+
+
+def generate_keys():
+    key = RSA.generate(2048)
+    return (key.export_key(), key.publickey().export_key(format='OpenSSH'))
+
+
+RemoteConnection = collections.namedtuple('remote_connection', ['hostname', 'username', 'password', 'port', 'private_key', 'public_key'])
+
+
+@integration_util.skip_unless_docker()
+def start_ssh_docker(container_name, jobs_directory, port=10022, image='agaveapi/slurm'):
+    private_key, public_key = generate_keys()
+    with tempfile.NamedTemporaryFile(delete=False) as f:
+        f.write(private_key)
+        private_key_file = f.name
+    with tempfile.NamedTemporaryFile(delete=False) as f:
+        f.write(public_key)
+        public_key_file = f.name
+    START_SLURM_DOCKER = ['docker',
+                          'run',
+                          '-h',
+                          'localhost',
+                          '-p',
+                          '{port}:22'.format(port=port),
+                          '-d',
+                          '--name',
+                          container_name,
+                          '--rm',
+                          '-v',
+                          "{jobs_directory}:{jobs_directory}".format(jobs_directory=jobs_directory),
+                          "-v",
+                          "{public_key_file}:/home/testuser/.ssh/authorized_keys".format(public_key_file=public_key_file),
+                          '--ulimit',
+                          'nofile=2048:2048',
+                          image]
+    subprocess.check_call(START_SLURM_DOCKER)
+    return RemoteConnection('localhost', 'testuser', 'testuser', port, private_key_file, public_key_file)
+
+
+def stop_ssh_docker(container_name, remote_connection):
+    subprocess.check_call(['docker', 'rm', '-f', container_name])
+    os.remove(remote_connection.private_key)
+    os.remove(remote_connection.public_key)
+
+
+def cli_job_config(remote_connection, shell_plugin='ParamikoShell', job_plugin='Slurm'):
+    job_conf_template = string.Template("""<job_conf>
+    <plugins>
+        <plugin id="cli" type="runner" load="galaxy.jobs.runners.cli:ShellJobRunner" workers="1"/>
+    </plugins>
+    <destinations default="ssh_slurm">
+        <destination id="ssh_slurm" runner="cli">
+            <param id="shell_plugin">$shell_plugin</param>
+            <param id="job_plugin">$job_plugin</param>
+            <param id="shell_username">$username</param>
+            <param id="shell_private_key">$private_key</param>
+            <param id="shell_hostname">$hostname</param>
+            <param id="shell_port">$port</param>
+            <param id="embed_metadata_in_job">False</param>
+            <env id="SOME_ENV_VAR">42</env>
+        </destination>
+    </destinations>
+</job_conf>
+""")
+    job_conf_str = job_conf_template.substitute(shell_plugin=shell_plugin,
+                                                job_plugin=job_plugin,
+                                                **remote_connection._asdict())
+    with tempfile.NamedTemporaryFile(suffix="_slurm_integration_job_conf", delete=False) as job_conf:
+        job_conf.write(job_conf_str)
+    return job_conf.name
+
+
+class BaseCliIntegrationTestCase(BaseJobEnvironmentIntegrationTestCase):
+
+    @classmethod
+    def setUpClass(cls):
+        if cls is BaseCliIntegrationTestCase:
+            raise unittest.SkipTest("Base class")
+        cls.container_name = "%s_container" % cls.__name__
+        cls.jobs_directory = tempfile.mkdtemp()
+        cls.remote_connection = start_ssh_docker(container_name=cls.container_name,
+                                                 jobs_directory=cls.jobs_directory,
+                                                 image=cls.image)
+        super(BaseCliIntegrationTestCase, cls).setUpClass()
+
+    @classmethod
+    def tearDownClass(cls):
+        stop_ssh_docker(cls.container_name, cls.remote_connection)
+        super(BaseCliIntegrationTestCase, cls).tearDownClass()
+
+    @classmethod
+    def handle_galaxy_config_kwds(cls, config, ):
+        config["jobs_directory"] = cls.jobs_directory
+        config["file_path"] = cls.jobs_directory
+        config["job_config_file"] = cli_job_config(remote_connection=cls.remote_connection,
+                                                   shell_plugin=cls.shell_plugin,
+                                                   job_plugin=cls.job_plugin)
+
+    @skip_without_tool("job_environment_default")
+    def test_running_cli_job(self):
+        job_env = self._run_and_get_environment_properties()
+        assert job_env.some_env == '42'
+
+
+class TorqueSetup(object):
+    job_plugin = 'Torque'
+    image = 'mvdbeek/galaxy-integration-docker-images:torque_latest'
+
+
+class SlurmSetup(object):
+    job_plugin = 'Slurm'
+    image = 'mvdbeek/galaxy-integration-docker-images:slurm_latest'
+
+
+class ParamikoShell(object):
+    shell_plugin = 'ParamikoShell'
+
+
+class SecureShell(object):
+    shell_plugin = 'SecureShell'
+
+
+class ParamikoCliSlurmIntegrationTestCase(SlurmSetup, ParamikoShell, BaseCliIntegrationTestCase):
+    pass
+
+
+class ShellJobCliSlurmIntegrationTestCase(SlurmSetup, SecureShell, BaseCliIntegrationTestCase):
+    pass
+
+
+class ParamikoCliTorqueIntegrationTestCase(TorqueSetup, ParamikoShell, BaseCliIntegrationTestCase):
+    pass
+
+
+class ShellJobCliTorqueIntegrationTestCase(TorqueSetup, SecureShell, BaseCliIntegrationTestCase):
+    pass

test/integration/test_job_environments.py

@@ -21,6 +21,7 @@ JobEnviromentProperties = collections.namedtuple("JobEnvironmentProperties", [
     "pwd",
     "home",
     "tmp",
+    "some_env",
 ])
 
 
@@ -38,8 +39,8 @@ class RunsEnvironmentJobs:
         pwd = self.dataset_populator.get_history_dataset_content(history_id, hid=3).strip()
         home = self.dataset_populator.get_history_dataset_content(history_id, hid=4).strip()
         tmp = self.dataset_populator.get_history_dataset_content(history_id, hid=5).strip()
-
-        return JobEnviromentProperties(user_id, group_id, pwd, home, tmp)
+        some_env = self.dataset_populator.get_history_dataset_content(history_id, hid=6).strip()
+        return JobEnviromentProperties(user_id, group_id, pwd, home, tmp, some_env)
 
 
 class BaseJobEnvironmentIntegrationTestCase(integration_util.IntegrationTestCase, RunsEnvironmentJobs):