pythongh-132742: Improve tests for fcntl.ioctl()

* Do not skip ALL ioctl() tests when /dev/tty is not available.
* Use better tests for integer argument.
* Add also parallel tests for tcflush() and tcflow().

Author: serhiy-storchaka

Lib/test/test_ioctl.py

@@ -1,32 +1,35 @@
 import array
+import os
+import struct
+import threading
 import unittest
 from test.support import get_attribute
+from test.support import threading_helper
 from test.support.import_helper import import_module
-import os, struct
 fcntl = import_module('fcntl')
 termios = import_module('termios')
-get_attribute(termios, 'TIOCGPGRP') #Can't run tests without this feature
-
-try:
-    tty = open("/dev/tty", "rb")
-except OSError:
-    raise unittest.SkipTest("Unable to open /dev/tty")
-else:
-    with tty:
-        # Skip if another process is in foreground
-        r = fcntl.ioctl(tty, termios.TIOCGPGRP, struct.pack("i", 0))
-    rpgrp = struct.unpack("i", r)[0]
-    if rpgrp not in (os.getpgrp(), os.getsid(0)):
-        raise unittest.SkipTest("Neither the process group nor the session "
-                                "are attached to /dev/tty")
-    del tty, r, rpgrp
 
 try:
     import pty
 except ImportError:
     pty = None
 
-class IoctlTests(unittest.TestCase):
+class IoctlTestsTty(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        TIOCGPGRP = get_attribute(termios, 'TIOCGPGRP')
+        try:
+            tty = open("/dev/tty", "rb")
+        except OSError:
+            raise unittest.SkipTest("Unable to open /dev/tty")
+        with tty:
+            # Skip if another process is in foreground
+            r = fcntl.ioctl(tty, TIOCGPGRP, struct.pack("i", 0))
+        rpgrp = struct.unpack("i", r)[0]
+        if rpgrp not in (os.getpgrp(), os.getsid(0)):
+            raise unittest.SkipTest("Neither the process group nor the session "
+                                    "are attached to /dev/tty")
+
     def test_ioctl_immutable_buf(self):
         # If this process has been put into the background, TIOCGPGRP returns
         # the session ID instead of the process group id.
@@ -132,23 +135,48 @@ def test_ioctl_mutate_2048(self):
         self._check_ioctl_mutate_len(2048)
         self.assertRaises(ValueError, self._check_ioctl_not_mutate_len, 2048)
 
-    def test_ioctl_tcflush(self):
-        with open("/dev/tty", "rb") as tty:
-            r = fcntl.ioctl(tty, termios.TCFLSH, termios.TCIFLUSH)
-            self.assertEqual(r, 0)
 
-    @unittest.skipIf(pty is None, 'pty module required')
+@unittest.skipIf(pty is None, 'pty module required')
+class IoctlTestsPty(unittest.TestCase):
+    def setUp(self):
+        self.master_fd, self.slave_fd = pty.openpty()
+        self.addCleanup(os.close, self.slave_fd)
+        self.addCleanup(os.close, self.master_fd)
+
+    def test_ioctl_clear_input(self):
+        os.write(self.slave_fd, b'abcdef')
+        self.assertEqual(os.read(self.master_fd, 2), b'ab')
+        fcntl.ioctl(self.master_fd, termios.TCFLSH, termios.TCOFLUSH)  # don't flush input
+        self.assertEqual(os.read(self.master_fd, 2), b'cd')
+        fcntl.ioctl(self.master_fd, termios.TCFLSH, termios.TCIFLUSH)  # flush input
+        os.write(self.slave_fd, b'ABCDEF')
+        self.assertEqual(os.read(self.master_fd, 1024), b'ABCDEF')
+
+    def test_tcflow_suspend_and_resume_output(self):
+        write_suspended = threading.Event()
+        write_finished = threading.Event()
+
+        def writer():
+            os.write(self.slave_fd, b'abc')
+            write_suspended.wait()
+            os.write(self.slave_fd, b'def')
+            write_finished.set()
+
+        with threading_helper.start_threads([threading.Thread(target=writer)]):
+            self.assertEqual(os.read(self.master_fd, 1024), b'abc')
+            termios.tcflow(self.slave_fd, termios.TCOOFF)
+            write_suspended.set()
+            self.assertFalse(write_finished.wait(0.5))
+            termios.tcflow(self.slave_fd, termios.TCOON)
+            self.assertTrue(write_finished.wait(0.5))
+            self.assertEqual(os.read(self.master_fd, 1024), b'def')
+
     def test_ioctl_set_window_size(self):
-        mfd, sfd = pty.openpty()
-        try:
-            # (rows, columns, xpixel, ypixel)
-            our_winsz = struct.pack("HHHH", 20, 40, 0, 0)
-            result = fcntl.ioctl(mfd, termios.TIOCSWINSZ, our_winsz)
-            new_winsz = struct.unpack("HHHH", result)
-            self.assertEqual(new_winsz[:2], (20, 40))
-        finally:
-            os.close(mfd)
-            os.close(sfd)
+        # (rows, columns, xpixel, ypixel)
+        our_winsz = struct.pack("HHHH", 20, 40, 0, 0)
+        result = fcntl.ioctl(self.master_fd, termios.TIOCSWINSZ, our_winsz)
+        new_winsz = struct.unpack("HHHH", result)
+        self.assertEqual(new_winsz[:2], (20, 40))
 
 
 if __name__ == "__main__":

Lib/test/test_termios.py

@@ -13,8 +13,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()
@@ -147,6 +147,15 @@ 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(self):
+        os.write(self.fd, b'abcdef')
+        self.assertEqual(os.read(self.master_fd, 2), b'ab')
+        termios.tcflush(self.master_fd, termios.TCOFLUSH)  # don't flush input
+        self.assertEqual(os.read(self.master_fd, 2), b'cd')
+        termios.tcflush(self.master_fd, termios.TCIFLUSH)  # flush input
+        os.write(self.fd, b'ABCDEF')
+        self.assertEqual(os.read(self.master_fd, 1024), b'ABCDEF')
+
     @support.skip_android_selinux('tcflow')
     def test_tcflow(self):
         termios.tcflow(self.fd, termios.TCOOFF)
@@ -165,6 +174,25 @@ def test_tcflow_errors(self):
         self.assertRaises(TypeError, termios.tcflow, object(), termios.TCOON)
         self.assertRaises(TypeError, termios.tcflow, self.fd)
 
+    def test_tcflow_suspend_and_resume_output(self):
+        write_suspended = threading.Event()
+        write_finished = threading.Event()
+
+        def writer():
+            os.write(self.fd, b'abc')
+            write_suspended.wait()
+            os.write(self.fd, b'def')
+            write_finished.set()
+
+        with threading_helper.start_threads([threading.Thread(target=writer)]):
+            self.assertEqual(os.read(self.master_fd, 1024), b'abc')
+            termios.tcflow(self.fd, termios.TCOOFF)
+            write_suspended.set()
+            self.assertFalse(write_finished.wait(0.5))
+            termios.tcflow(self.fd, termios.TCOON)
+            self.assertTrue(write_finished.wait(0.5))
+            self.assertEqual(os.read(self.master_fd, 1024), b'def')
+
     def test_tcgetwinsize(self):
         size = termios.tcgetwinsize(self.fd)
         self.assertIsInstance(size, tuple)