Lines Matching refs:self
82 def setUp(self):
83 old_alarm = signal.signal(signal.SIGALRM, self.handle_sig)
84 self.addCleanup(signal.signal, signal.SIGALRM, old_alarm)
86 old_sighup = signal.signal(signal.SIGHUP, self.handle_sighup)
87 self.addCleanup(signal.signal, signal.SIGHUP, old_sighup)
91 self.addCleanup(signal.alarm, 0)
95 self.stdin_dim = None
98 self.stdin_dim = tty.tcgetwinsize(pty.STDIN_FILENO)
99 self.addCleanup(tty.tcsetwinsize, pty.STDIN_FILENO,
100 self.stdin_dim)
104 def handle_sig(self, sig, frame):
105 self.fail("isatty hung")
112 def test_openpty(self):
121 if self.stdin_dim:
127 debug(f"original size: (row, col) = {self.stdin_dim}")
128 target_dim = (self.stdin_dim[0] + 1, self.stdin_dim[1] + 1)
135 self.assertEqual(new_dim, target_dim,
157 self.addCleanup(os.close, master_fd)
158 self.addCleanup(os.close, slave_fd)
160 self.assertTrue(os.isatty(slave_fd), "slave_fd is not a tty")
163 self.assertEqual(tty.tcgetattr(slave_fd), mode,
166 self.assertEqual(tty.tcgetwinsize(slave_fd), new_dim,
175 self.assertEqual(b'', s1)
186 self.assertEqual(b'I wish to buy a fish license.\n',
193 self.assertEqual(b'For my pet fish, Eric.\n', normalize_output(s2))
195 def test_fork(self):
198 self.addCleanup(os.close, master_fd)
261 self.fail("Child raised an unexpected exception in os.setsid()")
263 self.fail("pty.fork() failed to make child a session leader.")
265 self.fail("Child spawned by pty.fork() did not have a tty as stdout")
267 self.fail("pty.fork() failed for unknown reasons.")
277 def test_master_read(self):
283 self.addCleanup(os.close, master_fd)
294 self.assertEqual(data, b"")
296 def test_spawn_doesnt_hang(self):
302 def setUp(self):
303 self.orig_stdin_fileno = pty.STDIN_FILENO
304 self.orig_stdout_fileno = pty.STDOUT_FILENO
305 self.orig_pty_close = pty.close
306 self.orig_pty__copy = pty._copy
307 self.orig_pty_fork = pty.fork
308 self.orig_pty_select = pty.select
309 self.orig_pty_setraw = pty.setraw
310 self.orig_pty_tcgetattr = pty.tcgetattr
311 self.orig_pty_tcsetattr = pty.tcsetattr
312 self.orig_pty_waitpid = pty.waitpid
313 self.fds = [] # A list of file descriptors to close.
314 self.files = []
315 self.select_input = []
316 self.select_output = []
317 self.tcsetattr_mode_setting = None
319 def tearDown(self):
320 pty.STDIN_FILENO = self.orig_stdin_fileno
321 pty.STDOUT_FILENO = self.orig_stdout_fileno
322 pty.close = self.orig_pty_close
323 pty._copy = self.orig_pty__copy
324 pty.fork = self.orig_pty_fork
325 pty.select = self.orig_pty_select
326 pty.setraw = self.orig_pty_setraw
327 pty.tcgetattr = self.orig_pty_tcgetattr
328 pty.tcsetattr = self.orig_pty_tcsetattr
329 pty.waitpid = self.orig_pty_waitpid
330 for file in self.files:
335 for fd in self.fds:
341 def _pipe(self):
343 self.fds.extend(pipe_fds)
346 def _socketpair(self):
348 self.files.extend(socketpair)
351 def _mock_select(self, rfds, wfds, xfds):
353 self.assertEqual((rfds, wfds, xfds), self.select_input.pop(0))
354 return self.select_output.pop(0)
356 def _make_mock_fork(self, pid):
361 def _mock_tcsetattr(self, fileno, opt, mode):
362 self.tcsetattr_mode_setting = mode
364 def test__copy_to_each(self):
366 read_from_stdout_fd, mock_stdout_fd = self._pipe()
368 mock_stdin_fd, write_to_stdin_fd = self._pipe()
370 socketpair = self._socketpair()
378 pty.select = self._mock_select
379 self.select_input.append(([mock_stdin_fd, masters[0]], [], []))
380 self.select_output.append(([mock_stdin_fd, masters[0]], [], []))
381 self.select_input.append(([mock_stdin_fd, masters[0]], [mock_stdout_fd, masters[0]], []))
382 self.select_output.append(([], [mock_stdout_fd, masters[0]], []))
383 self.select_input.append(([mock_stdin_fd, masters[0]], [], []))
385 with self.assertRaises(IndexError):
390 self.assertEqual([read_from_stdout_fd, masters[1]], rfds)
391 self.assertEqual(os.read(read_from_stdout_fd, 20), b'from master')
392 self.assertEqual(os.read(masters[1], 20), b'from stdin')
394 def test__restore_tty_mode_normal_return(self):
399 pty.fork = self._make_mock_fork(1)
409 pty.tcsetattr = self._mock_tcsetattr
412 self.assertEqual(pty.spawn([]), status_sentinel, "pty.waitpid process status not returned by pty.spawn")
413 self.assertEqual(self.tcsetattr_mode_setting, mode_sentinel, "pty.tcsetattr not called with original mode value")