test_security.py 文件源码

def bounce(self, server, client, test_metadata=True):
        msg = [os.urandom(64), os.urandom(64)]
        client.send_multipart(msg)
        frames = self.recv_multipart(server, copy=False)
        recvd = list(map(lambda x: x.bytes, frames))

        try:
            if test_metadata and not PYPY:
                for frame in frames:
                    self.assertEqual(frame.get('User-Id'), 'anonymous')
                    self.assertEqual(frame.get('Hello'), 'World')
                    self.assertEqual(frame['Socket-Type'], 'DEALER')
        except zmq.ZMQVersionError:
            pass

        self.assertEqual(recvd, msg)
        server.send_multipart(recvd)
        msg2 = self.recv_multipart(client)
        self.assertEqual(msg2, msg)