traverseda
/
cpython
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

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.
This commit is contained in:
Gregory P. Smith 2020-10-20 17:37:20 -07:00 committed by GitHub
parent 7cdf30fff3
commit 786addd9d0
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 73 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

25
Lib/test/test_fcntl.py
View File

@ -190,17 +190,24 @@ class TestFcntl(unittest.TestCase):
res = fcntl.fcntl(self.f.fileno(), fcntl.F_GETPATH, bytes(len(expected))) res = fcntl.fcntl(self.f.fileno(), fcntl.F_GETPATH, bytes(len(expected)))
self.assertEqual(expected, res) self.assertEqual(expected, res)
@unittest.skipIf(not (hasattr(fcntl, "F_SETPIPE_SZ") and hasattr(fcntl, "F_GETPIPE_SZ")), @unittest.skipUnless(
"F_SETPIPE_SZ and F_GETPIPE_SZ are not available on all unix platforms.") 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): def test_fcntl_f_pipesize(self):
test_pipe_r, test_pipe_w = os.pipe() test_pipe_r, test_pipe_w = os.pipe()
# Get the default pipesize with F_GETPIPE_SZ try:
pipesize_default = fcntl.fcntl(test_pipe_w, fcntl.F_GETPIPE_SZ) # Get the default pipesize with F_GETPIPE_SZ
# Multiply the default with 2 to get a new value. pipesize_default = fcntl.fcntl(test_pipe_w, fcntl.F_GETPIPE_SZ)
fcntl.fcntl(test_pipe_w, fcntl.F_SETPIPE_SZ, pipesize_default * 2) pipesize = pipesize_default // 2 # A new value to detect change.
self.assertEqual(fcntl.fcntl(test_pipe_w, fcntl.F_GETPIPE_SZ), pipesize_default * 2) if pipesize < 512: # the POSIX minimum
os.close(test_pipe_r) raise unittest.SkitTest(
os.close(test_pipe_w) '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(): def test_main():

93
Lib/test/test_subprocess.py
View File

@ -666,45 +666,66 @@ class ProcessTestCase(BaseTestCase):
p.wait() p.wait()
self.assertEqual(p.stdin, None) 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): def test_pipesizes(self):
# stdin redirection test_pipe_r, test_pipe_w = os.pipe()
pipesize = 16 * 1024 try:
p = subprocess.Popen([sys.executable, "-c", # Get the default pipesize with F_GETPIPE_SZ
'import sys; sys.stdin.read(); sys.stdout.write("out"); sys.stderr.write("error!")'], pipesize_default = fcntl.fcntl(test_pipe_w, fcntl.F_GETPIPE_SZ)
stdin=subprocess.PIPE, finally:
stdout=subprocess.PIPE, os.close(test_pipe_r)
stderr=subprocess.PIPE, os.close(test_pipe_w)
pipesize=pipesize) pipesize = pipesize_default // 2
# We only assert pipe size has changed on platforms that support it. if pipesize < 512: # the POSIX minimum
if sys.platform != "win32" and hasattr(fcntl, "F_GETPIPE_SZ"): raise unittest.SkitTest(
for fifo in [p.stdin, p.stdout, p.stderr]: 'default pipesize too small to perform test.')
self.assertEqual(fcntl.fcntl(fifo.fileno(), fcntl.F_GETPIPE_SZ), pipesize) p = subprocess.Popen(
# Windows pipe size can be acquired with the GetNamedPipeInfoFunction [sys.executable, "-c",
# https://docs.microsoft.com/en-us/windows/win32/api/namedpipeapi/nf-namedpipeapi-getnamedpipeinfo 'import sys; sys.stdin.read(); sys.stdout.write("out"); '
# However, this function is not yet in _winapi. 'sys.stderr.write("error!")'],
p.stdin.write(b"pear") stdin=subprocess.PIPE, stdout=subprocess.PIPE,
p.stdin.close() stderr=subprocess.PIPE, pipesize=pipesize)
p.wait() try:
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"):
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]: for fifo in [p.stdin, p.stdout, p.stderr]:
self.assertEqual( self.assertEqual(
fcntl.fcntl(fifo.fileno(), fcntl.F_GETPIPE_SZ), default_pipesize) fcntl.fcntl(fifo.fileno(), fcntl.F_GETPIPE_SZ),
# On other platforms we cannot test the pipe size (yet). But above code pipesize)
# using pipesize=-1 should not crash. # Windows pipe size can be acquired via GetNamedPipeInfoFunction
p.stdin.close() # https://docs.microsoft.com/en-us/windows/win32/api/namedpipeapi/nf-namedpipeapi-getnamedpipeinfo
p.wait() # 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)
try:
fp_r, fp_w = os.pipe()
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): def test_env(self):
newenv = os.environ.copy() newenv = os.environ.copy()