text: """
  This test suite exercises some system calls subject to interruption with EINTR,
  to check that it is actually handled transparently.
  It is intended to be run by the main test suite within a child process, to
  ensure there is no background thread running (so that signals are delivered to
  the correct thread).
  Signals are generated in-process using setitimer(ITIMER_REAL), which allows
  sub-second periodicity (contrarily to signal()).
  """
  
  import contextlib
  import io
  import os
  import select
  import signal
  import socket
  import subprocess
  import sys
  import time
  import unittest
  
  from test import support
  
  @contextlib.contextmanager
  def kill_on_error(proc):
      """Context manager killing the subprocess if a Python exception is raised."""
      with proc:
          try:
              yield proc
          except:
              proc.kill()
              raise
  
  
  @unittest.skipUnless(hasattr(signal, "setitimer"), "requires setitimer()")
  class EINTRBaseTest(unittest.TestCase):
      """ Base class for EINTR tests. """
  
      # delay for initial signal delivery
      signal_delay = 0.1
      # signal delivery periodicity
      signal_period = 0.1
      # default sleep time for tests - should obviously have:
      # sleep_time > signal_period
      sleep_time = 0.2
  
      @classmethod
      def setUpClass(cls):
          cls.orig_handler = signal.signal(signal.SIGALRM, lambda *args: None)
          signal.setitimer(signal.ITIMER_REAL, cls.signal_delay,
                           cls.signal_period)
  
      @classmethod
      def stop_alarm(cls):
          signal.setitimer(signal.ITIMER_REAL, 0, 0)
  
      @classmethod
      def tearDownClass(cls):
          cls.stop_alarm()
          signal.signal(signal.SIGALRM, cls.orig_handler)
  
      def subprocess(self, *args, **kw):
          cmd_args = (sys.executable, '-c') + args
          return subprocess.Popen(cmd_args, **kw)
  
  
  @unittest.skipUnless(hasattr(signal, "setitimer"), "requires setitimer()")
  class OSEINTRTest(EINTRBaseTest):
      """ EINTR tests for the os module. """
  
      def new_sleep_process(self):
          code = 'import time; time.sleep(%r)' % self.sleep_time
          return self.subprocess(code)
  
      def _test_wait_multiple(self, wait_func):
          num = 3
          processes = [self.new_sleep_process() for _ in range(num)]
          for _ in range(num):
              wait_func()
  
      def test_wait(self):
          self._test_wait_multiple(os.wait)
  
      @unittest.skipUnless(hasattr(os, 'wait3'), 'requires wait3()')
      def test_wait3(self):
          self._test_wait_multiple(lambda: os.wait3(0))
  
      def _test_wait_single(self, wait_func):
          proc = self.new_sleep_process()
          wait_func(proc.pid)
  
      def test_waitpid(self):
          self._test_wait_single(lambda pid: os.waitpid(pid, 0))
  
      @unittest.skipUnless(hasattr(os, 'wait4'), 'requires wait4()')
      def test_wait4(self):
          self._test_wait_single(lambda pid: os.wait4(pid, 0))
  
      def test_read(self):
          rd, wr = os.pipe()
          self.addCleanup(os.close, rd)
          # wr closed explicitly by parent
  
          # the payload below are smaller than PIPE_BUF, hence the writes will be
          # atomic
          datas = [b"hello", b"world", b"spam"]
  
          code = '\n'.join((
              'import os, sys, time',
              '',
              'wr = int(sys.argv[1])',
              'datas = %r' % datas,
              'sleep_time = %r' % self.sleep_time,
              '',
              'for data in datas:',
              '    # let the parent block on read()',
              '    time.sleep(sleep_time)',
              '    os.write(wr, data)',
          ))
  
          proc = self.subprocess(code, str(wr), pass_fds=[wr])
          with kill_on_error(proc):
              os.close(wr)
              for data in datas:
                  self.assertEqual(data, os.read(rd, len(data)))
              self.assertEqual(proc.wait(), 0)
  
      def test_write(self):
          rd, wr = os.pipe()
          self.addCleanup(os.close, wr)
          # rd closed explicitly by parent
  
          # we must write enough data for the write() to block
          data = b"x" * support.PIPE_MAX_SIZE
  
          code = '\n'.join((
              'import io, os, sys, time',
              '',
              'rd = int(sys.argv[1])',
              'sleep_time = %r' % self.sleep_time,
              'data = b"x" * %s' % support.PIPE_MAX_SIZE,
              'data_len = len(data)',
              '',
              '# let the parent block on write()',
              'time.sleep(sleep_time)',
              '',
              'read_data = io.BytesIO()',
              'while len(read_data.getvalue()) < data_len:',
              '    chunk = os.read(rd, 2 * data_len)',
              '    read_data.write(chunk)',
              '',
              'value = read_data.getvalue()',
              'if value != data:',
              '    raise Exception("read error: %s vs %s bytes"',
              '                    % (len(value), data_len))',
          ))
  
          proc = self.subprocess(code, str(rd), pass_fds=[rd])
          with kill_on_error(proc):
              os.close(rd)
              written = 0
              while written < len(data):
                  written += os.write(wr, memoryview(data)[written:])
              self.assertEqual(proc.wait(), 0)
  
  
  @unittest.skipUnless(hasattr(signal, "setitimer"), "requires setitimer()")
  class SocketEINTRTest(EINTRBaseTest):
      """ EINTR tests for the socket module. """
  
      @unittest.skipUnless(hasattr(socket, 'socketpair'), 'needs socketpair()')
      def _test_recv(self, recv_func):
          rd, wr = socket.socketpair()
          self.addCleanup(rd.close)
          # wr closed explicitly by parent
  
          # single-byte payload guard us against partial recv
          datas = [b"x", b"y", b"z"]
  
          code = '\n'.join((
              'import os, socket, sys, time',
              '',
              'fd = int(sys.argv[1])',
              'family = %s' % int(wr.family),
              'sock_type = %s' % int(wr.type),
              'datas = %r' % datas,
              'sleep_time = %r' % self.sleep_time,
              '',
              'wr = socket.fromfd(fd, family, sock_type)',
              'os.close(fd)',
              '',
              'with wr:',
              '    for data in datas:',
              '        # let the parent block on recv()',
              '        time.sleep(sleep_time)',
              '        wr.sendall(data)',
          ))
  
          fd = wr.fileno()
          proc = self.subprocess(code, str(fd), pass_fds=[fd])
          with kill_on_error(proc):
              wr.close()
              for data in datas:
                  self.assertEqual(data, recv_func(rd, len(data)))
              self.assertEqual(proc.wait(), 0)
  
      def test_recv(self):
          self._test_recv(socket.socket.recv)
  
      @unittest.skipUnless(hasattr(socket.socket, 'recvmsg'), 'needs recvmsg()')
      def test_recvmsg(self):
          self._test_recv(lambda sock, data: sock.recvmsg(data)[0])
  
      def _test_send(self, send_func):
          rd, wr = socket.socketpair()
          self.addCleanup(wr.close)
          # rd closed explicitly by parent
  
          # we must send enough data for the send() to block
          data = b"xyz" * (support.SOCK_MAX_SIZE // 3)
  
          code = '\n'.join((
              'import os, socket, sys, time',
              '',
              'fd = int(sys.argv[1])',
              'family = %s' % int(rd.family),
              'sock_type = %s' % int(rd.type),
              'sleep_time = %r' % self.sleep_time,
              'data = b"xyz" * %s' % (support.SOCK_MAX_SIZE // 3),
              'data_len = len(data)',
              '',
              'rd = socket.fromfd(fd, family, sock_type)',
              'os.close(fd)',
              '',
              'with rd:',
              '    # let the parent block on send()',
              '    time.sleep(sleep_time)',
              '',
              '    received_data = bytearray(data_len)',
              '    n = 0',
              '    while n < data_len:',
              '        n += rd.recv_into(memoryview(received_data)[n:])',
              '',
              'if received_data != data:',
              '    raise Exception("recv error: %s vs %s bytes"',
              '                    % (len(received_data), data_len))',
          ))
  
          fd = rd.fileno()
          proc = self.subprocess(code, str(fd), pass_fds=[fd])
          with kill_on_error(proc):
              rd.close()
              written = 0
              while written < len(data):
                  sent = send_func(wr, memoryview(data)[written:])
                  # sendall() returns None
                  written += len(data) if sent is None else sent
              self.assertEqual(proc.wait(), 0)
  
      def test_send(self):
          self._test_send(socket.socket.send)
  
      def test_sendall(self):
          self._test_send(socket.socket.sendall)
  
      @unittest.skipUnless(hasattr(socket.socket, 'sendmsg'), 'needs sendmsg()')
      def test_sendmsg(self):
          self._test_send(lambda sock, data: sock.sendmsg([data]))
  
      def test_accept(self):
          sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
          self.addCleanup(sock.close)
  
          sock.bind((support.HOST, 0))
          port = sock.getsockname()[1]
          sock.listen()
  
          code = '\n'.join((
              'import socket, time',
              '',
              'host = %r' % support.HOST,
              'port = %s' % port,
              'sleep_time = %r' % self.sleep_time,
              '',
              '# let parent block on accept()',
              'time.sleep(sleep_time)',
              'with socket.create_connection((host, port)):',
              '    time.sleep(sleep_time)',
          ))
  
          proc = self.subprocess(code)
          with kill_on_error(proc):
              client_sock, _ = sock.accept()
              client_sock.close()
              self.assertEqual(proc.wait(), 0)
  
      # Issue #25122: There is a race condition in the FreeBSD kernel on
      # handling signals in the FIFO device. Skip the test until the bug is
      # fixed in the kernel.
      # https://bugs.freebsd.org/bugzilla/show_bug.cgi?id=203162
      @support.requires_freebsd_version(10, 3)
      @unittest.skipUnless(hasattr(os, 'mkfifo'), 'needs mkfifo()')
      def _test_open(self, do_open_close_reader, do_open_close_writer):
          filename = support.TESTFN
  
          # Use a fifo: until the child opens it for reading, the parent will
          # block when trying to open it for writing.
          support.unlink(filename)
          os.mkfifo(filename)
          self.addCleanup(support.unlink, filename)
  
          code = '\n'.join((
              'import os, time',
              '',
              'path = %a' % filename,
              'sleep_time = %r' % self.sleep_time,
              '',
              '# let the parent block',
              'time.sleep(sleep_time)',
              '',
              do_open_close_reader,
          ))
  
          proc = self.subprocess(code)
          with kill_on_error(proc):
              do_open_close_writer(filename)
              self.assertEqual(proc.wait(), 0)
  
      def python_open(self, path):
          fp = open(path, 'w')
          fp.close()
  
      def test_open(self):
          self._test_open("fp = open(path, 'r')\nfp.close()",
                          self.python_open)
  
      @unittest.skipIf(sys.platform == 'darwin', "hangs under OS X; see issue #25234")
      def os_open(self, path):
          fd = os.open(path, os.O_WRONLY)
          os.close(fd)
  
      @unittest.skipIf(sys.platform == "darwin", "hangs under OS X; see issue #25234")
      def test_os_open(self):
          self._test_open("fd = os.open(path, os.O_RDONLY)\nos.close(fd)",
                          self.os_open)
  
  
  @unittest.skipUnless(hasattr(signal, "setitimer"), "requires setitimer()")
  class TimeEINTRTest(EINTRBaseTest):
      """ EINTR tests for the time module. """
  
      def test_sleep(self):
          t0 = time.monotonic()
          time.sleep(self.sleep_time)
          self.stop_alarm()
          dt = time.monotonic() - t0
          self.assertGreaterEqual(dt, self.sleep_time)
  
  
  @unittest.skipUnless(hasattr(signal, "setitimer"), "requires setitimer()")
  class SignalEINTRTest(EINTRBaseTest):
      """ EINTR tests for the signal module. """
  
      @unittest.skipUnless(hasattr(signal, 'sigtimedwait'),
                           'need signal.sigtimedwait()')
      def test_sigtimedwait(self):
          t0 = time.monotonic()
          signal.sigtimedwait([signal.SIGUSR1], self.sleep_time)
          dt = time.monotonic() - t0
          self.assertGreaterEqual(dt, self.sleep_time)
  
      @unittest.skipUnless(hasattr(signal, 'sigwaitinfo'),
                           'need signal.sigwaitinfo()')
      def test_sigwaitinfo(self):
          # Issue #25277, #25868: give a few miliseconds to the parent process
          # between os.write() and signal.sigwaitinfo() to works around a race
          # condition
          self.sleep_time = 0.100
  
          signum = signal.SIGUSR1
          pid = os.getpid()
  
          old_handler = signal.signal(signum, lambda *args: None)
          self.addCleanup(signal.signal, signum, old_handler)
  
          rpipe, wpipe = os.pipe()
  
          code = '\n'.join((
              'import os, time',
              'pid = %s' % os.getpid(),
              'signum = %s' % int(signum),
              'sleep_time = %r' % self.sleep_time,
              'rpipe = %r' % rpipe,
              'os.read(rpipe, 1)',
              'os.close(rpipe)',
              'time.sleep(sleep_time)',
              'os.kill(pid, signum)',
          ))
  
          t0 = time.monotonic()
          proc = self.subprocess(code, pass_fds=(rpipe,))
          os.close(rpipe)
          with kill_on_error(proc):
              # sync child-parent
              os.write(wpipe, b'x')
              os.close(wpipe)
  
              # parent
              signal.sigwaitinfo([signum])
              dt = time.monotonic() - t0
              self.assertEqual(proc.wait(), 0)
  
          self.assertGreaterEqual(dt, self.sleep_time)
  
  
  @unittest.skipUnless(hasattr(signal, "setitimer"), "requires setitimer()")
  class SelectEINTRTest(EINTRBaseTest):
      """ EINTR tests for the select module. """
  
      def test_select(self):
          t0 = time.monotonic()
          select.select([], [], [], self.sleep_time)
          dt = time.monotonic() - t0
          self.stop_alarm()
          self.assertGreaterEqual(dt, self.sleep_time)
  
      @unittest.skipUnless(hasattr(select, 'poll'), 'need select.poll')
      def test_poll(self):
          poller = select.poll()
  
          t0 = time.monotonic()
          poller.poll(self.sleep_time * 1e3)
          dt = time.monotonic() - t0
          self.stop_alarm()
          self.assertGreaterEqual(dt, self.sleep_time)
  
      @unittest.skipUnless(hasattr(select, 'epoll'), 'need select.epoll')
      def test_epoll(self):
          poller = select.epoll()
          self.addCleanup(poller.close)
  
          t0 = time.monotonic()
          poller.poll(self.sleep_time)
          dt = time.monotonic() - t0
          self.stop_alarm()
          self.assertGreaterEqual(dt, self.sleep_time)
  
      @unittest.skipUnless(hasattr(select, 'kqueue'), 'need select.kqueue')
      def test_kqueue(self):
          kqueue = select.kqueue()
          self.addCleanup(kqueue.close)
  
          t0 = time.monotonic()
          kqueue.control(None, 1, self.sleep_time)
          dt = time.monotonic() - t0
          self.stop_alarm()
          self.assertGreaterEqual(dt, self.sleep_time)
  
      @unittest.skipUnless(hasattr(select, 'devpoll'), 'need select.devpoll')
      def test_devpoll(self):
          poller = select.devpoll()
          self.addCleanup(poller.close)
  
          t0 = time.monotonic()
          poller.poll(self.sleep_time * 1e3)
          dt = time.monotonic() - t0
          self.stop_alarm()
          self.assertGreaterEqual(dt, self.sleep_time)
  
  
  def test_main():
      support.run_unittest(
          OSEINTRTest,
          SocketEINTRTest,
          TimeEINTRTest,
          SignalEINTRTest,
          SelectEINTRTest)
  
  
  if __name__ == "__main__":
      test_main()
  