Python code coverage for Lib/test/test_pty.py

#	count	content
1	n/a	from test.support import verbose, import_module, reap_children
2	n/a
3	n/a	# Skip these tests if termios is not available
4	n/a	import_module('termios')
5	n/a
6	n/a	import errno
7	n/a	import pty
8	n/a	import os
9	n/a	import sys
10	n/a	import select
11	n/a	import signal
12	n/a	import socket
13	n/a	import unittest
14	n/a
15	n/a	TEST_STRING_1 = b"I wish to buy a fish license.\n"
16	n/a	TEST_STRING_2 = b"For my pet fish, Eric.\n"
17	n/a
18	n/a	if verbose:
19	n/a	def debug(msg):
20	n/a	print(msg)
21	n/a	else:
22	n/a	def debug(msg):
23	n/a	pass
24	n/a
25	n/a
26	n/a	def normalize_output(data):
27	n/a	# Some operating systems do conversions on newline. We could possibly
28	n/a	# fix that by doing the appropriate termios.tcsetattr()s. I couldn't
29	n/a	# figure out the right combo on Tru64 and I don't have an IRIX box.
30	n/a	# So just normalize the output and doc the problem O/Ses by allowing
31	n/a	# certain combinations for some platforms, but avoid allowing other
32	n/a	# differences (like extra whitespace, trailing garbage, etc.)
33	n/a
34	n/a	# This is about the best we can do without getting some feedback
35	n/a	# from someone more knowledgable.
36	n/a
37	n/a	# OSF/1 (Tru64) apparently turns \n into \r\r\n.
38	n/a	if data.endswith(b'\r\r\n'):
39	n/a	return data.replace(b'\r\r\n', b'\n')
40	n/a
41	n/a	# IRIX apparently turns \n into \r\n.
42	n/a	if data.endswith(b'\r\n'):
43	n/a	return data.replace(b'\r\n', b'\n')
44	n/a
45	n/a	return data
46	n/a
47	n/a
48	n/a	# Marginal testing of pty suite. Cannot do extensive 'do or fail' testing
49	n/a	# because pty code is not too portable.
50	n/a	# XXX(nnorwitz): these tests leak fds when there is an error.
51	n/a	class PtyTest(unittest.TestCase):
52	n/a	def setUp(self):
53	n/a	# isatty() and close() can hang on some platforms. Set an alarm
54	n/a	# before running the test to make sure we don't hang forever.
55	n/a	self.old_alarm = signal.signal(signal.SIGALRM, self.handle_sig)
56	n/a	signal.alarm(10)
57	n/a
58	n/a	def tearDown(self):
59	n/a	# remove alarm, restore old alarm handler
60	n/a	signal.alarm(0)
61	n/a	signal.signal(signal.SIGALRM, self.old_alarm)
62	n/a
63	n/a	def handle_sig(self, sig, frame):
64	n/a	self.fail("isatty hung")
65	n/a
66	n/a	def test_basic(self):
67	n/a	try:
68	n/a	debug("Calling master_open()")
69	n/a	master_fd, slave_name = pty.master_open()
70	n/a	debug("Got master_fd '%d', slave_name '%s'" %
71	n/a	(master_fd, slave_name))
72	n/a	debug("Calling slave_open(%r)" % (slave_name,))
73	n/a	slave_fd = pty.slave_open(slave_name)
74	n/a	debug("Got slave_fd '%d'" % slave_fd)
75	n/a	except OSError:
76	n/a	# " An optional feature could not be imported " ... ?
77	n/a	raise unittest.SkipTest("Pseudo-terminals (seemingly) not functional.")
78	n/a
79	n/a	self.assertTrue(os.isatty(slave_fd), 'slave_fd is not a tty')
80	n/a
81	n/a	# Solaris requires reading the fd before anything is returned.
82	n/a	# My guess is that since we open and close the slave fd
83	n/a	# in master_open(), we need to read the EOF.
84	n/a
85	n/a	# Ensure the fd is non-blocking in case there's nothing to read.
86	n/a	blocking = os.get_blocking(master_fd)
87	n/a	try:
88	n/a	os.set_blocking(master_fd, False)
89	n/a	try:
90	n/a	s1 = os.read(master_fd, 1024)
91	n/a	self.assertEqual(b'', s1)
92	n/a	except OSError as e:
93	n/a	if e.errno != errno.EAGAIN:
94	n/a	raise
95	n/a	finally:
96	n/a	# Restore the original flags.
97	n/a	os.set_blocking(master_fd, blocking)
98	n/a
99	n/a	debug("Writing to slave_fd")
100	n/a	os.write(slave_fd, TEST_STRING_1)
101	n/a	s1 = os.read(master_fd, 1024)
102	n/a	self.assertEqual(b'I wish to buy a fish license.\n',
103	n/a	normalize_output(s1))
104	n/a
105	n/a	debug("Writing chunked output")
106	n/a	os.write(slave_fd, TEST_STRING_2[:5])
107	n/a	os.write(slave_fd, TEST_STRING_2[5:])
108	n/a	s2 = os.read(master_fd, 1024)
109	n/a	self.assertEqual(b'For my pet fish, Eric.\n', normalize_output(s2))
110	n/a
111	n/a	os.close(slave_fd)
112	n/a	os.close(master_fd)
113	n/a
114	n/a
115	n/a	def test_fork(self):
116	n/a	debug("calling pty.fork()")
117	n/a	pid, master_fd = pty.fork()
118	n/a	if pid == pty.CHILD:
119	n/a	# stdout should be connected to a tty.
120	n/a	if not os.isatty(1):
121	n/a	debug("Child's fd 1 is not a tty?!")
122	n/a	os._exit(3)
123	n/a
124	n/a	# After pty.fork(), the child should already be a session leader.
125	n/a	# (on those systems that have that concept.)
126	n/a	debug("In child, calling os.setsid()")
127	n/a	try:
128	n/a	os.setsid()
129	n/a	except OSError:
130	n/a	# Good, we already were session leader
131	n/a	debug("Good: OSError was raised.")
132	n/a	pass
133	n/a	except AttributeError:
134	n/a	# Have pty, but not setsid()?
135	n/a	debug("No setsid() available?")
136	n/a	pass
137	n/a	except:
138	n/a	# We don't want this error to propagate, escaping the call to
139	n/a	# os._exit() and causing very peculiar behavior in the calling
140	n/a	# regrtest.py !
141	n/a	# Note: could add traceback printing here.
142	n/a	debug("An unexpected error was raised.")
143	n/a	os._exit(1)
144	n/a	else:
145	n/a	debug("os.setsid() succeeded! (bad!)")
146	n/a	os._exit(2)
147	n/a	os._exit(4)
148	n/a	else:
149	n/a	debug("Waiting for child (%d) to finish." % pid)
150	n/a	# In verbose mode, we have to consume the debug output from the
151	n/a	# child or the child will block, causing this test to hang in the
152	n/a	# parent's waitpid() call. The child blocks after a
153	n/a	# platform-dependent amount of data is written to its fd. On
154	n/a	# Linux 2.6, it's 4000 bytes and the child won't block, but on OS
155	n/a	# X even the small writes in the child above will block it. Also
156	n/a	# on Linux, the read() will raise an OSError (input/output error)
157	n/a	# when it tries to read past the end of the buffer but the child's
158	n/a	# already exited, so catch and discard those exceptions. It's not
159	n/a	# worth checking for EIO.
160	n/a	while True:
161	n/a	try:
162	n/a	data = os.read(master_fd, 80)
163	n/a	except OSError:
164	n/a	break
165	n/a	if not data:
166	n/a	break
167	n/a	sys.stdout.write(str(data.replace(b'\r\n', b'\n'),
168	n/a	encoding='ascii'))
169	n/a
170	n/a	##line = os.read(master_fd, 80)
171	n/a	##lines = line.replace('\r\n', '\n').split('\n')
172	n/a	##if False and lines != ['In child, calling os.setsid()',
173	n/a	## 'Good: OSError was raised.', '']:
174	n/a	## raise TestFailed("Unexpected output from child: %r" % line)
175	n/a
176	n/a	(pid, status) = os.waitpid(pid, 0)
177	n/a	res = status >> 8
178	n/a	debug("Child (%d) exited with status %d (%d)." % (pid, res, status))
179	n/a	if res == 1:
180	n/a	self.fail("Child raised an unexpected exception in os.setsid()")
181	n/a	elif res == 2:
182	n/a	self.fail("pty.fork() failed to make child a session leader.")
183	n/a	elif res == 3:
184	n/a	self.fail("Child spawned by pty.fork() did not have a tty as stdout")
185	n/a	elif res != 4:
186	n/a	self.fail("pty.fork() failed for unknown reasons.")
187	n/a
188	n/a	##debug("Reading from master_fd now that the child has exited")
189	n/a	##try:
190	n/a	## s1 = os.read(master_fd, 1024)
191	n/a	##except OSError:
192	n/a	## pass
193	n/a	##else:
194	n/a	## raise TestFailed("Read from master_fd did not raise exception")
195	n/a
196	n/a	os.close(master_fd)
197	n/a
198	n/a	# pty.fork() passed.
199	n/a
200	n/a
201	n/a	class SmallPtyTests(unittest.TestCase):
202	n/a	"""These tests don't spawn children or hang."""
203	n/a
204	n/a	def setUp(self):
205	n/a	self.orig_stdin_fileno = pty.STDIN_FILENO
206	n/a	self.orig_stdout_fileno = pty.STDOUT_FILENO
207	n/a	self.orig_pty_select = pty.select
208	n/a	self.fds = [] # A list of file descriptors to close.
209	n/a	self.files = []
210	n/a	self.select_rfds_lengths = []
211	n/a	self.select_rfds_results = []
212	n/a
213	n/a	def tearDown(self):
214	n/a	pty.STDIN_FILENO = self.orig_stdin_fileno
215	n/a	pty.STDOUT_FILENO = self.orig_stdout_fileno
216	n/a	pty.select = self.orig_pty_select
217	n/a	for file in self.files:
218	n/a	try:
219	n/a	file.close()
220	n/a	except OSError:
221	n/a	pass
222	n/a	for fd in self.fds:
223	n/a	try:
224	n/a	os.close(fd)
225	n/a	except OSError:
226	n/a	pass
227	n/a
228	n/a	def _pipe(self):
229	n/a	pipe_fds = os.pipe()
230	n/a	self.fds.extend(pipe_fds)
231	n/a	return pipe_fds
232	n/a
233	n/a	def _socketpair(self):
234	n/a	socketpair = socket.socketpair()
235	n/a	self.files.extend(socketpair)
236	n/a	return socketpair
237	n/a
238	n/a	def _mock_select(self, rfds, wfds, xfds):
239	n/a	# This will raise IndexError when no more expected calls exist.
240	n/a	self.assertEqual(self.select_rfds_lengths.pop(0), len(rfds))
241	n/a	return self.select_rfds_results.pop(0), [], []
242	n/a
243	n/a	def test__copy_to_each(self):
244	n/a	"""Test the normal data case on both master_fd and stdin."""
245	n/a	read_from_stdout_fd, mock_stdout_fd = self._pipe()
246	n/a	pty.STDOUT_FILENO = mock_stdout_fd
247	n/a	mock_stdin_fd, write_to_stdin_fd = self._pipe()
248	n/a	pty.STDIN_FILENO = mock_stdin_fd
249	n/a	socketpair = self._socketpair()
250	n/a	masters = [s.fileno() for s in socketpair]
251	n/a
252	n/a	# Feed data. Smaller than PIPEBUF. These writes will not block.
253	n/a	os.write(masters[1], b'from master')
254	n/a	os.write(write_to_stdin_fd, b'from stdin')
255	n/a
256	n/a	# Expect two select calls, the last one will cause IndexError
257	n/a	pty.select = self._mock_select
258	n/a	self.select_rfds_lengths.append(2)
259	n/a	self.select_rfds_results.append([mock_stdin_fd, masters[0]])
260	n/a	self.select_rfds_lengths.append(2)
261	n/a
262	n/a	with self.assertRaises(IndexError):
263	n/a	pty._copy(masters[0])
264	n/a
265	n/a	# Test that the right data went to the right places.
266	n/a	rfds = select.select([read_from_stdout_fd, masters[1]], [], [], 0)[0]
267	n/a	self.assertEqual([read_from_stdout_fd, masters[1]], rfds)
268	n/a	self.assertEqual(os.read(read_from_stdout_fd, 20), b'from master')
269	n/a	self.assertEqual(os.read(masters[1], 20), b'from stdin')
270	n/a
271	n/a	def test__copy_eof_on_all(self):
272	n/a	"""Test the empty read EOF case on both master_fd and stdin."""
273	n/a	read_from_stdout_fd, mock_stdout_fd = self._pipe()
274	n/a	pty.STDOUT_FILENO = mock_stdout_fd
275	n/a	mock_stdin_fd, write_to_stdin_fd = self._pipe()
276	n/a	pty.STDIN_FILENO = mock_stdin_fd
277	n/a	socketpair = self._socketpair()
278	n/a	masters = [s.fileno() for s in socketpair]
279	n/a
280	n/a	socketpair[1].close()
281	n/a	os.close(write_to_stdin_fd)
282	n/a
283	n/a	# Expect two select calls, the last one will cause IndexError
284	n/a	pty.select = self._mock_select
285	n/a	self.select_rfds_lengths.append(2)
286	n/a	self.select_rfds_results.append([mock_stdin_fd, masters[0]])
287	n/a	# We expect that both fds were removed from the fds list as they
288	n/a	# both encountered an EOF before the second select call.
289	n/a	self.select_rfds_lengths.append(0)
290	n/a
291	n/a	with self.assertRaises(IndexError):
292	n/a	pty._copy(masters[0])
293	n/a
294	n/a
295	n/a	def tearDownModule():
296	n/a	reap_children()
297	n/a
298	n/a	if __name__ == "__main__":
299	n/a	unittest.main()