Method _interrupted_reads

Lib/test/_test_eintr.py:126–154

Make a fd which will force block on read of expected bytes.

(self)

        self._test_wait_single(lambda pid: os.wait4(pid, 0))

    def _interrupted_reads(self):
        """Make a fd which will force block on read of expected bytes."""
        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
        data = [b"hello", b"world", b"spam"]

        code = '\n'.join((
            'import os, sys, time',
            '',
            'wr = int(sys.argv[1])',
            f'data = {data!r}',
            f'sleep_time = {self.sleep_time!r}',
            '',
            'for item in data:',
            '    # let the parent block on read()',
            '    time.sleep(sleep_time)',
            '    os.write(wr, item)',
        ))

        proc = self.subprocess(code, str(wr), pass_fds=[wr])
        with kill_on_error(proc):
            os.close(wr)
            for datum in data:
                yield rd, datum
            self.assertEqual(proc.wait(), 0)

    def test_read(self):
        for fd, expected in self._interrupted_reads():