ยปCore Development>Code coverage>Lib/test/test_pty.py

Python code coverage for Lib/test/test_pty.py

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