print warning and resolve to absolute path for commands specified with relative path

lib/vsc/utils/py2vs3/py2.py

@@ -133,3 +133,28 @@ def _rmtree(self, path):
             self._rmdir(path)
         except OSError:
             pass
+
+
+def which(cmd, mode=os.F_OK | os.X_OK, path=None):
+    """
+    Return the first absolute path that matches the given command.
+
+    Searches via specified path, or $PATH as fallback.
+
+    Returns absolute path to command if found, None if not.
+    """
+    res = None
+    if os.path.isabs(cmd):
+        if os.access(cmd, cmd):
+            res = cmd
+    else:
+        if path is None:
+            path = os.environ.get('PATH', '')
+
+        for cand_path in path.split(os.pathsep):
+            cmd_path = os.path.join(cand_path, cmd)
+            if os.access(cmd_path, mode):
+                res = cmd_path
+                break
+
+    return res

lib/vsc/utils/py2vs3/py3.py

@@ -32,6 +32,7 @@
 import pickle  # noqa
 from io import StringIO  # noqa
 from shlex import quote  # noqa
+from shutil import which  # noqa
 from tempfile import TemporaryDirectory  # noqa
 from urllib.parse import urlencode, unquote  # noqa
 from urllib.request import HTTPError, HTTPSHandler, Request, build_opener, urlopen  # noqa

lib/vsc/utils/run.py

@@ -74,7 +74,7 @@
 import time
 
 from vsc.utils.fancylogger import getLogger
-from vsc.utils.py2vs3 import ensure_ascii_string, is_py3, is_string
+from vsc.utils.py2vs3 import ensure_ascii_string, is_py3, is_string, which
 
 PROCESS_MODULE_ASYNCPROCESS_PATH = 'vsc.utils.asyncprocess'
 PROCESS_MODULE_SUBPROCESS_PATH = 'subprocess'
@@ -87,6 +87,38 @@
 SHELL = BASH
 
 
+def ensure_cmd_abs_path(cmd):
+    """Make sure that command is specified via an absolute path."""
+    if not cmd:
+        raise ValueError("Empty command specified!")
+    if is_string(cmd):
+        cmd_path = cmd.split(' ')[0]
+    elif isinstance(cmd, (list, tuple,)):
+        cmd_path = cmd[0]
+    else:
+        raise ValueError("Unknown type of command: %s (type %s)" % (cmd, type(cmd)))
+
+    if not os.path.isabs(cmd_path):
+        sys.stderr.write("WARNING: Command to run is specified via relative path: %s" % cmd_path)
+
+        # resolve to absolute path via $PATH
+        cmd_abs_path = which(cmd_path)
+        if cmd_abs_path is None:
+            raise OSError("Command %s not found in $PATH!" % cmd_path)
+
+        # re-assemble command with absolute path
+        if is_string(cmd):
+            cmd = ' '.join([cmd_abs_path] + cmd.split(' ')[1:])
+        elif isinstance(cmd, list):
+            cmd[0] = cmd_abs_path
+        elif isinstance(cmd, tuple):
+            cmd = tuple([cmd_abs_path] + list(cmd[1:]))
+        else:
+            raise ValueError("Unknown type of command: %s (type %s)" % (cmd, type(cmd)))
+
+    return cmd
+
+
 class CmdList(list):
     """Wrapper for 'list' type to be used for constructing a list of options & arguments for a command."""
 
@@ -172,7 +204,7 @@ def __init__(self, cmd=None, **kwargs):
         if not hasattr(self, 'log'):
             self.log = getLogger(self._get_log_name())
 
-        self.cmd = cmd  # actual command
+        self.cmd = ensure_cmd_abs_path(cmd)  # actual command
 
         self._cwd_before_startpath = None
 

setup.py

