From ed8e886f4f59df4eceefeb7eef2e3d146967df34 Mon Sep 17 00:00:00 2001 From: Serhiy Storchaka Date: Mon, 28 Apr 2025 10:42:40 +0300 Subject: gh-132742: Improve tests for fcntl.ioctl() (GH-132791) * Use better tests for integer argument. * Add also parallel tests for tcflush() and tcflow(). --- Lib/test/test_ioctl.py | 55 +++++++++++++++++++++++++++++++++++++++++++----- Lib/test/test_termios.py | 55 ++++++++++++++++++++++++++++++++++++++++++++++-- 2 files changed, 103 insertions(+), 7 deletions(-) diff --git a/Lib/test/test_ioctl.py b/Lib/test/test_ioctl.py index b291333..9fd3b61 100644 --- 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 d6eb001..18965f0 100644 --- 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) -- cgit v0.12