test_standard.py 文件源码

def test_no_hosts_yields_single_default_host(self):
        client = StandardRedisClient()

        payload = {'test': 'test_no_hosts_yields_single_default_host'}

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

        message = client.get_connection(
            'test_no_hosts_yields_single_default_host',
        ).lpop('test_no_hosts_yields_single_default_host')

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