written to test interrupted system calls interfering with our many buffered io implementations http bugs python orgissue12268 it was suggested that this code could be merged into testio and the tests made to work using the same method as the existing signal tests in testio i was unable to get single process tests using alarm or setitimer that way to reproduce the eintr problems this process based test suite reproduces the problems prior to the issue12268 patch reliably on linux and osx gregory p smith test import all of the things we re about to try testing up front returns the infile line of code for the reader process subclasseses should override this to test different io objects a common way to cleanup and fail with useful debug output kills the process if it is still running collects remaining output and fails the test with an error message including the output args why text to go after error from io process in the message stdout stderr standard output and error from the process so far to include in the error message communicate bool when true we call communicate on the process after killing it to gather additional output generic buffered read method test harness to validate eintr behavior also validates that python signal handlers are run during the read args datatowrite string to write to the child process for reading before sending it a signal confirming the signal was handled writing a final newline and closing the infile pipe readandverifycode single line of code to read from a file object named infile and validate the result this will be executed as part of a python subprocess fed datatowrite total pipe io in this function is smaller than the minimum posix os pipe buffer size of 512 bytes no writer should block start a subprocess to call our read method while handling a signal wait for the signal handler to be installed we don t know when the readandverifycode in our child is actually executing within the read system call we want to interrupt this loop waits for a bit before sending the first signal to increase the likelihood of that implementations without correct eintr and signal handling usually fail this test this assumes anything unexpected that writes to stderr will also write a newline that is true of the traceback printing code we append a newline to our input so that a readline call can end on its own before the eof is seen and so that we re testing the read call that was interrupted by a signal before the end of the data stream has been reached pass string format for the readandverifycode used by read methods readline must handle signals and not lose data self testreading datatowriteb hello world readandverifycodeself readingcodetemplate format readmethodname readline expectedb hello world n def testreadlinesself readall must handle signals and not lose data self testreading datatowriteb hellonworld readandverifycodeself readingcodetemplate format readmethodname readall expectedb hellonworld n read is the same thing as readall self testreading datatowriteb hellonworld readandverifycodeself readingcodetemplate format readmethodname read expectedb hellonworld n class ctestfileiosignalinterrupttestfileiosignalinterrupt unittest testcase modname io class pytestfileiosignalinterrupttestfileiosignalinterrupt unittest testcase modname pyio class testbufferediosignalinterrupttestfileiosignalinterrupt def generateinfilesetupcodeself bufferedreader read must handle signals and not lose data self testreading datatowriteb hellonworld readandverifycodeself readingcodetemplate format readmethodname read expectedb hellonworld n class ctestbufferediosignalinterrupttestbufferediosignalinterrupt unittest testcase modname io class pytestbufferediosignalinterrupttestbufferediosignalinterrupt unittest testcase modname pyio class testtextiosignalinterrupttestfileiosignalinterrupt def generateinfilesetupcodeself readline must handle signals and not lose data self testreading datatowriteb hello world readandverifycodeself readingcodetemplate format readmethodname readline expected hello world n def testreadlinesself read must handle signals and not lose data self testreading datatowriteb hellonworld readandverifycodeself readingcodetemplate format readmethodname read expectedhellonworld n class ctesttextiosignalinterrupttesttextiosignalinterrupt unittest testcase modname io class pytesttextiosignalinterrupttesttextiosignalinterrupt unittest testcase modname pyio if name main unittest main written to test interrupted system calls interfering with our many buffered io implementations http bugs python org issue12268 it was suggested that this code could be merged into test_io and the tests made to work using the same method as the existing signal tests in test_io i was unable to get single process tests using alarm or setitimer that way to reproduce the eintr problems this process based test suite reproduces the problems prior to the issue12268 patch reliably on linux and osx gregory p smith test import all of the things we re about to try testing up front returns the infile line of code for the reader process subclasseses should override this to test different io objects a common way to cleanup and fail with useful debug output kills the process if it is still running collects remaining output and fails the test with an error message including the output args why text to go after error from io process in the message stdout stderr standard output and error from the process so far to include in the error message communicate bool when true we call communicate on the process after killing it to gather additional output give it time to finish printing the error ensure it dies generic buffered read method test harness to validate eintr behavior also validates that python signal handlers are run during the read args data_to_write string to write to the child process for reading before sending it a signal confirming the signal was handled writing a final newline and closing the infile pipe read_and_verify_code single line of code to read from a file object named infile and validate the result this will be executed as part of a python subprocess fed data_to_write total pipe io in this function is smaller than the minimum posix os pipe buffer size of 512 bytes no writer should block start a subprocess to call our read method while handling a signal wait for the signal handler to be installed see also dune by frank herbert we don t know when the read_and_verify_code in our child is actually executing within the read system call we want to interrupt this loop waits for a bit before sending the first signal to increase the likelihood of that implementations without correct eintr and signal handling usually fail this test this assumes anything unexpected that writes to stderr will also write a newline that is true of the traceback printing code we append a newline to our input so that a readline call can end on its own before the eof is seen and so that we re testing the read call that was interrupted by a signal before the end of the data stream has been reached pass string format for the read_and_verify_code used by read methods readline must handle signals and not lose data readlines must handle signals and not lose data readall must handle signals and not lose data read is the same thing as readall returns the infile line of code to make a bufferedreader bufferedreader read must handle signals and not lose data returns the infile line of code to make a textiowrapper readline must handle signals and not lose data readlines must handle signals and not lose data read must handle signals and not lose data
import os
import select
import signal
import subprocess
import sys
import time
import unittest
from test import support

