Issue #8513: os.get_exec_path() supports b'PATH' key and bytes value.

subprocess.Popen() and os._execvpe() support bytes program name. Add
os.supports_bytes_environ flag: True if the native OS type of the environment
is bytes (eg. False on Windows).

4 files changed, 116 insertions, 21 deletions

diff --git a/Lib/os.py b/Lib/os.py
index 13ab18c..8f47137 100644
--- a/Lib/os.py
+++ b/Lib/os.py
@@ -355,7 +355,11 @@ def _execvpe(file, args, env=None):
         return
     last_exc = saved_exc = None
     saved_tb = None
-    for dir in get_exec_path(env):
+    path_list = get_exec_path(env)
+    if name != 'nt':
+        file = fsencode(file)
+        path_list = map(fsencode, path_list)
+    for dir in path_list:
         fullname = path.join(dir, file)
         try:
             exec_func(fullname, *argrest)
@@ -380,7 +384,30 @@ def get_exec_path(env=None):
     """
     if env is None:
         env = environ
-    return env.get('PATH', defpath).split(pathsep)
+
+    try:
+        path_list = env.get('PATH')
+    except TypeError:
+        path_list = None
+
+    if supports_bytes_environ:
+        try:
+            path_listb = env[b'PATH']
+        except (KeyError, TypeError):
+            pass
+        else:
+            if path_list is not None:
+                raise ValueError(
+                    "env cannot contain 'PATH' and b'PATH' keys")
+            path_list = path_listb
+
+        if path_list is not None and isinstance(path_list, bytes):
+            path_list = path_list.decode(sys.getfilesystemencoding(),
+                                         'surrogateescape')
+
+    if path_list is None:
+        path_list = defpath
+    return path_list.split(pathsep)
 
 
 # Change environ to automatically call putenv(), unsetenv if they exist.
@@ -482,9 +509,11 @@ def getenv(key, default=None):
     The optional second argument can specify an alternate default.
     key, default and the result are str."""
     return environ.get(key, default)
-__all__.append("getenv")
 
-if name not in ('os2', 'nt'):
+supports_bytes_environ = name not in ('os2', 'nt')
+__all__.extend(("getenv", "supports_bytes_environ"))
+
+if supports_bytes_environ:
     def _check_bytes(value):
         if not isinstance(value, bytes):
             raise TypeError("bytes expected, not %s" % type(value).__name__)
diff --git a/Lib/subprocess.py b/Lib/subprocess.py
index 14f1f67..adbee0b 100644
--- a/Lib/subprocess.py
+++ b/Lib/subprocess.py
@@ -1096,15 +1096,14 @@ class Popen(object):
                                         for k, v in env.items()]
                         else:
                             env_list = None  # Use execv instead of execve.
+                        executable = os.fsencode(executable)
                         if os.path.dirname(executable):
-                            executable_list = (os.fsencode(executable),)
+                            executable_list = (executable,)
                         else:
                             # This matches the behavior of os._execvpe().
