mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-09-26 18:29:57 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 10

Issue #12196: Add PIPE_MAX_SIZE to test.support, constant larger than the

Browse source 
underlying OS pipe buffer size.
This commit is contained in:
Charles-François Natali 2011-05-29 16:36:44 +02:00
parent b30eed981f
commit 2d51721832
3 changed files with 13 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

9
Lib/test/support.py
View file

@ -48,7 +48,7 @@ __all__ = [
"threading_cleanup", "reap_children", "cpython_only", "check_impl_detail", "threading_cleanup", "reap_children", "cpython_only", "check_impl_detail",
"get_attribute", "swap_item", "swap_attr", "requires_IEEE_754", "get_attribute", "swap_item", "swap_attr", "requires_IEEE_754",
"TestHandler", "Matcher", "can_symlink", "skip_unless_symlink", "TestHandler", "Matcher", "can_symlink", "skip_unless_symlink",
"import_fresh_module", "requires_zlib" "import_fresh_module", "requires_zlib", "PIPE_MAX_SIZE"
] ]
class Error(Exception): class Error(Exception):
@ -409,6 +409,13 @@ def _is_ipv6_enabled():
IPV6_ENABLED = _is_ipv6_enabled() IPV6_ENABLED = _is_ipv6_enabled()
# A constant likely larger than the underlying OS pipe buffer size.
# Windows limit seems to be around 512B, and most Unix kernels have a 64K pipe
# buffer size: take 1M to be sure.
PIPE_MAX_SIZE = 1024 * 1024
# decorator for skipping tests on non-IEEE 754 platforms # decorator for skipping tests on non-IEEE 754 platforms
requires_IEEE_754 = unittest.skipUnless( requires_IEEE_754 = unittest.skipUnless(
float.__getformat__("double").startswith("IEEE"), float.__getformat__("double").startswith("IEEE"),

2
Lib/test/test_io.py
View file

@ -2683,7 +2683,7 @@ class SignalsTest(unittest.TestCase):
# The buffered IO layer must check for pending signal # The buffered IO layer must check for pending signal
# handlers, which in this case will invoke alarm_interrupt(). # handlers, which in this case will invoke alarm_interrupt().
self.assertRaises(ZeroDivisionError, self.assertRaises(ZeroDivisionError,
wio.write, item * (1024 * 1024)) wio.write, item * (support.PIPE_MAX_SIZE // len(item)))
t.join() t.join()
# We got one byte, get another one and check that it isn't a # We got one byte, get another one and check that it isn't a
# repeat of the first one. # repeat of the first one.

11
Lib/test/test_subprocess.py
View file

@ -489,24 +489,21 @@ class ProcessTestCase(BaseTestCase):
# This test will probably deadlock rather than fail, if # This test will probably deadlock rather than fail, if
# communicate() does not work properly. # communicate() does not work properly.
x, y = os.pipe() x, y = os.pipe()
if mswindows:
pipe_buf = 512
else:
pipe_buf = os.fpathconf(x, "PC_PIPE_BUF")
os.close(x) os.close(x)
os.close(y) os.close(y)
p = subprocess.Popen([sys.executable, "-c", p = subprocess.Popen([sys.executable, "-c",
'import sys,os;' 'import sys,os;'
'sys.stdout.write(sys.stdin.read(47));' 'sys.stdout.write(sys.stdin.read(47));'
'sys.stderr.write("xyz"*%d);' 'sys.stderr.write("x" * %d);'
'sys.stdout.write(sys.stdin.read())' % pipe_buf], 'sys.stdout.write(sys.stdin.read())' %
support.PIPE_MAX_SIZE],
stdin=subprocess.PIPE, stdin=subprocess.PIPE,
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.PIPE) stderr=subprocess.PIPE)
self.addCleanup(p.stdout.close) self.addCleanup(p.stdout.close)
self.addCleanup(p.stderr.close) self.addCleanup(p.stderr.close)
self.addCleanup(p.stdin.close) self.addCleanup(p.stdin.close)
string_to_write = b"abc"*pipe_buf string_to_write = b"a" * support.PIPE_MAX_SIZE
(stdout, stderr) = p.communicate(string_to_write) (stdout, stderr) = p.communicate(string_to_write)
self.assertEqual(stdout, string_to_write) self.assertEqual(stdout, string_to_write)