class_docstr: EINTR tests for the os module. 
class_docstr_tok: ["EINTR","tests","for","the","os","module","."]
code: 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)
  
imports: import contextlib
  import faulthandler
  import fcntl
  import os
  import platform
  import select
  import signal
  import socket
  import subprocess
  import sys
  import time
  import unittest
  from test import support
  from test.support import os_helper
  from test.support import socket_helper
label_desc: Use image node_id 9 for calling the OSEINTRTest obj's underlying member method code with example usage: obj.test_write() without return types