1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
|
import array
import os
import struct
import sys
import threading
import unittest
from test import support
from test.support import os_helper, threading_helper
from test.support.import_helper import import_module
fcntl = import_module('fcntl')
termios = import_module('termios')
class IoctlTestsTty(unittest.TestCase):
@classmethod
def setUpClass(cls):
TIOCGPGRP = support.get_attribute(termios, 'TIOCGPGRP')
try:
tty = open("/dev/tty", "rb")
except OSError:
raise unittest.SkipTest("Unable to open /dev/tty")
with tty:
# Skip if another process is in foreground
r = fcntl.ioctl(tty, TIOCGPGRP, struct.pack("i", 0))
rpgrp = struct.unpack("i", r)[0]
if rpgrp not in (os.getpgrp(), os.getsid(0)):
raise unittest.SkipTest("Neither the process group nor the session "
"are attached to /dev/tty")
def test_ioctl_immutable_buf(self):
# If this process has been put into the background, TIOCGPGRP returns
# the session ID instead of the process group id.
ids = (os.getpgrp(), os.getsid(0))
with open("/dev/tty", "rb") as tty:
# string
buf = " "*8
r = fcntl.ioctl(tty, termios.TIOCGPGRP, buf)
self.assertIsInstance(r, bytes)
rpgrp = memoryview(r).cast('i')[0]
self.assertIn(rpgrp, ids)
# bytes
buf = b" "*8
r = fcntl.ioctl(tty, termios.TIOCGPGRP, buf)
self.assertIsInstance(r, bytes)
rpgrp = memoryview(r).cast('i')[0]
self.assertIn(rpgrp, ids)
# read-only buffer
r = fcntl.ioctl(tty, termios.TIOCGPGRP, memoryview(buf))
self.assertIsInstance(r, bytes)
rpgrp = memoryview(r).cast('i')[0]
self.assertIn(rpgrp, ids)
def test_ioctl_mutable_buf(self):
ids = (os.getpgrp(), os.getsid(0))
with open("/dev/tty", "rb") as tty:
buf = bytearray(b" "*8)
r = fcntl.ioctl(tty, termios.TIOCGPGRP, buf)
self.assertEqual(r, 0)
rpgrp = memoryview(buf).cast('i')[0]
self.assertIn(rpgrp, ids)
def test_ioctl_no_mutate_buf(self):
ids = (os.getpgrp(), os.getsid(0))
with open("/dev/tty", "rb") as tty:
buf = bytearray(b" "*8)
save_buf = bytes(buf)
r = fcntl.ioctl(tty, termios.TIOCGPGRP, buf, False)
self.assertEqual(bytes(buf), save_buf)
self.assertIsInstance(r, bytes)
rpgrp = memoryview(r).cast('i')[0]
self.assertIn(rpgrp, ids)
def _create_int_buf(self, nbytes=None):
buf = array.array('i')
intsize = buf.itemsize
# A fill value unlikely to be in `ids`
fill = -12345
if nbytes is not None:
# Extend the buffer so that it is exactly `nbytes` bytes long
buf.extend([fill] * (nbytes // intsize))
self.assertEqual(len(buf) * intsize, nbytes) # sanity check
else:
buf.append(fill)
return buf
def _check_ioctl_mutate_len(self, nbytes=None):
ids = (os.getpgrp(), os.getsid(0))
buf = self._create_int_buf(nbytes)
with open("/dev/tty", "rb") as tty:
r = fcntl.ioctl(tty, termios.TIOCGPGRP, buf)
rpgrp = buf[0]
self.assertEqual(r, 0)
self.assertIn(rpgrp, ids)
def _check_ioctl_not_mutate_len(self, nbytes=None):
ids = (os.getpgrp(), os.getsid(0))
buf = self._create_int_buf(nbytes)
save_buf = bytes(buf)
with open("/dev/tty", "rb") as tty:
r = fcntl.ioctl(tty, termios.TIOCGPGRP, buf, False)
self.assertIsInstance(r, bytes)
self.assertEqual(len(r), len(save_buf))
self.assertEqual(bytes(buf), save_buf)
rpgrp = array.array('i', r)[0]
rpgrp = memoryview(r).cast('i')[0]
self.assertIn(rpgrp, ids)
buf = bytes(buf)
with open("/dev/tty", "rb") as tty:
r = fcntl.ioctl(tty, termios.TIOCGPGRP, buf, True)
self.assertIsInstance(r, bytes)
self.assertEqual(len(r), len(save_buf))
self.assertEqual(buf, save_buf)
rpgrp = array.array('i', r)[0]
rpgrp = memoryview(r).cast('i')[0]
self.assertIn(rpgrp, ids)
def test_ioctl_mutate(self):
self._check_ioctl_mutate_len()
self._check_ioctl_not_mutate_len()
def test_ioctl_mutate_1024(self):
# Issue #9758: a mutable buffer of exactly 1024 bytes wouldn't be
# copied back after the system call.
self._check_ioctl_mutate_len(1024)
self._check_ioctl_not_mutate_len(1024)
def test_ioctl_mutate_2048(self):
# Test with a larger buffer, just for the record.
self._check_ioctl_mutate_len(2048)
self.assertRaises(ValueError, self._check_ioctl_not_mutate_len, 2048)
@unittest.skipUnless(hasattr(os, 'openpty'), "need os.openpty()")
class IoctlTestsPty(unittest.TestCase):
def setUp(self):
self.master_fd, self.slave_fd = os.openpty()
self.addCleanup(os.close, self.slave_fd)
self.addCleanup(os.close, self.master_fd)
@unittest.skipUnless(hasattr(termios, 'TCFLSH'), 'requires termios.TCFLSH')
def test_ioctl_clear_input_or_output(self):
wfd = self.slave_fd
rfd = self.master_fd
# The data is buffered in the input buffer on Linux, and in
# the output buffer on other platforms.
inbuf = sys.platform in ('linux', 'android')
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')
self.assertTrue(write_suspended.wait(support.SHORT_TIMEOUT))
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:
try:
fcntl.ioctl(wfd, termios.TCXONC, termios.TCOOFF)
finally:
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(support.SHORT_TIMEOUT),
'output was not resumed')
self.assertEqual(os.read(rfd, 1024), b'def')
def test_ioctl_signed_unsigned_code_param(self):
if termios.TIOCSWINSZ < 0:
set_winsz_opcode_maybe_neg = termios.TIOCSWINSZ
set_winsz_opcode_pos = termios.TIOCSWINSZ & 0xffffffff
else:
set_winsz_opcode_pos = termios.TIOCSWINSZ
set_winsz_opcode_maybe_neg, = struct.unpack("i",
struct.pack("I", termios.TIOCSWINSZ))
our_winsz = struct.pack("HHHH",80,25,0,0)
# test both with a positive and potentially negative ioctl code
new_winsz = fcntl.ioctl(self.master_fd, set_winsz_opcode_pos, our_winsz)
new_winsz = fcntl.ioctl(self.master_fd, set_winsz_opcode_maybe_neg, our_winsz)
@unittest.skipUnless(hasattr(fcntl, 'FICLONE'), 'need fcntl.FICLONE')
def test_bad_fd(self):
# gh-134744: Test error handling
fd = os_helper.make_bad_fd()
with self.assertRaises(OSError):
fcntl.ioctl(fd, fcntl.FICLONE, fd)
with self.assertRaises(OSError):
fcntl.ioctl(fd, fcntl.FICLONE, b'\0' * 1024)
if __name__ == "__main__":
unittest.main()
|
