test_standard.py 文件源码

def test_string_host_yields_single_host(self):
        client = StandardRedisClient(hosts=['redis://localhost:1234/0'])

        payload = {'test': 'test_string_host_yields_single_host'}

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

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

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