1 files changed, 58 insertions, 1 deletions
diff --git a/Lib/test/test_fcntl.py b/Lib/test/test_fcntl.py
index b84c98ef3a2..7140a7b4f29 100644
--- a/Lib/test/test_fcntl.py
+++ b/Lib/test/test_fcntl.py
@@ -11,7 +11,7 @@ from test.support import (
     cpython_only, get_pagesize, is_apple, requires_subprocess, verbose
 )
 from test.support.import_helper import import_module
-from test.support.os_helper import TESTFN, unlink
+from test.support.os_helper import TESTFN, unlink, make_bad_fd
 
 
 # Skip test if no fcntl module.
@@ -228,6 +228,63 @@ class TestFcntl(unittest.TestCase):
             os.close(test_pipe_r)
             os.close(test_pipe_w)
 
+    def _check_fcntl_not_mutate_len(self, nbytes=None):
+        self.f = open(TESTFN, 'wb')
+        buf = struct.pack('ii', fcntl.F_OWNER_PID, os.getpid())
+        if nbytes is not None:
+            buf += b' ' * (nbytes - len(buf))
+        else:
+            nbytes = len(buf)
+        save_buf = bytes(buf)
+        r = fcntl.fcntl(self.f, fcntl.F_SETOWN_EX, buf)
+        self.assertIsInstance(r, bytes)
+        self.assertEqual(len(r), len(save_buf))
+        self.assertEqual(buf, save_buf)
+        type, pid = memoryview(r).cast('i')[:2]
+        self.assertEqual(type, fcntl.F_OWNER_PID)
+        self.assertEqual(pid, os.getpid())
+
+        buf = b' ' * nbytes
+        r = fcntl.fcntl(self.f, fcntl.F_GETOWN_EX, buf)
+        self.assertIsInstance(r, bytes)
+        self.assertEqual(len(r), len(save_buf))
+        self.assertEqual(buf, b' ' * nbytes)
+        type, pid = memoryview(r).cast('i')[:2]
+        self.assertEqual(type, fcntl.F_OWNER_PID)
+        self.assertEqual(pid, os.getpid())
+
+        buf = memoryview(b' ' * nbytes)
+        r = fcntl.fcntl(self.f, fcntl.F_GETOWN_EX, buf)
+        self.assertIsInstance(r, bytes)
+        self.assertEqual(len(r), len(save_buf))
+        self.assertEqual(bytes(buf), b' ' * nbytes)
+        type, pid = memoryview(r).cast('i')[:2]
+        self.assertEqual(type, fcntl.F_OWNER_PID)
+        self.assertEqual(pid, os.getpid())
+
+    @unittest.skipUnless(
+        hasattr(fcntl, "F_SETOWN_EX") and hasattr(fcntl, "F_GETOWN_EX"),
+        "requires F_SETOWN_EX and F_GETOWN_EX")
+    def test_fcntl_small_buffer(self):
+        self._check_fcntl_not_mutate_len()
+
+    @unittest.skipUnless(
+        hasattr(fcntl, "F_SETOWN_EX") and hasattr(fcntl, "F_GETOWN_EX"),
+        "requires F_SETOWN_EX and F_GETOWN_EX")
+    def test_fcntl_large_buffer(self):
+        self._check_fcntl_not_mutate_len(2024)
+
+    @unittest.skipUnless(hasattr(fcntl, 'F_DUPFD'), 'need fcntl.F_DUPFD')
+    def test_bad_fd(self):
+        # gh-134744: Test error handling
+        fd = make_bad_fd()
+        with self.assertRaises(OSError):
+            fcntl.fcntl(fd, fcntl.F_DUPFD, 0)
+        with self.assertRaises(OSError):
+            fcntl.fcntl(fd, fcntl.F_DUPFD, b'\0' * 10)
+        with self.assertRaises(OSError):
+            fcntl.fcntl(fd, fcntl.F_DUPFD, b'\0' * 2048)
+
 
 if __name__ == '__main__':
     unittest.main()