diff --git a/Doc/library/os.rst b/Doc/library/os.rst index a4c5fbb481521e..b3099886889599 100644 --- a/Doc/library/os.rst +++ b/Doc/library/os.rst @@ -32,7 +32,7 @@ Notes on the availability of these functions: objects, and result in an object of the same type, if a path or file name is returned. -* On VxWorks, os.fork, os.execv and os.spawn*p* are not supported. +* On VxWorks, os.popen, os.fork, os.execv and os.spawn*p* are not supported. .. note:: diff --git a/Lib/os.py b/Lib/os.py index b794159f86c33e..05e9c32c5a7117 100644 --- a/Lib/os.py +++ b/Lib/os.py @@ -36,7 +36,7 @@ __all__ = ["altsep", "curdir", "pardir", "sep", "pathsep", "linesep", "defpath", "name", "path", "devnull", "SEEK_SET", "SEEK_CUR", "SEEK_END", "fsencode", "fsdecode", "get_exec_path", "fdopen", - "popen", "extsep"] + "extsep"] def _exists(name): return name in globals() @@ -969,51 +969,55 @@ def spawnlpe(mode, file, *args): __all__.extend(["spawnlp", "spawnlpe"]) - -# Supply os.popen() -def popen(cmd, mode="r", buffering=-1): - if not isinstance(cmd, str): - raise TypeError("invalid cmd type (%s, expected string)" % type(cmd)) - if mode not in ("r", "w"): - raise ValueError("invalid mode %r" % mode) - if buffering == 0 or buffering is None: - raise ValueError("popen() does not support unbuffered streams") - import subprocess, io - if mode == "r": - proc = subprocess.Popen(cmd, - shell=True, - stdout=subprocess.PIPE, - bufsize=buffering) - return _wrap_close(io.TextIOWrapper(proc.stdout), proc) - else: - proc = subprocess.Popen(cmd, - shell=True, - stdin=subprocess.PIPE, - bufsize=buffering) - return _wrap_close(io.TextIOWrapper(proc.stdin), proc) - -# Helper for popen() -- a proxy for a file whose close waits for the process -class _wrap_close: - def __init__(self, stream, proc): - self._stream = stream - self._proc = proc - def close(self): - self._stream.close() - returncode = self._proc.wait() - if returncode == 0: - return None - if name == 'nt': - return returncode +# VxWorks has no user space shell provided. As a result, running +# command in a shell can't be supported. +if sys.platform != 'vxworks': + # Supply os.popen() + def popen(cmd, mode="r", buffering=-1): + if not isinstance(cmd, str): + raise TypeError("invalid cmd type (%s, expected string)" % type(cmd)) + if mode not in ("r", "w"): + raise ValueError("invalid mode %r" % mode) + if buffering == 0 or buffering is None: + raise ValueError("popen() does not support unbuffered streams") + import subprocess, io + if mode == "r": + proc = subprocess.Popen(cmd, + shell=True, + stdout=subprocess.PIPE, + bufsize=buffering) + return _wrap_close(io.TextIOWrapper(proc.stdout), proc) else: - return returncode << 8 # Shift left to match old behavior - def __enter__(self): - return self - def __exit__(self, *args): - self.close() - def __getattr__(self, name): - return getattr(self._stream, name) - def __iter__(self): - return iter(self._stream) + proc = subprocess.Popen(cmd, + shell=True, + stdin=subprocess.PIPE, + bufsize=buffering) + return _wrap_close(io.TextIOWrapper(proc.stdin), proc) + + # Helper for popen() -- a proxy for a file whose close waits for the process + class _wrap_close: + def __init__(self, stream, proc): + self._stream = stream + self._proc = proc + def close(self): + self._stream.close() + returncode = self._proc.wait() + if returncode == 0: + return None + if name == 'nt': + return returncode + else: + return returncode << 8 # Shift left to match old behavior + def __enter__(self): + return self + def __exit__(self, *args): + self.close() + def __getattr__(self, name): + return getattr(self._stream, name) + def __iter__(self): + return iter(self._stream) + + __all__.append("popen") # Supply os.fdopen() def fdopen(fd, *args, **kwargs): diff --git a/Lib/test/test_os.py b/Lib/test/test_os.py index f127cb199bbb5c..7e9b7d54d31bc6 100644 --- a/Lib/test/test_os.py +++ b/Lib/test/test_os.py @@ -987,6 +987,7 @@ def _empty_mapping(self): # Bug 1110478 @unittest.skipUnless(unix_shell and os.path.exists(unix_shell), 'requires a shell') + @unittest.skipUnless(hasattr(os, 'popen'), "needs os.popen()") def test_update2(self): os.environ.clear() os.environ.update(HELLO="World") @@ -996,6 +997,7 @@ def test_update2(self): @unittest.skipUnless(unix_shell and os.path.exists(unix_shell), 'requires a shell') + @unittest.skipUnless(hasattr(os, 'popen'), "needs os.popen()") def test_os_popen_iter(self): with os.popen("%s -c 'echo \"line1\nline2\nline3\"'" % unix_shell) as popen: diff --git a/Lib/test/test_popen.py b/Lib/test/test_popen.py index ab1bc776552ca9..cac2f6177f3257 100644 --- a/Lib/test/test_popen.py +++ b/Lib/test/test_popen.py @@ -7,6 +7,9 @@ from test import support import os, sys +if not hasattr(os, 'popen'): + raise unittest.SkipTest("need os.popen()") + # Test that command-lines get down as we expect. # To do this we execute: # python -c "import sys;print(sys.argv)" {rest_of_commandline} diff --git a/Lib/test/test_posix.py b/Lib/test/test_posix.py index a522717751ac17..401e6c0e2fb5a9 100644 --- a/Lib/test/test_posix.py +++ b/Lib/test/test_posix.py @@ -1040,6 +1040,7 @@ def test_getgrouplist(self): @unittest.skipUnless(hasattr(os, 'getegid'), "test needs os.getegid()") + @unittest.skipUnless(hasattr(os, 'popen'), "test needs os.popen()") def test_getgroups(self): with os.popen('id -G 2>/dev/null') as idg: groups = idg.read().strip() @@ -1929,7 +1930,7 @@ def test_posix_spawnp(self): class TestPosixWeaklinking(unittest.TestCase): # These test cases verify that weak linking support on macOS works # as expected. These cases only test new behaviour introduced by weak linking, - # regular behaviour is tested by the normal test cases. + # regular behaviour is tested by the normal test cases. # # See the section on Weak Linking in Mac/README.txt for more information. def setUp(self): diff --git a/Lib/test/test_select.py b/Lib/test/test_select.py index 458998a62fdf5e..0df04e43aa5390 100644 --- a/Lib/test/test_select.py +++ b/Lib/test/test_select.py @@ -44,6 +44,7 @@ def test_returned_list_identity(self): self.assertIsNot(r, x) self.assertIsNot(w, x) + @unittest.skipUnless(hasattr(os, 'popen'), "need os.popen()") def test_select(self): cmd = 'for i in 0 1 2 3 4 5 6 7 8 9; do echo testing...; sleep 1; done' with os.popen(cmd) as p: diff --git a/Misc/NEWS.d/next/Tests/2020-07-30-18-06-15.bpo-31904.y3d8dk.rst b/Misc/NEWS.d/next/Tests/2020-07-30-18-06-15.bpo-31904.y3d8dk.rst new file mode 100644 index 00000000000000..fa2974963bf019 --- /dev/null +++ b/Misc/NEWS.d/next/Tests/2020-07-30-18-06-15.bpo-31904.y3d8dk.rst @@ -0,0 +1 @@ +Disable os.popen and impacted tests on VxWorks