8000 gh-132742: Improve tests for fcntl.ioctl() (GH-132791) · python/cpython@ed8e886 · GitHub
[go: up one dir, main page]
Skip to content

Commit ed8e886

Browse files
serhiy-storchakaserhiy-storchaka
authored
gh-132742: Improve tests for fcntl.ioctl() (GH-132791)
* Use better tests for integer argument. * Add also parallel tests for tcflush() and tcflow().
1 parent 632978f commit ed8e886

File tree

2 files changed

+103
-7
lines changed

2 files changed

+103
-7
lines changed

‎Lib/test/test_ioctl.py

Lines changed: 50 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
import array
22
import os
33
import struct
4+
import sys
45
import threading
56
import unittest
67
from test.support import get_attribute
@@ -139,11 +140,55 @@ def setUp(self):
139140
self.addCleanup(os.close, self.master_fd)
140141

141142
@unittest.skipUnless(hasattr(termios, 'TCFLSH'), 'requires termios.TCFLSH')
142-
def test_ioctl_tcflush(self):
143-
r = fcntl.ioctl(self.slave_fd, termios.TCFLSH, termios.TCIFLUSH)
144-
self.assertEqual(r, 0)
145-
r = fcntl.ioctl(self.slave_fd, termios.TCFLSH, termios.TCOFLUSH)
146-
self.assertEqual(r, 0)
143+
def test_ioctl_clear_input_or_output(self):
144+
wfd = self.slave_fd
145+
rfd = self.master_fd
146+
inbuf = sys.platform == 'linux'
147+
148+
os.write(wfd, b'abcdef')
149+
self.assertEqual(os.read(rfd, 2), b'ab')
150+
if inbuf:
151+
# don't flush input
152+
fcntl.ioctl(rfd, termios.TCFLSH, termios.TCOFLUSH)
153+
else:
154+
# don't flush output
155+
fcntl.ioctl(wfd, termios.TCFLSH, termios.TCIFLUSH)
156+
self.assertEqual(os.read(rfd, 2), b'cd')
157+
if inbuf:
158+
# flush input
159+
fcntl.ioctl(rfd, termios.TCFLSH, termios.TCIFLUSH)
160+
else:
161+
# flush output
162+
fcntl.ioctl(wfd, termios.TCFLSH, termios.TCOFLUSH)
163+
os.write(wfd, b'ABCDEF')
164+
self.assertEqual(os.read(rfd, 1024), b'ABCDEF')
165+
166+
@unittest.skipUnless(sys.platform == 'linux', 'only works on Linux')
167+
@unittest.skipUnless(hasattr(termios, 'TCXONC'), 'requires termios.TCXONC')
168+
def test_ioctl_suspend_and_resume_output(self):
169+
wfd = self.slave_fd
170+
rfd = self.master_fd
171+
write_suspended = threading.Event()
172+
write_finished = threading.Event()
173+
174+
def writer():
175+
os.write(wfd, b'abc')
176+
write_suspended.wait()
177+
os.write(wfd, b'def')
178+
write_finished.set()
179+
180+
with threading_helper.start_threads([threading.Thread(target=writer)]):
181+
self.assertEqual(os.read(rfd, 3), b'abc')
182+
try:
183+
fcntl.ioctl(wfd, termios.TCXONC, termios.TCOOFF)
184+
write_suspended.set()
185+
self.assertFalse(write_finished.wait(0.5),
186+
'output was not suspended')
187+
finally:
188+
fcntl.ioctl(wfd, termios.TCXONC, termios.TCOON)
189+
self.assertTrue(write_finished.wait(0.5),
190+
'output was not resumed')
191+
self.assertEqual(os.read(rfd, 1024), b'def')
147192

148193
def test_ioctl_set_window_size(self):
149194
# (rows, columns, xpixel, ypixel)

‎Lib/test/test_termios.py

Lines changed: 53 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,8 +2,10 @@
22
import os
33
import sys
44
import tempfile
5+
import threading
56
import unittest
67
from test import support
8+
from test.support import threading_helper
79
from test.support.import_helper import import_module
810

911
termios = import_module('termios')
@@ -13,8 +15,8 @@
1315
class TestFunctions(unittest.TestCase):
1416

1517
def setUp(self):
16-
master_fd, self.fd = os.openpty()
17-
self.addCleanup(os.close, master_fd)
18+
self.master_fd, self.fd = os.openpty()
19+
self.addCleanup(os.close, self.master_fd)
1820
self.stream = self.enterContext(open(self.fd, 'wb', buffering=0))
1921
tmp = self.enterContext(tempfile.TemporaryFile(mode='wb', buffering=0))
2022
self.bad_fd = tmp.fileno()
@@ -147,6 +149,29 @@ def test_tcflush_errors(self):
147149
self.assertRaises(TypeError, termios.tcflush, object(), termios.TCIFLUSH)
148150
self.assertRaises(TypeError, termios.tcflush, self.fd)
149151

152+
def test_tcflush_clear_input_or_output(self):
153+
wfd = self.fd
154+
rfd = self.master_fd
155+
inbuf = sys.platform == 'linux'
156+
157+
os.write(wfd, b'abcdef')
158+
self.assertEqual(os.read(rfd, 2), b'ab')
159+
if inbuf:
160+
# don't flush input
161+
termios.tcflush(rfd, termios.TCOFLUSH)
162+
else:
163+
# don't flush output
164+
termios.tcflush(wfd, termios.TCIFLUSH)
165+
self.assertEqual(os.read(rfd, 2), b'cd')
166+
if inbuf:
167+
# flush input
168+
termios.tcflush(rfd, termios.TCIFLUSH)
169+
else:
170+
# flush output
171+
termios.tcflush(wfd, termios.TCOFLUSH)
172+
os.write(wfd, b'ABCDEF')
173+
self.assertEqual(os.read(rfd, 1024), b'ABCDEF')
174+
150175
@support.skip_android_selinux('tcflow')
151176
def test_tcflow(self):
152177
termios.tcflow(self.fd, termios.TCOOFF)
@@ -165,6 +190,32 @@ def test_tcflow_errors(self):
165190
self.assertRaises(TypeError, termios.tcflow, object(), termios.TCOON)
166191
self.assertRaises(TypeError, termios.tcflow, self.fd)
167192

193+
@unittest.skipUnless(sys.platform == 'linux', 'only works on Linux')
194+
def test_tcflow_suspend_and_resume_output(self):
195+
wfd = self.fd
196+
rfd = self.master_fd
197+
write_suspended = threading.Event()
198+
write_finished = threading.Event()
199+
200+
def writer():
201+
os.write(wfd, b'abc')
202+
write_suspended.wait()
203+
os.write(wfd, b'def')
204+
write_finished.set()
205+
206+
with threading_helper.start_threads([threading.Thread(target=writer)]):
207+
self.assertEqual(os.read(rfd, 3), b'abc')
208+
try:
209+
termios.tcflow(wfd, termios.TCOOFF)
210+
write_suspended.set()
211+
self.assertFalse(write_finished.wait(0.5),
212+
'output was not suspended')
213+
finally:
214+
termios.tcflow(wfd, termios.TCOON)
215+
self.assertTrue(write_finished.wait(0.5),
216+
'output was not resumed')
217+
self.assertEqual(os.read(rfd, 1024), b'def')
218+
168219
def test_tcgetwinsize(self):
169220
size = termios.tcgetwinsize(self.fd)
170221
self.assertIsInstance(size, tuple)

0 commit comments

Comments
 (0)
0