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