mirrors/cpython
1
0
Fork 0
mirror of https://github.com/python/cpython.git synced 2025-08-04 17:08:35 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 13

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

Browse source 
* Use better tests for integer argument.
* Add also parallel tests for tcflush() and tcflow().
This commit is contained in:
Serhiy Storchaka 2025-04-28 10:42:40 +03:00 • committed by GitHub
parent 632978f005
commit ed8e886f4f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 103 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

55
Lib/test/test_ioctl.py
View file

@ -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)

55
Lib/test/test_termios.py
View file

@ -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)