Method testSendAndRecvFds

Lib/test/test_socket.py:7373–7408

(self)

@requireAttrs(socket, "AF_UNIX")
class SendRecvFdsTests(unittest.TestCase):
    def testSendAndRecvFds(self):
        def close_pipes(pipes):
            for fd1, fd2 in pipes:
                os.close(fd1)
                os.close(fd2)

        def close_fds(fds):
            for fd in fds:
                os.close(fd)

        # send 10 file descriptors
        pipes = [os.pipe() for _ in range(10)]
        self.addCleanup(close_pipes, pipes)
        fds = [rfd for rfd, wfd in pipes]

        # use a UNIX socket pair to exchange file descriptors locally
        sock1, sock2 = socket.socketpair(socket.AF_UNIX, socket.SOCK_STREAM)
        with sock1, sock2:
            socket.send_fds(sock1, [MSG], fds)
            # request more data and file descriptors than expected
            msg, fds2, flags, addr = socket.recv_fds(sock2, len(MSG) * 2, len(fds) * 2)
            self.addCleanup(close_fds, fds2)

        self.assertEqual(msg, MSG)
        self.assertEqual(len(fds2), len(fds))
        self.assertEqual(flags, 0)
        # don't test addr

        # test that file descriptors are connected
        for index, fds in enumerate(pipes):
            rfd, wfd = fds
            os.write(wfd, str(index).encode())

        for index, rfd in enumerate(fds2):
            data = os.read(rfd, 100)
            self.assertEqual(data,  str(index).encode())


class FreeThreadingTests(unittest.TestCase):