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