[Python-checkins] bpo-41586: Attempt to make the pipesize tests more robust. (GH-22839)
gpshead
webhook-mailer at python.org
Tue Oct 20 20:37:34 EDT 2020
https://github.com/python/cpython/commit/786addd9d07b6c712b8ea9ee06e1f9f41c1b67a1
commit: 786addd9d07b6c712b8ea9ee06e1f9f41c1b67a1
branch: master
author: Gregory P. Smith <greg at krypto.org>
committer: gpshead <greg at krypto.org>
date: 2020-10-20T17:37:20-07:00
summary:
bpo-41586: Attempt to make the pipesize tests more robust. (GH-22839)
Several buildbots are failing on these, likely due to an inability to
set the pipe size to the desired test value.
files:
M Lib/test/test_fcntl.py
M Lib/test/test_subprocess.py
diff --git a/Lib/test/test_fcntl.py b/Lib/test/test_fcntl.py
index 8d6e9ff788454..83ee7a5b0d012 100644
--- a/Lib/test/test_fcntl.py
+++ b/Lib/test/test_fcntl.py
@@ -190,17 +190,24 @@ def test_fcntl_f_getpath(self):
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 8b576c036ef0d..9fc4434649dbc 100644
--- a/Lib/test/test_subprocess.py
+++ b/Lib/test/test_subprocess.py
@@ -666,45 +666,66 @@ def test_stdin_devnull(self):
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()
More information about the Python-checkins
mailing list