if not support.has_subprocess_support:
    raise unittest.SkipTest("test module requires subprocess")


import _io
import _pyio

@unittest.skipUnless(os.name == 'posix', 'tests requires a posix system.')
class TestFileIOSignalInterrupt:
    def setUp(self):
        self._process = None

    def tearDown(self):
        if self._process and self._process.poll() is None:
            try:
                self._process.kill()
            except OSError:
                pass

    def _generate_infile_setup_code(self):
        
        return ('import %s as io ;'
                'infile = io.FileIO(sys.stdin.fileno(), "rb")' %
                self.modname)

    def fail_with_process_info(self, why, stdout=b'', stderr=b'',
                               communicate=True):
        
        if self._process.poll() is None:
            time.sleep(0.1)  
            try:
                self._process.terminate()  
            except OSError:
                pass
        if communicate:
            stdout_end, stderr_end = self._process.communicate()
            stdout += stdout_end
            stderr += stderr_end
        self.fail('Error from IO process %s:\nSTDOUT:\n%sSTDERR:\n%s\n' %
                  (why, stdout.decode(), stderr.decode()))

    def _test_reading(self, data_to_write, read_and_verify_code):
        
        infile_setup_code = self._generate_infile_setup_code()
        
        
        assert len(data_to_write) < 512, 'data_to_write must fit in pipe buf.'

        
        self._process = subprocess.Popen(
                [sys.executable, '-u', '-c',
                 'import signal, sys ;'
                 'signal.signal(signal.SIGINT, '
                               'lambda s, f: sys.stderr.write("$\\n")) ;'
                 + infile_setup_code + ' ;' +
                 'sys.stderr.write("Worm Sign!\\n") ;'
                 + read_and_verify_code + ' ;' +
                 'infile.close()'
                ],
                stdin=subprocess.PIPE, stdout=subprocess.PIPE,
                stderr=subprocess.PIPE)

        
        worm_sign = self._process.stderr.read(len(b'Worm Sign!\n'))
        if worm_sign != b'Worm Sign!\n':  
            self.fail_with_process_info('while awaiting a sign',
                                        stderr=worm_sign)
        self._process.stdin.write(data_to_write)

        signals_sent = 0
        rlist = []
        
        
        
        
        
        while not rlist:
            rlist, _, _ = select.select([self._process.stderr], (), (), 0.05)
            self._process.send_signal(signal.SIGINT)
            signals_sent += 1
            if signals_sent > 200:
                self._process.kill()
                self.fail('reader process failed to handle our signals.')
        
        
        signal_line = self._process.stderr.readline()
        if signal_line != b'$\n':
            self.fail_with_process_info('while awaiting signal',
                                        stderr=signal_line)

        
        
        
        
        stdout, stderr = self._process.communicate(input=b'\n')
        if self._process.returncode:
            self.fail_with_process_info(
                    'exited rc=%d' % self._process.returncode,
                    stdout, stderr, communicate=False)
        

    
    _READING_CODE_TEMPLATE = (
            'got = infile.{read_method_name}() ;'
            'expected = {expected!r} ;'
            'assert got == expected, ('
                    '"{read_method_name} returned wrong data.\\n"'
                    '"got data %r\\nexpected %r" % (got, expected))'
            )

    def test_readline(self):
        
        self._test_reading(
                data_to_write=b'hello, world!',
                read_and_verify_code=self._READING_CODE_TEMPLATE.format(
                        read_method_name='readline',
                        expected=b'hello, world!\n'))

    def test_readlines(self):
        
        self._test_reading(
                data_to_write=b'hello\nworld!',
                read_and_verify_code=self._READING_CODE_TEMPLATE.format(
                        read_method_name='readlines',
                        expected=[b'hello\n', b'world!\n']))

    def test_readall(self):
        
        self._test_reading(
                data_to_write=b'hello\nworld!',
                read_and_verify_code=self._READING_CODE_TEMPLATE.format(
                        read_method_name='readall',
                        expected=b'hello\nworld!\n'))
        
        self._test_reading(
                data_to_write=b'hello\nworld!',
                read_and_verify_code=self._READING_CODE_TEMPLATE.format(
                        read_method_name='read',
                        expected=b'hello\nworld!\n'))


