docstr: 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.
docstr_tok: ["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","."]
code: def _test_reading(self, data_to_write, read_and_verify_code):
      """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.
      """
      infile_setup_code = self._generate_infile_setup_code()
      # Total pipe IO in this function is smaller than the minimum posix OS
      # pipe buffer size of 512 bytes.  No writer should block.
      assert (
          len(data_to_write) < 512
      ), "data_to_write must fit in pipe buf."
  
      # Start a subprocess to call our read method while handling a signal.
      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,
      )
  
      # Wait for the signal handler to be installed.
      worm_sign = self._process.stderr.read(len(b"Worm Sign!\n"))
      if (
          worm_sign != b"Worm Sign!\n"
      ):  # See also, Dune by Frank Herbert.
          self.fail_with_process_info(
              "while awaiting a sign", stderr=worm_sign
          )
      self._process.stdin.write(data_to_write)
  
      signals_sent = 0
      rlist = []
      # 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.
      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.")
      # This assumes anything unexpected that writes to stderr will also
      # write a newline.  That is true of the traceback printing code.
      signal_line = self._process.stderr.readline()
      if signal_line != b"$\n":
          self.fail_with_process_info(
              "while awaiting signal", stderr=signal_line
          )
  
      # 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.
      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,
          )
  
imports: import os
  import select
  import signal
  import subprocess
  import sys
  import time
  import unittest
  from test import support
  import _io
  import _pyio
label_desc: Use image node_id 5 for calling the TestFileIOSignalInterrupt obj's underlying member method code with example usage: obj._test_reading(data_to_write, read_and_verify_code) without return types