diff options
| -rw-r--r-- | Lib/test/test_fcntl.py | 25 | ||||
| -rw-r--r-- | Lib/test/test_subprocess.py | 89 |
2 files changed, 71 insertions, 43 deletions
diff --git a/Lib/test/test_fcntl.py b/Lib/test/test_fcntl.py index 8d6e9ff..83ee7a5 100644 --- a/Lib/test/test_fcntl.py +++ b/Lib/test/test_fcntl.py @@ -190,17 +190,24 @@ class TestFcntl(unittest.TestCase): res = fcntl.fcntl(self.f.fileno(), fcntl.F_GETPATH, bytes(len(expected))) self.assertEqual(expected, res) - @unittest.skipIf(not (hasattr(fcntl, "F_SETPIPE_SZ") and hasattr(fcntl, "F_GETPIPE_SZ")), - "F_SETPIPE_SZ and F_GETPIPE_SZ are not available on all unix platforms.") + @unittest.skipUnless( + hasattr(fcntl, "F_SETPIPE_SZ") and hasattr(fcntl, "F_GETPIPE_SZ"), + "F_SETPIPE_SZ and F_GETPIPE_SZ are not available on all platforms.") def test_fcntl_f_pipesize(self): test_pipe_r, test_pipe_w = os.pipe() - # Get the default pipesize with F_GETPIPE_SZ - pipesize_default = fcntl.fcntl(test_pipe_w, fcntl.F_GETPIPE_SZ) - # Multiply the default with 2 to get a new value. - fcntl.fcntl(test_pipe_w, fcntl.F_SETPIPE_SZ, pipesize_default * 2) - self.assertEqual(fcntl.fcntl(test_pipe_w, fcntl.F_GETPIPE_SZ), pipesize_default * 2) - os.close(test_pipe_r) - os.close(test_pipe_w) + try: + # Get the default pipesize with F_GETPIPE_SZ + pipesize_default = fcntl.fcntl(test_pipe_w, fcntl.F_GETPIPE_SZ) + pipesize = pipesize_default // 2 # A new value to detect change. + if pipesize < 512: # the POSIX minimum + raise unittest.SkitTest( + 'default pipesize too small to perform test.') + fcntl.fcntl(test_pipe_w, fcntl.F_SETPIPE_SZ, pipesize) + self.assertEqual(fcntl.fcntl(test_pipe_w, fcntl.F_GETPIPE_SZ), + pipesize) + finally: + os.close(test_pipe_r) + os.close(test_pipe_w) def test_main(): diff --git a/Lib/test/test_subprocess.py b/Lib/test/test_subprocess.py index 8b576c0..9fc4434 100644 --- a/Lib/test/test_subprocess.py +++ b/Lib/test/test_subprocess.py @@ -666,45 +666,66 @@ class ProcessTestCase(BaseTestCase): p.wait() self.assertEqual(p.stdin, None) + @unittest.skipUnless(fcntl and hasattr(fcntl, 'F_GETPIPE_SZ'), + 'fcntl.F_GETPIPE_SZ required for test.') def test_pipesizes(self): - # stdin redirection - pipesize = 16 * 1024 - p = subprocess.Popen([sys.executable, "-c", - 'import sys; sys.stdin.read(); sys.stdout.write("out"); sys.stderr.write("error!")'], - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - pipesize=pipesize) - # We only assert pipe size has changed on platforms that support it. - if sys.platform != "win32" and hasattr(fcntl, "F_GETPIPE_SZ"): + test_pipe_r, test_pipe_w = os.pipe() + try: + # Get the default pipesize with F_GETPIPE_SZ + pipesize_default = fcntl.fcntl(test_pipe_w, fcntl.F_GETPIPE_SZ) + finally: + os.close(test_pipe_r) + os.close(test_pipe_w) + pipesize = pipesize_default // 2 + if pipesize < 512: # the POSIX minimum + raise unittest.SkitTest( + 'default pipesize too small to perform test.') + p = subprocess.Popen( + [sys.executable, "-c", + 'import sys; sys.stdin.read(); sys.stdout.write("out"); ' + 'sys.stderr.write("error!")'], + stdin=subprocess.PIPE, stdout=subprocess.PIPE, + stderr=subprocess.PIPE, pipesize=pipesize) + try: for fifo in [p.stdin, p.stdout, p.stderr]: - self.assertEqual(fcntl.fcntl(fifo.fileno(), fcntl.F_GETPIPE_SZ), pipesize) - # Windows pipe size can be acquired with the GetNamedPipeInfoFunction - # https://docs.microsoft.com/en-us/windows/win32/api/namedpipeapi/nf-namedpipeapi-getnamedpipeinfo - # However, this function is not yet in _winapi. - p.stdin.write(b"pear") - p.stdin.close() - p.wait() + self.assertEqual( + fcntl.fcntl(fifo.fileno(), fcntl.F_GETPIPE_SZ), + pipesize) + # Windows pipe size can be acquired via GetNamedPipeInfoFunction + # https://docs.microsoft.com/en-us/windows/win32/api/namedpipeapi/nf-namedpipeapi-getnamedpipeinfo + # However, this function is not yet in _winapi. + p.stdin.write(b"pear") + p.stdin.close() + finally: + p.kill() + p.wait() + @unittest.skipUnless(fcntl and hasattr(fcntl, 'F_GETPIPE_SZ'), + 'fcntl.F_GETPIPE_SZ required for test.') def test_pipesize_default(self): - p = subprocess.Popen([sys.executable, "-c", - 'import sys; sys.stdin.read(); sys.stdout.write("out");' - ' sys.stderr.write("error!")'], - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - pipesize=-1) - # UNIX tests using fcntl - if sys.platform != "win32" and hasattr(fcntl, "F_GETPIPE_SZ"): + p = subprocess.Popen( + [sys.executable, "-c", + 'import sys; sys.stdin.read(); sys.stdout.write("out"); ' + 'sys.stderr.write("error!")'], + stdin=subprocess.PIPE, stdout=subprocess.PIPE, + stderr=subprocess.PIPE, pipesize=-1) + try: fp_r, fp_w = os.pipe() - default_pipesize = fcntl.fcntl(fp_w, fcntl.F_GETPIPE_SZ) - for fifo in [p.stdin, p.stdout, p.stderr]: - self.assertEqual( - fcntl.fcntl(fifo.fileno(), fcntl.F_GETPIPE_SZ), default_pipesize) - # On other platforms we cannot test the pipe size (yet). But above code - # using pipesize=-1 should not crash. - p.stdin.close() - p.wait() + try: + default_pipesize = fcntl.fcntl(fp_w, fcntl.F_GETPIPE_SZ) + for fifo in [p.stdin, p.stdout, p.stderr]: + self.assertEqual( + fcntl.fcntl(fifo.fileno(), fcntl.F_GETPIPE_SZ), + default_pipesize) + finally: + os.close(fp_r) + os.close(fp_w) + # On other platforms we cannot test the pipe size (yet). But above + # code using pipesize=-1 should not crash. + p.stdin.close() + finally: + p.kill() + p.wait() def test_env(self): newenv = os.environ.copy() |