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