]> git.ipfire.org Git - thirdparty/Python/cpython.git/commitdiff
[3.11] gh-110673: test_pty raises on short write (GH-110677) (#110743)
authorMiss Islington (bot) <31488909+miss-islington@users.noreply.github.com>
Thu, 12 Oct 2023 08:09:49 +0000 (10:09 +0200)
committerGitHub <noreply@github.com>
Thu, 12 Oct 2023 08:09:49 +0000 (10:09 +0200)
gh-110673: test_pty raises on short write (GH-110677)

Add write_all() helper function to test_pty to raise an exception on
short write: if os.writes() does not write all bytes. It should not
happen for a PTY.
(cherry picked from commit b4e8049766a46a9e6548b18d7e9a0c9f573cd122)

Co-authored-by: Victor Stinner <vstinner@python.org>
Lib/test/test_pty.py

index a971f6b0250efba90fe1801163f4d9cb0a8eb097..f31a68c5d84e037585b69cfed1d83877bd2ab3c5 100644 (file)
@@ -76,6 +76,15 @@ def expectedFailureIfStdinIsTTY(fun):
         pass
     return fun
 
+
+def write_all(fd, data):
+    written = os.write(fd, data)
+    if written != len(data):
+        # gh-73256, gh-110673: It should never happen, but check just in case
+        raise Exception(f"short write: os.write({fd}, {len(data)} bytes) "
+                        f"wrote {written} bytes")
+
+
 # Marginal testing of pty suite. Cannot do extensive 'do or fail' testing
 # because pty code is not too portable.
 class PtyTest(unittest.TestCase):
@@ -170,14 +179,14 @@ class PtyTest(unittest.TestCase):
             os.set_blocking(master_fd, blocking)
 
         debug("Writing to slave_fd")
-        os.write(slave_fd, TEST_STRING_1)
+        write_all(slave_fd, TEST_STRING_1)
         s1 = _readline(master_fd)
         self.assertEqual(b'I wish to buy a fish license.\n',
                          normalize_output(s1))
 
         debug("Writing chunked output")
-        os.write(slave_fd, TEST_STRING_2[:5])
-        os.write(slave_fd, TEST_STRING_2[5:])
+        write_all(slave_fd, TEST_STRING_2[:5])
+        write_all(slave_fd, TEST_STRING_2[5:])
         s2 = _readline(master_fd)
         self.assertEqual(b'For my pet fish, Eric.\n', normalize_output(s2))
 
@@ -360,8 +369,8 @@ class SmallPtyTests(unittest.TestCase):
         masters = [s.fileno() for s in socketpair]
 
         # Feed data.  Smaller than PIPEBUF.  These writes will not block.
-        os.write(masters[1], b'from master')
-        os.write(write_to_stdin_fd, b'from stdin')
+        write_all(masters[1], b'from master')
+        write_all(write_to_stdin_fd, b'from stdin')
 
         # Expect three select calls, the last one will cause IndexError
         pty.select = self._mock_select