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.

1 files changed, 16 insertions, 9 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():