Method _test_send

Lib/test/_test_eintr.py:252–296

(self, send_func)

        self._test_recv(lambda sock, data: sock.recvmsg(data)[0])

    def _test_send(self, send_func):
        rd, wr = socket.socketpair()
        self.addCleanup(wr.close)
        # rd closed explicitly by parent

        # we must send enough data for the send() to block
        data = b"xyz" * (support.SOCK_MAX_SIZE // 3)

        code = '\n'.join((
            'import os, socket, sys, time',
            '',
            'fd = int(sys.argv[1])',
            f'family = {int(rd.family)}',
            f'sock_type = {int(rd.type)}',
            f'sleep_time = {self.sleep_time!r}',
            f'data = b"xyz" * {support.SOCK_MAX_SIZE // 3}',
            'data_len = len(data)',
            '',
            'rd = socket.fromfd(fd, family, sock_type)',
            'os.close(fd)',
            '',
            'with rd:',
            '    # let the parent block on send()',
            '    time.sleep(sleep_time)',
            '',
            '    received_data = bytearray(data_len)',
            '    n = 0',
            '    while n < data_len:',
            '        n += rd.recv_into(memoryview(received_data)[n:])',
            '',
            'if received_data != data:',
            '    raise Exception(f"recv error: {len(received_data)}'
                                  ' vs {data_len} bytes")',
        ))

        fd = rd.fileno()
        proc = self.subprocess(code, str(fd), pass_fds=[fd])
        with kill_on_error(proc):
            rd.close()
            written = 0
            while written < len(data):
                sent = send_func(wr, memoryview(data)[written:])
                # sendall() returns None
                written += len(data) if sent is None else sent
            self.assertEqual(proc.wait(), 0)

    def test_send(self):
        self._test_send(socket.socket.send)