[3.13] gh-132742: Improve tests for fcntl.ioctl() (GH-132791)

Lib/test/test_ioctl.py

@@ -1,6 +1,7 @@
 import array
 import os
 import struct
+import sys
 import threading
 import unittest
 from test.support import get_attribute
@@ -139,11 +140,55 @@ def setUp(self):
         self.addCleanup(os.close, self.master_fd)
 
     @unittest.skipUnless(hasattr(termios, 'TCFLSH'), 'requires termios.TCFLSH')
-    def test_ioctl_tcflush(self):
-        r = fcntl.ioctl(self.slave_fd, termios.TCFLSH, termios.TCIFLUSH)
-        self.assertEqual(r, 0)
-        r = fcntl.ioctl(self.slave_fd, termios.TCFLSH, termios.TCOFLUSH)
-        self.assertEqual(r, 0)
+    def test_ioctl_clear_input_or_output(self):
+        wfd = self.slave_fd
+        rfd = self.master_fd
+        inbuf = sys.platform == 'linux'
+
+        os.write(wfd, b'abcdef')
+        self.assertEqual(os.read(rfd, 2), b'ab')
+        if inbuf:
+            # don't flush input
+            fcntl.ioctl(rfd, termios.TCFLSH, termios.TCOFLUSH)
+        else:
+            # don't flush output
+            fcntl.ioctl(wfd, termios.TCFLSH, termios.TCIFLUSH)
+        self.assertEqual(os.read(rfd, 2), b'cd')
+        if inbuf:
+            # flush input
+            fcntl.ioctl(rfd, termios.TCFLSH, termios.TCIFLUSH)
+        else:
+            # flush output
+            fcntl.ioctl(wfd, termios.TCFLSH, termios.TCOFLUSH)
+        os.write(wfd, b'ABCDEF')
+        self.assertEqual(os.read(rfd, 1024), b'ABCDEF')
+
+    @unittest.skipUnless(sys.platform == 'linux', 'only works on Linux')
+    @unittest.skipUnless(hasattr(termios, 'TCXONC'), 'requires termios.TCXONC')
+    def test_ioctl_suspend_and_resume_output(self):
+        wfd = self.slave_fd
+        rfd = self.master_fd
+        write_suspended = threading.Event()
+        write_finished = threading.Event()
+
+        def writer():
+            os.write(wfd, b'abc')
+            write_suspended.wait()
+            os.write(wfd, b'def')
+            write_finished.set()
+
+        with threading_helper.start_threads([threading.Thread(target=writer)]):
+            self.assertEqual(os.read(rfd, 3), b'abc')
+            try:
+                fcntl.ioctl(wfd, termios.TCXONC, termios.TCOOFF)
+                write_suspended.set()
+                self.assertFalse(write_finished.wait(0.5),
+                                 'output was not suspended')
+            finally:
+                fcntl.ioctl(wfd, termios.TCXONC, termios.TCOON)
+            self.assertTrue(write_finished.wait(0.5),
+                            'output was not resumed')
+            self.assertEqual(os.read(rfd, 1024), b'def')
 
     def test_ioctl_signed_unsigned_code_param(self):
         if termios.TIOCSWINSZ < 0:

Lib/test/test_termios.py

@@ -2,7 +2,10 @@
 import os
 import sys
 import tempfile
+import threading
 import unittest
+from test import support
+from test.support import threading_helper
 from test.support.import_helper import import_module
 
 termios = import_module('termios')
@@ -12,8 +15,8 @@
 class TestFunctions(unittest.TestCase):
 
     def setUp(self):
-        master_fd, self.fd = os.openpty()
-        self.addCleanup(os.close, master_fd)
+        self.master_fd, self.fd = os.openpty()
+        self.addCleanup(os.close, self.master_fd)
         self.stream = self.enterContext(open(self.fd, 'wb', buffering=0))
         tmp = self.enterContext(tempfile.TemporaryFile(mode='wb', buffering=0))
         self.bad_fd = tmp.fileno()
@@ -136,6 +139,29 @@ def test_tcflush_errors(self):
         self.assertRaises(TypeError, termios.tcflush, object(), termios.TCIFLUSH)
         self.assertRaises(TypeError, termios.tcflush, self.fd)
 
+    def test_tcflush_clear_input_or_output(self):
+        wfd = self.fd
+        rfd = self.master_fd
+        inbuf = sys.platform == 'linux'
+
+        os.write(wfd, b'abcdef')
+        self.assertEqual(os.read(rfd, 2), b'ab')
+        if inbuf:
+            # don't flush input
+            termios.tcflush(rfd, termios.TCOFLUSH)
+        else:
+            # don't flush output
+            termios.tcflush(wfd, termios.TCIFLUSH)
+        self.assertEqual(os.read(rfd, 2), b'cd')
+        if inbuf:
+            # flush input
+            termios.tcflush(rfd, termios.TCIFLUSH)
+        else:
+            # flush output
+            termios.tcflush(wfd, termios.TCOFLUSH)
+        os.write(wfd, b'ABCDEF')
+        self.assertEqual(os.read(rfd, 1024), b'ABCDEF')
+
     def test_tcflow(self):
         termios.tcflow(self.fd, termios.TCOOFF)
         termios.tcflow(self.fd, termios.TCOON)
@@ -152,6 +178,32 @@ def test_tcflow_errors(self):
         self.assertRaises(TypeError, termios.tcflow, object(), termios.TCOON)
         self.assertRaises(TypeError, termios.tcflow, self.fd)
 
+    @unittest.skipUnless(sys.platform == 'linux', 'only works on Linux')
+    def test_tcflow_suspend_and_resume_output(self):
+        wfd = self.fd
+        rfd = self.master_fd
+        write_suspended = threading.Event()
+        write_finished = threading.Event()
+
+        def writer():
+            os.write(wfd, b'abc')
+            write_suspended.wait()
+            os.write(wfd, b'def')
+            write_finished.set()
+
+        with threading_helper.start_threads([threading.Thread(target=writer)]):
+            self.assertEqual(os.read(rfd, 3), b'abc')
+            try:
+                termios.tcflow(wfd, termios.TCOOFF)
+                write_suspended.set()
+                self.assertFalse(write_finished.wait(0.5),
+                                 'output was not suspended')
+            finally:
+                termios.tcflow(wfd, termios.TCOON)
+            self.assertTrue(write_finished.wait(0.5),
+                            'output was not resumed')
+            self.assertEqual(os.read(rfd, 1024), b'def')
+
     def test_tcgetwinsize(self):
         size = termios.tcgetwinsize(self.fd)
         self.assertIsInstance(size, tuple)