test_sentinel.py 文件源码

def test_no_hosts_send_receive(self):
        client = SentinelRedisClient()

        payload = {'test': 'test_no_hosts_send_receive'}

        client.send_message_to_queue(
            queue_key='test_no_hosts_send_receive',
            message=msgpack.packb(payload),
            expiry=10,
            capacity=10,
            connection=client.get_connection('test_no_hosts_send_receive'),
        )

        message = None
        for i in range(3):
            # Message will be on random server
            message = message or client.get_connection('test_no_hosts_send_receive').lpop('test_no_hosts_send_receive')

        self.assertIsNotNone(message)
        self.assertEqual(payload, msgpack.unpackb(message, encoding='utf-8'))