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

* Use better tests for integer argument.
* Add also parallel tests for tcflush() and tcflow().

diff --git a/Lib/test/test_ioctl.py b/Lib/test/test_ioctl.py
index b291333b00010bcba3cb0a0fc2051fae24dfde96..9fd3b6196b81c2eddb709d95ae7df4ae396a5ce8 100644 (file)
--- a/Lib/test/test_ioctl.py
+++ b/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 @@ class IoctlTestsPty(unittest.TestCase):
         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_set_window_size(self):
         # (rows, columns, xpixel, ypixel)

diff --git a/Lib/test/test_termios.py b/Lib/test/test_termios.py
index d6eb00175b4c8595212cc9ea9626f3d032376e8c..18965f09a2569efd626094edc5af7dea2c1aa960 100644 (file)
--- a/Lib/test/test_termios.py
+++ b/Lib/test/test_termios.py
@@ -2,8 +2,10 @@ import errno
 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')
@@ -13,8 +15,8 @@ termios = import_module('termios')
 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 +149,29 @@ class TestFunctions(unittest.TestCase):
         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')
+
     @support.skip_android_selinux('tcflow')
     def test_tcflow(self):
         termios.tcflow(self.fd, termios.TCOOFF)
@@ -165,6 +190,32 @@ class TestFunctions(unittest.TestCase):
         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)