Method test_write

Lib/test/_test_eintr.py:166–202

(self)

            self.assertEqual(buffer, expected)

    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])',
            f'sleep_time = {self.sleep_time!r}',
            f'data = b"x" * {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(f"read error: {len(value)}'
                                  ' vs {data_len} bytes")',
        ))

        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()")