@@ -44,7 +44,7 @@
 ]
 
 PACKAGE = {
-    'version': '3.4.0',
+    'version': '3.4.1',
     'author': [sdw, jt, ag, kh],
     'maintainer': [sdw, jt, ag, kh],
     # as long as 1.0.0 is not out, vsc-base should still provide vsc.fancylogger

test/py2vs3.py

@@ -30,9 +30,11 @@
 @author: Kenneth Hoste (Ghent University)
 """
 import os
+import stat
 import sys
 
-from vsc.utils.py2vs3 import ensure_ascii_string, is_py_ver, is_py2, is_py3, is_string, pickle, TemporaryDirectory
+from vsc.utils.py2vs3 import ensure_ascii_string, is_py_ver, is_py2, is_py3, is_string, pickle, which
+from vsc.utils.py2vs3 import TemporaryDirectory
 from vsc.install.testing import TestCase
 
 
@@ -143,12 +145,11 @@ def test_ensure_ascii_string(self):
 
     def test_urllib_imports(self):
         """Test importing urllib* stuff from py2vs3."""
-        from vsc.utils.py2vs3 import HTTPError, HTTPSHandler, Request, build_opener, unquote, urlencode, urlopen
+        from vsc.utils.py2vs3 import HTTPError, HTTPSHandler, Request, build_opener, unquote, urlencode, urlopen  # noqa
 
     def test_temporary_directory(self):
         """Test the class TemporaryDirectory."""
         with TemporaryDirectory() as temp_dir:
-            path = temp_dir
             self.assertTrue(os.path.exists(temp_dir), 'Directory created by TemporaryDirectory should work')
             self.assertTrue(os.path.isdir(temp_dir), 'Directory created by TemporaryDirectory should be a directory')
         self.assertFalse(os.path.exists(temp_dir), 'Directory created by TemporaryDirectory should cleanup automagically')
@@ -164,3 +165,27 @@ def test_os_exceptions(self):
             f.write("abc")
 
         self.assertRaises(FileExistsErrorExc, lambda: os.open(afile, os.O_CREAT | os.O_WRONLY | os.O_EXCL))
+
+    def test_which(self):
+        """Test which function."""
+
+        self.assertTrue(which('ls'))
+        self.assertEqual(which('nosuchcommand'), None)
+
+        # create test to test with
+        test_cmd = os.path.join(self.tmpdir, 'test123')
+        with open(test_cmd, 'w') as fp:
+            fp.write("echo 123")
+
+        # missing exec permission
+        self.assertEqual(which('test123'), None)
+        self.assertEqual(which('test123', mode=os.R_OK, path=self.tmpdir), test_cmd)
+
+        os.chmod(test_cmd, stat.S_IRUSR | stat.S_IXUSR)
+
+        # not available via $PATH
+        self.assertEqual(which('test123'), None)
+        self.assertEqual(which('test123', path=self.tmpdir), test_cmd)
+
+        os.environ['PATH'] = self.tmpdir + os.pathsep + os.getenv('PATH')
+        self.assertEqual(which('test123'), test_cmd)

test/run.py

@@ -38,16 +38,16 @@
 import shutil
 
 # Uncomment when debugging, cannot enable permanetnly, messes up tests that toggle debugging
-#logging.basicConfig(level=logging.DEBUG)
+# logging.basicConfig(level=logging.DEBUG)
 
 from vsc.utils.missing import shell_quote
 from vsc.utils.run import (
     CmdList, run, run_simple, asyncloop, run_asyncloop,
     run_timeout, RunTimeout,
     RunQA, RunNoShellQA,
-    async_to_stdout, run_async_to_stdout,
+    async_to_stdout, ensure_cmd_abs_path, run_async_to_stdout,
 )
-from vsc.utils.py2vs3 import StringIO, is_py2, is_py3, is_string
+from vsc.utils.py2vs3 import is_py2, is_py3, is_string
 from vsc.utils.run import RUNRUN_TIMEOUT_OUTPUT, RUNRUN_TIMEOUT_EXITCODE, RUNRUN_QA_MAX_MISS_EXITCODE
 from vsc.install.testing import TestCase
 
@@ -58,14 +58,17 @@
 SCRIPT_NESTED = os.path.join(SCRIPTS_DIR, 'run_nested.sh')
 
 
-TEST_GLOB = ['ls','test/sandbox/testpkg/*']
+TEST_GLOB = ['ls', 'test/sandbox/testpkg/*']
+
 
 class RunQAShort(RunNoShellQA):
     LOOP_MAX_MISS_COUNT = 3  # approx 3 sec
 
+
 class RunLegQAShort(RunQA):
     LOOP_MAX_MISS_COUNT = 3  # approx 3 sec
 
+
 run_qas = RunQAShort.run
 run_legacy_qas = RunLegQAShort.run
 
@@ -355,13 +358,14 @@ def test_qa_list_of_answers(self):
         qa_reg_dict = {
             "Enter a number \(.*\):": ['2', '3', '5', '0'] + ['100'] * 100,
         }
-        ec, output = run_qas([sys.executable, SCRIPT_QA, 'ask_number', '100'], qa_reg=qa_reg_dict)
+        cmd = [sys.executable, SCRIPT_QA, 'ask_number', '100']
+        ec, output = run_qas(cmd, qa_reg=qa_reg_dict)
         self.assertEqual(ec, 0)
         answer_re = re.compile(".*Answer: 10$")
         self.assertTrue(answer_re.match(output), "'%s' matches pattern '%s'" % (output, answer_re.pattern))
 
         # verify type checking on answers
-        self.assertErrorRegex(TypeError, "Invalid type for answer", run_qas, [], qa={'q': 1})
+        self.assertErrorRegex(TypeError, "Invalid type for answer", run_qas, cmd, qa={'q': 1})
 
         # test more questions than answers, both with and without cycling
         qa_reg_dict = {
@@ -465,6 +469,56 @@ def test_cmdlist(self):
         self.assertErrorRegex(ValueError, "Found one or more spaces", CmdList, 'this has spaces', allow_spaces=False)
 
     def test_env(self):
-        ec, output = run(cmd="/usr/bin/env", env = {"MYENVVAR": "something"})
+        ec, output = run(cmd="/usr/bin/env", env={"MYENVVAR": "something"})
         self.assertEqual(ec, 0)
         self.assertTrue('myenvvar=something' in output.lower())
+
+    def test_ensure_cmd_abs_path(self):
+        """Test ensure_cmd_abs_path function."""
+
+        def check_warning(cmd):
+            """Pass cmd to ensure_cmd_abs_path, and check for warning being printed."""
+            if isinstance(cmd, (list, tuple)):
+                cmd_name = cmd[0]
+            else:
+                cmd_name = cmd.split(' ')[0]
+
+            self.mock_stderr(True)
+            self.mock_stdout(True)
+            res = ensure_cmd_abs_path(cmd)
+            stderr, stdout = self.get_stderr(), self.get_stdout()
+            self.mock_stderr(False)
+            self.mock_stdout(False)
+            self.assertFalse(stdout)
+            self.assertEqual(stderr, "WARNING: Command to run is specified via relative path: %s" % cmd_name)
+            return res
+
+        ls = check_warning('ls')
+        self.assertTrue(os.path.isabs(ls) and os.path.exists(ls))
+
+        cmd = check_warning('ls foo')
+        self.assertTrue(cmd.endswith('/ls foo'))
+        cmd_path = cmd.split(' ')[0]
+        self.assertTrue(os.path.isabs(cmd_path) and os.path.exists(cmd_path))
+
+        for input_cmd in (['ls', 'foo'], ('ls', 'foo')):
+            cmd = check_warning(input_cmd)
+            self.assertTrue(isinstance(cmd, type(input_cmd)))
+            self.assertEqual(len(cmd), 2)
+            self.assertTrue(os.path.isabs(cmd[0]) and os.path.exists(cmd[0]))
+            self.assertEqual(cmd[1], 'foo')
+
+        # command is left untouched if it already uses an absolute path
+        for cmd in ('/bin/nosuchcommand', ['/foo'], ('/bin/ls', 'foo')):
+            self.assertEqual(ensure_cmd_abs_path(cmd), cmd)
+
+        # check handling of faulty input
+        error_msg = r"Command nosuchcommand not found in \$PATH!"
+        self.assertErrorRegex(OSError, error_msg, ensure_cmd_abs_path, 'nosuchcommand')
+
+        error_msg = "Empty command specified!"
+        self.assertErrorRegex(ValueError, error_msg, ensure_cmd_abs_path, [])
+        self.assertErrorRegex(ValueError, error_msg, ensure_cmd_abs_path, ())
+
+        error_msg = "Unknown type of command: 1"
+        self.assertErrorRegex(ValueError, error_msg, ensure_cmd_abs_path, 1)