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

Python code coverage for Lib/test/test_file_eintr.py

#countcontent
1n/a# Written to test interrupted system calls interfering with our many buffered
2n/a# IO implementations. http://bugs.python.org/issue12268
3n/a#
4n/a# It was suggested that this code could be merged into test_io and the tests
5n/a# made to work using the same method as the existing signal tests in test_io.
6n/a# I was unable to get single process tests using alarm or setitimer that way
7n/a# to reproduce the EINTR problems. This process based test suite reproduces
8n/a# the problems prior to the issue12268 patch reliably on Linux and OSX.
9n/a# - gregory.p.smith
10n/a
11n/aimport os
12n/aimport select
13n/aimport signal
14n/aimport subprocess
15n/aimport sys
16n/aimport time
17n/aimport unittest
18n/a
19n/a# Test import all of the things we're about to try testing up front.
20n/aimport _io
21n/aimport _pyio
22n/a
23n/a
24n/a@unittest.skipUnless(os.name == 'posix', 'tests requires a posix system.')
25n/aclass TestFileIOSignalInterrupt:
26n/a def setUp(self):
27n/a self._process = None
28n/a
29n/a def tearDown(self):
30n/a if self._process and self._process.poll() is None:
31n/a try:
32n/a self._process.kill()
33n/a except OSError:
34n/a pass
35n/a
36n/a def _generate_infile_setup_code(self):
37n/a """Returns the infile = ... line of code for the reader process.
38n/a
39n/a subclasseses should override this to test different IO objects.
40n/a """
41n/a return ('import %s as io ;'
42n/a 'infile = io.FileIO(sys.stdin.fileno(), "rb")' %
43n/a self.modname)
44n/a
45n/a def fail_with_process_info(self, why, stdout=b'', stderr=b'',
46n/a communicate=True):
47n/a """A common way to cleanup and fail with useful debug output.
48n/a
49n/a Kills the process if it is still running, collects remaining output
50n/a and fails the test with an error message including the output.
51n/a
52n/a Args:
53n/a why: Text to go after "Error from IO process" in the message.
54n/a stdout, stderr: standard output and error from the process so
55n/a far to include in the error message.
56n/a communicate: bool, when True we call communicate() on the process
57n/a after killing it to gather additional output.
58n/a """
59n/a if self._process.poll() is None:
60n/a time.sleep(0.1) # give it time to finish printing the error.
61n/a try:
62n/a self._process.terminate() # Ensure it dies.
63n/a except OSError:
64n/a pass
65n/a if communicate:
66n/a stdout_end, stderr_end = self._process.communicate()
67n/a stdout += stdout_end
68n/a stderr += stderr_end
69n/a self.fail('Error from IO process %s:\nSTDOUT:\n%sSTDERR:\n%s\n' %
70n/a (why, stdout.decode(), stderr.decode()))
71n/a
72n/a def _test_reading(self, data_to_write, read_and_verify_code):
73n/a """Generic buffered read method test harness to validate EINTR behavior.
74n/a
75n/a Also validates that Python signal handlers are run during the read.
76n/a
77n/a Args:
78n/a data_to_write: String to write to the child process for reading
79n/a before sending it a signal, confirming the signal was handled,
80n/a writing a final newline and closing the infile pipe.
81n/a read_and_verify_code: Single "line" of code to read from a file
82n/a object named 'infile' and validate the result. This will be
83n/a executed as part of a python subprocess fed data_to_write.
84n/a """
85n/a infile_setup_code = self._generate_infile_setup_code()
86n/a # Total pipe IO in this function is smaller than the minimum posix OS
87n/a # pipe buffer size of 512 bytes. No writer should block.
88n/a assert len(data_to_write) < 512, 'data_to_write must fit in pipe buf.'
89n/a
90n/a # Start a subprocess to call our read method while handling a signal.
91n/a self._process = subprocess.Popen(
92n/a [sys.executable, '-u', '-c',
93n/a 'import signal, sys ;'
94n/a 'signal.signal(signal.SIGINT, '
95n/a 'lambda s, f: sys.stderr.write("$\\n")) ;'
96n/a + infile_setup_code + ' ;' +
97n/a 'sys.stderr.write("Worm Sign!\\n") ;'
98n/a + read_and_verify_code + ' ;' +
99n/a 'infile.close()'
100n/a ],
101n/a stdin=subprocess.PIPE, stdout=subprocess.PIPE,
102n/a stderr=subprocess.PIPE)
103n/a
104n/a # Wait for the signal handler to be installed.
105n/a worm_sign = self._process.stderr.read(len(b'Worm Sign!\n'))
106n/a if worm_sign != b'Worm Sign!\n': # See also, Dune by Frank Herbert.
107n/a self.fail_with_process_info('while awaiting a sign',
108n/a stderr=worm_sign)
109n/a self._process.stdin.write(data_to_write)
110n/a
111n/a signals_sent = 0
112n/a rlist = []
113n/a # We don't know when the read_and_verify_code in our child is actually
114n/a # executing within the read system call we want to interrupt. This
115n/a # loop waits for a bit before sending the first signal to increase
116n/a # the likelihood of that. Implementations without correct EINTR
117n/a # and signal handling usually fail this test.
118n/a while not rlist:
119n/a rlist, _, _ = select.select([self._process.stderr], (), (), 0.05)
120n/a self._process.send_signal(signal.SIGINT)
121n/a signals_sent += 1
122n/a if signals_sent > 200:
123n/a self._process.kill()
124n/a self.fail('reader process failed to handle our signals.')
125n/a # This assumes anything unexpected that writes to stderr will also
126n/a # write a newline. That is true of the traceback printing code.
127n/a signal_line = self._process.stderr.readline()
128n/a if signal_line != b'$\n':
129n/a self.fail_with_process_info('while awaiting signal',
130n/a stderr=signal_line)
131n/a
132n/a # We append a newline to our input so that a readline call can
133n/a # end on its own before the EOF is seen and so that we're testing
134n/a # the read call that was interrupted by a signal before the end of
135n/a # the data stream has been reached.
136n/a stdout, stderr = self._process.communicate(input=b'\n')
137n/a if self._process.returncode:
138n/a self.fail_with_process_info(
139n/a 'exited rc=%d' % self._process.returncode,
140n/a stdout, stderr, communicate=False)
141n/a # PASS!
142n/a
143n/a # String format for the read_and_verify_code used by read methods.
144n/a _READING_CODE_TEMPLATE = (
145n/a 'got = infile.{read_method_name}() ;'
146n/a 'expected = {expected!r} ;'
147n/a 'assert got == expected, ('
148n/a '"{read_method_name} returned wrong data.\\n"'
149n/a '"got data %r\\nexpected %r" % (got, expected))'
150n/a )
151n/a
152n/a def test_readline(self):
153n/a """readline() must handle signals and not lose data."""
154n/a self._test_reading(
155n/a data_to_write=b'hello, world!',
156n/a read_and_verify_code=self._READING_CODE_TEMPLATE.format(
157n/a read_method_name='readline',
158n/a expected=b'hello, world!\n'))
159n/a
160n/a def test_readlines(self):
161n/a """readlines() must handle signals and not lose data."""
162n/a self._test_reading(
163n/a data_to_write=b'hello\nworld!',
164n/a read_and_verify_code=self._READING_CODE_TEMPLATE.format(
165n/a read_method_name='readlines',
166n/a expected=[b'hello\n', b'world!\n']))
167n/a
168n/a def test_readall(self):
169n/a """readall() must handle signals and not lose data."""
170n/a self._test_reading(
171n/a data_to_write=b'hello\nworld!',
172n/a read_and_verify_code=self._READING_CODE_TEMPLATE.format(
173n/a read_method_name='readall',
174n/a expected=b'hello\nworld!\n'))
175n/a # read() is the same thing as readall().
176n/a self._test_reading(
177n/a data_to_write=b'hello\nworld!',
178n/a read_and_verify_code=self._READING_CODE_TEMPLATE.format(
179n/a read_method_name='read',
180n/a expected=b'hello\nworld!\n'))
181n/a
182n/a
183n/aclass CTestFileIOSignalInterrupt(TestFileIOSignalInterrupt, unittest.TestCase):
184n/a modname = '_io'
185n/a
186n/aclass PyTestFileIOSignalInterrupt(TestFileIOSignalInterrupt, unittest.TestCase):
187n/a modname = '_pyio'
188n/a
189n/a
190n/aclass TestBufferedIOSignalInterrupt(TestFileIOSignalInterrupt):
191n/a def _generate_infile_setup_code(self):
192n/a """Returns the infile = ... line of code to make a BufferedReader."""
193n/a return ('import %s as io ;infile = io.open(sys.stdin.fileno(), "rb") ;'
194n/a 'assert isinstance(infile, io.BufferedReader)' %
195n/a self.modname)
196n/a
197n/a def test_readall(self):
198n/a """BufferedReader.read() must handle signals and not lose data."""
199n/a self._test_reading(
200n/a data_to_write=b'hello\nworld!',
201n/a read_and_verify_code=self._READING_CODE_TEMPLATE.format(
202n/a read_method_name='read',
203n/a expected=b'hello\nworld!\n'))
204n/a
205n/aclass CTestBufferedIOSignalInterrupt(TestBufferedIOSignalInterrupt, unittest.TestCase):
206n/a modname = '_io'
207n/a
208n/aclass PyTestBufferedIOSignalInterrupt(TestBufferedIOSignalInterrupt, unittest.TestCase):
209n/a modname = '_pyio'
210n/a
211n/a
212n/aclass TestTextIOSignalInterrupt(TestFileIOSignalInterrupt):
213n/a def _generate_infile_setup_code(self):
214n/a """Returns the infile = ... line of code to make a TextIOWrapper."""
215n/a return ('import %s as io ;'
216n/a 'infile = io.open(sys.stdin.fileno(), "rt", newline=None) ;'
217n/a 'assert isinstance(infile, io.TextIOWrapper)' %
218n/a self.modname)
219n/a
220n/a def test_readline(self):
221n/a """readline() must handle signals and not lose data."""
222n/a self._test_reading(
223n/a data_to_write=b'hello, world!',
224n/a read_and_verify_code=self._READING_CODE_TEMPLATE.format(
225n/a read_method_name='readline',
226n/a expected='hello, world!\n'))
227n/a
228n/a def test_readlines(self):
229n/a """readlines() must handle signals and not lose data."""
230n/a self._test_reading(
231n/a data_to_write=b'hello\r\nworld!',
232n/a read_and_verify_code=self._READING_CODE_TEMPLATE.format(
233n/a read_method_name='readlines',
234n/a expected=['hello\n', 'world!\n']))
235n/a
236n/a def test_readall(self):
237n/a """read() must handle signals and not lose data."""
238n/a self._test_reading(
239n/a data_to_write=b'hello\nworld!',
240n/a read_and_verify_code=self._READING_CODE_TEMPLATE.format(
241n/a read_method_name='read',
242n/a expected="hello\nworld!\n"))
243n/a
244n/aclass CTestTextIOSignalInterrupt(TestTextIOSignalInterrupt, unittest.TestCase):
245n/a modname = '_io'
246n/a
247n/aclass PyTestTextIOSignalInterrupt(TestTextIOSignalInterrupt, unittest.TestCase):
248n/a modname = '_pyio'
249n/a
250n/a
251n/aif __name__ == '__main__':
252n/a unittest.main()