class CTestFileIOSignalInterrupt(TestFileIOSignalInterrupt, unittest.TestCase):
    modname = '_io'

class PyTestFileIOSignalInterrupt(TestFileIOSignalInterrupt, unittest.TestCase):
    modname = '_pyio'


class TestBufferedIOSignalInterrupt(TestFileIOSignalInterrupt):
    def _generate_infile_setup_code(self):
        
        return ('import %s as io ;infile = io.open(sys.stdin.fileno(), "rb") ;'
                'assert isinstance(infile, io.BufferedReader)' %
                self.modname)

    def test_readall(self):
        
        self._test_reading(
                data_to_write=b'hello\nworld!',
                read_and_verify_code=self._READING_CODE_TEMPLATE.format(
                        read_method_name='read',
                        expected=b'hello\nworld!\n'))

class CTestBufferedIOSignalInterrupt(TestBufferedIOSignalInterrupt, unittest.TestCase):
    modname = '_io'

class PyTestBufferedIOSignalInterrupt(TestBufferedIOSignalInterrupt, unittest.TestCase):
    modname = '_pyio'


class TestTextIOSignalInterrupt(TestFileIOSignalInterrupt):
    def _generate_infile_setup_code(self):
        
        return ('import %s as io ;'
                'infile = io.open(sys.stdin.fileno(), encoding="utf-8", newline=None) ;'
                'assert isinstance(infile, io.TextIOWrapper)' %
                self.modname)

    def test_readline(self):
        
        self._test_reading(
                data_to_write=b'hello, world!',
                read_and_verify_code=self._READING_CODE_TEMPLATE.format(
                        read_method_name='readline',
                        expected='hello, world!\n'))

    def test_readlines(self):
        
        self._test_reading(
                data_to_write=b'hello\r\nworld!',
                read_and_verify_code=self._READING_CODE_TEMPLATE.format(
                        read_method_name='readlines',
                        expected=['hello\n', 'world!\n']))

    def test_readall(self):
        
        self._test_reading(
                data_to_write=b'hello\nworld!',
                read_and_verify_code=self._READING_CODE_TEMPLATE.format(
                        read_method_name='read',
                        expected="hello\nworld!\n"))

class CTestTextIOSignalInterrupt(TestTextIOSignalInterrupt, unittest.TestCase):
    modname = '_io'

class PyTestTextIOSignalInterrupt(TestTextIOSignalInterrupt, unittest.TestCase):
    modname = '_pyio'


if __name__ == '__main__':
    unittest.main()