-                            path_list = os.get_exec_path(env)
-                            executable_list = (os.path.join(dir, executable)
-                                               for dir in path_list)
-                            executable_list = tuple(os.fsencode(exe)
-                                                    for exe in executable_list)
+                            executable_list = tuple(
+                                os.path.join(os.fsencode(dir), executable)
+                                for dir in os.get_exec_path(env))
                         self.pid = _posixsubprocess.fork_exec(
                                 args, executable_list,
                                 close_fds, cwd, env_list,
diff --git a/Lib/test/test_os.py b/Lib/test/test_os.py
index 37abe22..6a63717 100644
--- a/Lib/test/test_os.py
+++ b/Lib/test/test_os.py
@@ -370,7 +370,7 @@ class EnvironTests(mapping_tests.BasicTestMappingProtocol):
 
     def setUp(self):
         self.__save = dict(os.environ)
-        if os.name not in ('os2', 'nt'):
+        if os.supports_bytes_environ:
             self.__saveb = dict(os.environb)
         for key, value in self._reference().items():
             os.environ[key] = value
@@ -378,7 +378,7 @@ class EnvironTests(mapping_tests.BasicTestMappingProtocol):
     def tearDown(self):
         os.environ.clear()
         os.environ.update(self.__save)
-        if os.name not in ('os2', 'nt'):
+        if os.supports_bytes_environ:
             os.environb.clear()
             os.environb.update(self.__saveb)
 
@@ -445,7 +445,21 @@ class EnvironTests(mapping_tests.BasicTestMappingProtocol):
         # Supplied PATH environment variable
         self.assertSequenceEqual(test_path, os.get_exec_path(test_env))
 
-    @unittest.skipIf(sys.platform == "win32", "POSIX specific test")
+        if os.supports_bytes_environ:
+            # env cannot contain 'PATH' and b'PATH' keys
+            self.assertRaises(ValueError,
+                os.get_exec_path, {'PATH': '1', b'PATH': b'2'})
+
+            # bytes key and/or value
+            self.assertSequenceEqual(os.get_exec_path({b'PATH': b'abc'}),
+                ['abc'])
+            self.assertSequenceEqual(os.get_exec_path({b'PATH': 'abc'}),
+                ['abc'])
+            self.assertSequenceEqual(os.get_exec_path({'PATH': b'abc'}),
+                ['abc'])
+
+    @unittest.skipUnless(os.supports_bytes_environ,
+                         "os.environb required for this test.")
     def test_environb(self):
         # os.environ -> os.environb
         value = 'euro\u20ac'
@@ -669,22 +683,54 @@ class ExecTests(unittest.TestCase):
 
     @unittest.skipUnless(hasattr(os, '_execvpe'),
                          "No internal os._execvpe function to test.")
-    def test_internal_execvpe(self):
-        program_path = os.sep+'absolutepath'
-        program = 'executable'
-        fullpath = os.path.join(program_path, program)
-        arguments = ['progname', 'arg1', 'arg2']
+    def _test_internal_execvpe(self, test_type):
+        program_path = os.sep + 'absolutepath'
+        if test_type is bytes:
+            program = b'executable'
+            fullpath = os.path.join(os.fsencode(program_path), program)
+            native_fullpath = fullpath
+            arguments = [b'progname', 'arg1', 'arg2']
+        else:
+            program = 'executable'
+            arguments = ['progname', 'arg1', 'arg2']
+            fullpath = os.path.join(program_path, program)
+            if os.name != "nt":
+                native_fullpath = os.fsencode(fullpath)
+            else:
+                native_fullpath = fullpath
         env = {'spam': 'beans'}
 
+        # test os._execvpe() with an absolute path
         with _execvpe_mockup() as calls:
-            self.assertRaises(RuntimeError, os._execvpe, fullpath, arguments)
+            self.assertRaises(RuntimeError,
+                os._execvpe, fullpath, arguments)
             self.assertEqual(len(calls), 1)
             self.assertEqual(calls[0], ('execv', fullpath, (arguments,)))
 
+        # test os._execvpe() with a relative path:
+        # os.get_exec_path() returns defpath
         with _execvpe_mockup(defpath=program_path) as calls:
-            self.assertRaises(OSError, os._execvpe, program, arguments, env=env)
+            self.assertRaises(OSError,
+                os._execvpe, program, arguments, env=env)
+            self.assertEqual(len(calls), 1)
+            self.assertSequenceEqual(calls[0],
+                ('execve', native_fullpath, (arguments, env)))
+
+        # test os._execvpe() with a relative path:
+        # os.get_exec_path() reads the 'PATH' variable
+        with _execvpe_mockup() as calls:
+            env_path = env.copy()
+            env_path['PATH'] = program_path
+            self.assertRaises(OSError,
+                os._execvpe, program, arguments, env=env_path)
             self.assertEqual(len(calls), 1)
-            self.assertEqual(calls[0], ('execve', fullpath, (arguments, env)))
+            self.assertSequenceEqual(calls[0],
+                ('execve', native_fullpath, (arguments, env_path)))
+
+    def test_internal_execvpe_str(self):
+        self._test_internal_execvpe(str)
+        if os.name != "nt":
+            self._test_internal_execvpe(bytes)
 
 
 class Win32ErrorTests(unittest.TestCase):
diff --git a/Lib/test/test_subprocess.py b/Lib/test/test_subprocess.py
index eb96706..96c8ebf 100644
--- a/Lib/test/test_subprocess.py
+++ b/Lib/test/test_subprocess.py
@@ -825,6 +825,27 @@ class POSIXProcessTestCase(BaseTestCase):
             stdout = stdout.rstrip(b'\n\r')
             self.assertEquals(stdout.decode('ascii'), repr(value))
 
+    def test_bytes_program(self):
+        abs_program = os.fsencode(sys.executable)
+        path, program = os.path.split(sys.executable)
+        program = os.fsencode(program)
+
+        # absolute bytes path
+        exitcode = subprocess.call([abs_program, "-c", "pass"])
+        self.assertEquals(exitcode, 0)
+
+        # bytes program, unicode PATH
+        env = os.environ.copy()
+        env["PATH"] = path
+        exitcode = subprocess.call([program, "-c", "pass"], env=env)
+        self.assertEquals(exitcode, 0)
+
+        # bytes program, bytes PATH
+        envb = os.environb.copy()
+        envb[b"PATH"] = os.fsencode(path)
+        exitcode = subprocess.call([program, "-c", "pass"], env=envb)
+        self.assertEquals(exitcode, 0)
+
 
 @unittest.skipUnless(mswindows, "Windows specific tests")
 class Win32ProcessTestCase(BaseTestCase):