Method _test_recv

Lib/test/_test_eintr.py:210–243

(self, recv_func)

    @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
        data = [b"x", b"y", b"z"]

        code = '\n'.join((
            'import os, socket, sys, time',
            '',
            'fd = int(sys.argv[1])',
            f'family = {int(wr.family)}',
            f'sock_type = {int(wr.type)}',
            f'data = {data!r}',
            f'sleep_time = {self.sleep_time!r}',
            '',
            'wr = socket.fromfd(fd, family, sock_type)',
            'os.close(fd)',
            '',
            'with wr:',
            '    for item in data:',
            '        # let the parent block on recv()',
            '        time.sleep(sleep_time)',
            '        wr.sendall(item)',
        ))

        fd = wr.fileno()
        proc = self.subprocess(code, str(fd), pass_fds=[fd])
        with kill_on_error(proc):
            wr.close()
            for item in data:
                self.assertEqual(item, recv_func(rd, len(item)))
            self.assertEqual(proc.wait(), 0)

    def test_recv(self):
        self._test_recv(socket.socket.recv)