[Python-checkins] cpython: Issue #12196: Add PIPE_MAX_SIZE to test.support, constant larger than the

[charles-francois.natali]

http://hg.python.org/cpython/rev/866d959dea8e
changeset:   70474:866d959dea8e
user:        Charles-François Natali <neologix at free.fr>
date:        Sun May 29 16:36:44 2011 +0200
summary:
  Issue #12196: Add PIPE_MAX_SIZE to test.support, constant larger than the
underlying OS pipe buffer size.

files:
  Lib/test/support.py         |   9 ++++++++-
  Lib/test/test_io.py         |   2 +-
  Lib/test/test_subprocess.py |  11 ++++-------
  3 files changed, 13 insertions(+), 9 deletions(-)


diff --git a/Lib/test/support.py b/Lib/test/support.py
--- a/Lib/test/support.py
+++ b/Lib/test/support.py
@@ -48,7 +48,7 @@
     "threading_cleanup", "reap_children", "cpython_only", "check_impl_detail",
     "get_attribute", "swap_item", "swap_attr", "requires_IEEE_754",
     "TestHandler", "Matcher", "can_symlink", "skip_unless_symlink",
-    "import_fresh_module", "requires_zlib"
+    "import_fresh_module", "requires_zlib", "PIPE_MAX_SIZE"
     ]
 
 class Error(Exception):
@@ -409,6 +409,13 @@
 
 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
 requires_IEEE_754 = unittest.skipUnless(
     float.__getformat__("double").startswith("IEEE"),
diff --git a/Lib/test/test_io.py b/Lib/test/test_io.py
--- a/Lib/test/test_io.py
+++ b/Lib/test/test_io.py
@@ -2683,7 +2683,7 @@
             # The buffered IO layer must check for pending signal
             # handlers, which in this case will invoke alarm_interrupt().
             self.assertRaises(ZeroDivisionError,
-                              wio.write, item * (1024 * 1024))
+                        wio.write, item * (support.PIPE_MAX_SIZE // len(item)))
             t.join()
             # We got one byte, get another one and check that it isn't a
             # repeat of the first one.
diff --git a/Lib/test/test_subprocess.py b/Lib/test/test_subprocess.py
--- a/Lib/test/test_subprocess.py
+++ b/Lib/test/test_subprocess.py
@@ -489,24 +489,21 @@
         # This test will probably deadlock rather than fail, if
         # communicate() does not work properly.
         x, y = os.pipe()
-        if mswindows:
-            pipe_buf = 512
-        else:
-            pipe_buf = os.fpathconf(x, "PC_PIPE_BUF")
         os.close(x)
         os.close(y)
         p = subprocess.Popen([sys.executable, "-c",
                               'import sys,os;'
                               'sys.stdout.write(sys.stdin.read(47));'
-                              'sys.stderr.write("xyz"*%d);'
-                              'sys.stdout.write(sys.stdin.read())' % pipe_buf],
+                              'sys.stderr.write("x" * %d);'
+                              'sys.stdout.write(sys.stdin.read())' %
+                              support.PIPE_MAX_SIZE],
                              stdin=subprocess.PIPE,
                              stdout=subprocess.PIPE,
                              stderr=subprocess.PIPE)
         self.addCleanup(p.stdout.close)
         self.addCleanup(p.stderr.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)
         self.assertEqual(stdout, string_to_write)
 

-- 
Repository URL: http://hg.python.org/cpython