blob: ef95268e19e593012a1e6be5a2e922cb73cbf9e7 [file] [log] [blame]
Antoine Pitrou8189ab82011-03-20 17:35:32 +01001from test.support import verbose, run_unittest, import_module, reap_children
R. David Murrayeb3615d2009-04-22 02:24:39 +00002
3#Skip these tests if either fcntl or termios is not available
4fcntl = import_module('fcntl')
5import_module('termios')
6
Guido van Rossum360e4b82007-05-14 22:51:27 +00007import errno
R. David Murrayeb3615d2009-04-22 02:24:39 +00008import pty
Guido van Rossumd8faa362007-04-27 19:54:29 +00009import os
10import sys
Gregory P. Smith05f59532012-02-16 00:29:12 -080011import select
Guido van Rossumd8faa362007-04-27 19:54:29 +000012import signal
Gregory P. Smith05f59532012-02-16 00:29:12 -080013import socket
Guido van Rossumd8faa362007-04-27 19:54:29 +000014import unittest
Fred Drake4c136ee2000-06-30 23:22:35 +000015
Antoine Pitrou9cadb1b2008-09-15 23:02:56 +000016TEST_STRING_1 = b"I wish to buy a fish license.\n"
17TEST_STRING_2 = b"For my pet fish, Eric.\n"
Fred Drake4c136ee2000-06-30 23:22:35 +000018
19if verbose:
20 def debug(msg):
Guido van Rossumbe19ed72007-02-09 05:37:30 +000021 print(msg)
Fred Drake4c136ee2000-06-30 23:22:35 +000022else:
23 def debug(msg):
24 pass
25
Guido van Rossumd8faa362007-04-27 19:54:29 +000026
Thomas Wouters49fd7fa2006-04-21 10:40:58 +000027def normalize_output(data):
28 # Some operating systems do conversions on newline. We could possibly
29 # fix that by doing the appropriate termios.tcsetattr()s. I couldn't
30 # figure out the right combo on Tru64 and I don't have an IRIX box.
31 # So just normalize the output and doc the problem O/Ses by allowing
32 # certain combinations for some platforms, but avoid allowing other
33 # differences (like extra whitespace, trailing garbage, etc.)
34
35 # This is about the best we can do without getting some feedback
36 # from someone more knowledgable.
37
38 # OSF/1 (Tru64) apparently turns \n into \r\r\n.
Walter Dörwald812d8342007-05-29 18:57:42 +000039 if data.endswith(b'\r\r\n'):
40 return data.replace(b'\r\r\n', b'\n')
Thomas Wouters49fd7fa2006-04-21 10:40:58 +000041
42 # IRIX apparently turns \n into \r\n.
Walter Dörwald812d8342007-05-29 18:57:42 +000043 if data.endswith(b'\r\n'):
44 return data.replace(b'\r\n', b'\n')
Thomas Wouters49fd7fa2006-04-21 10:40:58 +000045
46 return data
47
Guido van Rossumd8faa362007-04-27 19:54:29 +000048
Fred Drake4c136ee2000-06-30 23:22:35 +000049# Marginal testing of pty suite. Cannot do extensive 'do or fail' testing
50# because pty code is not too portable.
Guido van Rossum360e4b82007-05-14 22:51:27 +000051# XXX(nnorwitz): these tests leak fds when there is an error.
Guido van Rossumd8faa362007-04-27 19:54:29 +000052class PtyTest(unittest.TestCase):
53 def setUp(self):
54 # isatty() and close() can hang on some platforms. Set an alarm
55 # before running the test to make sure we don't hang forever.
56 self.old_alarm = signal.signal(signal.SIGALRM, self.handle_sig)
57 signal.alarm(10)
Fred Drake4c136ee2000-06-30 23:22:35 +000058
Guido van Rossumd8faa362007-04-27 19:54:29 +000059 def tearDown(self):
60 # remove alarm, restore old alarm handler
61 signal.alarm(0)
62 signal.signal(signal.SIGALRM, self.old_alarm)
Neal Norwitz7d814522003-03-21 01:39:14 +000063
Guido van Rossumd8faa362007-04-27 19:54:29 +000064 def handle_sig(self, sig, frame):
65 self.fail("isatty hung")
Neal Norwitz7d814522003-03-21 01:39:14 +000066
Guido van Rossumd8faa362007-04-27 19:54:29 +000067 def test_basic(self):
68 try:
69 debug("Calling master_open()")
70 master_fd, slave_name = pty.master_open()
71 debug("Got master_fd '%d', slave_name '%s'" %
72 (master_fd, slave_name))
73 debug("Calling slave_open(%r)" % (slave_name,))
74 slave_fd = pty.slave_open(slave_name)
75 debug("Got slave_fd '%d'" % slave_fd)
76 except OSError:
77 # " An optional feature could not be imported " ... ?
Benjamin Petersone549ead2009-03-28 21:42:05 +000078 raise unittest.SkipTest("Pseudo-terminals (seemingly) not functional.")
Neal Norwitz7d814522003-03-21 01:39:14 +000079
Guido van Rossumd8faa362007-04-27 19:54:29 +000080 self.assertTrue(os.isatty(slave_fd), 'slave_fd is not a tty')
Neal Norwitz7d814522003-03-21 01:39:14 +000081
Guido van Rossum360e4b82007-05-14 22:51:27 +000082 # Solaris requires reading the fd before anything is returned.
83 # My guess is that since we open and close the slave fd
84 # in master_open(), we need to read the EOF.
85
86 # Ensure the fd is non-blocking in case there's nothing to read.
87 orig_flags = fcntl.fcntl(master_fd, fcntl.F_GETFL)
88 fcntl.fcntl(master_fd, fcntl.F_SETFL, orig_flags | os.O_NONBLOCK)
89 try:
90 s1 = os.read(master_fd, 1024)
Ezio Melotti19f2aeb2010-11-21 01:30:29 +000091 self.assertEqual(b'', s1)
Guido van Rossum360e4b82007-05-14 22:51:27 +000092 except OSError as e:
93 if e.errno != errno.EAGAIN:
94 raise
95 # Restore the original flags.
96 fcntl.fcntl(master_fd, fcntl.F_SETFL, orig_flags)
97
Guido van Rossumd8faa362007-04-27 19:54:29 +000098 debug("Writing to slave_fd")
99 os.write(slave_fd, TEST_STRING_1)
100 s1 = os.read(master_fd, 1024)
Ezio Melotti19f2aeb2010-11-21 01:30:29 +0000101 self.assertEqual(b'I wish to buy a fish license.\n',
102 normalize_output(s1))
Neal Norwitz7d814522003-03-21 01:39:14 +0000103
Guido van Rossumd8faa362007-04-27 19:54:29 +0000104 debug("Writing chunked output")
105 os.write(slave_fd, TEST_STRING_2[:5])
106 os.write(slave_fd, TEST_STRING_2[5:])
107 s2 = os.read(master_fd, 1024)
Ezio Melotti19f2aeb2010-11-21 01:30:29 +0000108 self.assertEqual(b'For my pet fish, Eric.\n', normalize_output(s2))
Neal Norwitz7d814522003-03-21 01:39:14 +0000109
Guido van Rossumd8faa362007-04-27 19:54:29 +0000110 os.close(slave_fd)
111 os.close(master_fd)
Thomas Wouters9fe394c2007-02-05 01:24:16 +0000112
113
Guido van Rossumd8faa362007-04-27 19:54:29 +0000114 def test_fork(self):
115 debug("calling pty.fork()")
116 pid, master_fd = pty.fork()
117 if pid == pty.CHILD:
118 # stdout should be connected to a tty.
119 if not os.isatty(1):
120 debug("Child's fd 1 is not a tty?!")
121 os._exit(3)
Fred Drake4c136ee2000-06-30 23:22:35 +0000122
Guido van Rossumd8faa362007-04-27 19:54:29 +0000123 # After pty.fork(), the child should already be a session leader.
124 # (on those systems that have that concept.)
125 debug("In child, calling os.setsid()")
126 try:
127 os.setsid()
128 except OSError:
129 # Good, we already were session leader
130 debug("Good: OSError was raised.")
131 pass
132 except AttributeError:
133 # Have pty, but not setsid()?
134 debug("No setsid() available?")
135 pass
136 except:
137 # We don't want this error to propagate, escaping the call to
138 # os._exit() and causing very peculiar behavior in the calling
139 # regrtest.py !
140 # Note: could add traceback printing here.
141 debug("An unexpected error was raised.")
142 os._exit(1)
143 else:
144 debug("os.setsid() succeeded! (bad!)")
145 os._exit(2)
146 os._exit(4)
147 else:
148 debug("Waiting for child (%d) to finish." % pid)
149 # In verbose mode, we have to consume the debug output from the
150 # child or the child will block, causing this test to hang in the
151 # parent's waitpid() call. The child blocks after a
152 # platform-dependent amount of data is written to its fd. On
153 # Linux 2.6, it's 4000 bytes and the child won't block, but on OS
154 # X even the small writes in the child above will block it. Also
155 # on Linux, the read() will throw an OSError (input/output error)
156 # when it tries to read past the end of the buffer but the child's
157 # already exited, so catch and discard those exceptions. It's not
158 # worth checking for EIO.
159 while True:
160 try:
161 data = os.read(master_fd, 80)
162 except OSError:
163 break
164 if not data:
165 break
Alexandre Vassalottia351f772008-03-03 02:59:49 +0000166 sys.stdout.write(str(data.replace(b'\r\n', b'\n'),
167 encoding='ascii'))
Guido van Rossumd8faa362007-04-27 19:54:29 +0000168
169 ##line = os.read(master_fd, 80)
170 ##lines = line.replace('\r\n', '\n').split('\n')
171 ##if False and lines != ['In child, calling os.setsid()',
172 ## 'Good: OSError was raised.', '']:
173 ## raise TestFailed("Unexpected output from child: %r" % line)
174
175 (pid, status) = os.waitpid(pid, 0)
176 res = status >> 8
177 debug("Child (%d) exited with status %d (%d)." % (pid, res, status))
178 if res == 1:
179 self.fail("Child raised an unexpected exception in os.setsid()")
180 elif res == 2:
181 self.fail("pty.fork() failed to make child a session leader.")
182 elif res == 3:
183 self.fail("Child spawned by pty.fork() did not have a tty as stdout")
184 elif res != 4:
185 self.fail("pty.fork() failed for unknown reasons.")
186
187 ##debug("Reading from master_fd now that the child has exited")
188 ##try:
189 ## s1 = os.read(master_fd, 1024)
190 ##except os.error:
191 ## pass
192 ##else:
193 ## raise TestFailed("Read from master_fd did not raise exception")
194
195 os.close(master_fd)
196
197 # pty.fork() passed.
198
Gregory P. Smith05f59532012-02-16 00:29:12 -0800199
200class SmallPtyTests(unittest.TestCase):
201 """These tests don't spawn children or hang."""
202
203 def setUp(self):
204 self.orig_stdin_fileno = pty.STDIN_FILENO
205 self.orig_stdout_fileno = pty.STDOUT_FILENO
206 self.orig_pty_select = pty.select
207 self.fds = [] # A list of file descriptors to close.
Victor Stinnerb1f7f632012-03-06 02:04:58 +0100208 self.files = []
Gregory P. Smith05f59532012-02-16 00:29:12 -0800209 self.select_rfds_lengths = []
210 self.select_rfds_results = []
211
212 def tearDown(self):
213 pty.STDIN_FILENO = self.orig_stdin_fileno
214 pty.STDOUT_FILENO = self.orig_stdout_fileno
215 pty.select = self.orig_pty_select
Victor Stinnerb1f7f632012-03-06 02:04:58 +0100216 for file in self.files:
217 try:
218 file.close()
219 except OSError:
220 pass
Gregory P. Smith05f59532012-02-16 00:29:12 -0800221 for fd in self.fds:
222 try:
223 os.close(fd)
Victor Stinnerb1f7f632012-03-06 02:04:58 +0100224 except OSError:
Gregory P. Smith05f59532012-02-16 00:29:12 -0800225 pass
226
227 def _pipe(self):
228 pipe_fds = os.pipe()
229 self.fds.extend(pipe_fds)
230 return pipe_fds
231
Victor Stinnerb1f7f632012-03-06 02:04:58 +0100232 def _socketpair(self):
233 socketpair = socket.socketpair()
234 self.files.extend(socketpair)
235 return socketpair
236
Gregory P. Smith05f59532012-02-16 00:29:12 -0800237 def _mock_select(self, rfds, wfds, xfds):
238 # This will raise IndexError when no more expected calls exist.
239 self.assertEqual(self.select_rfds_lengths.pop(0), len(rfds))
240 return self.select_rfds_results.pop(0), [], []
241
242 def test__copy_to_each(self):
243 """Test the normal data case on both master_fd and stdin."""
244 read_from_stdout_fd, mock_stdout_fd = self._pipe()
245 pty.STDOUT_FILENO = mock_stdout_fd
246 mock_stdin_fd, write_to_stdin_fd = self._pipe()
247 pty.STDIN_FILENO = mock_stdin_fd
Victor Stinnerb1f7f632012-03-06 02:04:58 +0100248 socketpair = self._socketpair()
Gregory P. Smith05f59532012-02-16 00:29:12 -0800249 masters = [s.fileno() for s in socketpair]
Gregory P. Smith05f59532012-02-16 00:29:12 -0800250
251 # Feed data. Smaller than PIPEBUF. These writes will not block.
252 os.write(masters[1], b'from master')
253 os.write(write_to_stdin_fd, b'from stdin')
254
255 # Expect two select calls, the last one will cause IndexError
256 pty.select = self._mock_select
257 self.select_rfds_lengths.append(2)
258 self.select_rfds_results.append([mock_stdin_fd, masters[0]])
259 self.select_rfds_lengths.append(2)
260
261 with self.assertRaises(IndexError):
262 pty._copy(masters[0])
263
264 # Test that the right data went to the right places.
265 rfds = select.select([read_from_stdout_fd, masters[1]], [], [], 0)[0]
Gregory P. Smith5b791fb2012-02-16 00:35:43 -0800266 self.assertEqual([read_from_stdout_fd, masters[1]], rfds)
Gregory P. Smith05f59532012-02-16 00:29:12 -0800267 self.assertEqual(os.read(read_from_stdout_fd, 20), b'from master')
268 self.assertEqual(os.read(masters[1], 20), b'from stdin')
269
270 def test__copy_eof_on_all(self):
271 """Test the empty read EOF case on both master_fd and stdin."""
272 read_from_stdout_fd, mock_stdout_fd = self._pipe()
273 pty.STDOUT_FILENO = mock_stdout_fd
274 mock_stdin_fd, write_to_stdin_fd = self._pipe()
275 pty.STDIN_FILENO = mock_stdin_fd
Victor Stinnerb1f7f632012-03-06 02:04:58 +0100276 socketpair = self._socketpair()
Gregory P. Smith05f59532012-02-16 00:29:12 -0800277 masters = [s.fileno() for s in socketpair]
Gregory P. Smith05f59532012-02-16 00:29:12 -0800278
279 os.close(masters[1])
280 socketpair[1].close()
281 os.close(write_to_stdin_fd)
282
283 # Expect two select calls, the last one will cause IndexError
284 pty.select = self._mock_select
285 self.select_rfds_lengths.append(2)
286 self.select_rfds_results.append([mock_stdin_fd, masters[0]])
287 # We expect that both fds were removed from the fds list as they
288 # both encountered an EOF before the second select call.
289 self.select_rfds_lengths.append(0)
290
291 with self.assertRaises(IndexError):
292 pty._copy(masters[0])
293
294
Guido van Rossumd8faa362007-04-27 19:54:29 +0000295def test_main(verbose=None):
Antoine Pitrou8189ab82011-03-20 17:35:32 +0100296 try:
Gregory P. Smith05f59532012-02-16 00:29:12 -0800297 run_unittest(SmallPtyTests, PtyTest)
Antoine Pitrou8189ab82011-03-20 17:35:32 +0100298 finally:
299 reap_children()
Guido van Rossumd8faa362007-04-27 19:54:29 +0000300
301if __name__ == "__main__":
302 test_main()