https://github.com/python/cpython/commit/ed8e886f4f59df4eceefeb7eef2e3d146967df34
commit: ed8e886f4f59df4eceefeb7eef2e3d146967df34
branch: main
author: Serhiy Storchaka <storch...@gmail.com>
committer: serhiy-storchaka <storch...@gmail.com>
date: 2025-04-28T10:42:40+03:00
summary:
gh-132742: Improve tests for fcntl.ioctl() (GH-132791)
* Use better tests for integer argument.
* Add also parallel tests for tcflush() and tcflow().
files:
M Lib/test/test_ioctl.py
M Lib/test/test_termios.py
diff --git a/Lib/test/test_ioctl.py b/Lib/test/test_ioctl.py
index b291333b00010b..9fd3b6196b81c2 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 @@ 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_set_window_size(self):
# (rows, columns, xpixel, ypixel)
diff --git a/Lib/test/test_termios.py b/Lib/test/test_termios.py
index d6eb00175b4c85..18965f09a2569e 100644
--- a/Lib/test/test_termios.py
+++ b/Lib/test/test_termios.py
@@ -2,8 +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')
@@ -13,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()
@@ -147,6 +149,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')
+
@support.skip_android_selinux('tcflow')
def test_tcflow(self):
termios.tcflow(self.fd, termios.TCOOFF)
@@ -165,6 +190,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)
_______________________________________________
Python-checkins mailing list -- python-checkins@python.org
To unsubscribe send an email to python-checkins-le...@python.org
https://mail.python.org/mailman3/lists/python-checkins.python.org/
Member address: arch...@mail-archive.com
