# Path: tests/cases/async.py
# class AsyncTestCase(testing.AsyncTestCase):
# 
#     def future_value(self, value):
#         f = concurrent.Future()
#         f.set_result(value)
#         return f
# 
#     def future_error(self, exc, instance=None, tb=None):
#         f = concurrent.Future()
#         if instance is None or tb is None:
#             f.set_exception(exc)
#         else:
#             f.set_exc_info((exc, instance, tb))
# 
#         return f


import collections
import six
from tornado import gen
from mock import patch, Mock
from .async import AsyncTestCase

        @gen.coroutine
        def refresh_metadata():
            cluster.topics.clear()
            cluster.leaders.clear()
            for topic, leaders in six.iteritems(self.topic_leaders):
                for partition in list(range(len(leaders))):
                    cluster.topics[topic].append(partition)
                    cluster.leaders[topic][partition] = leaders[partition]

        cluster.start.side_effect = refresh_metadata
        cluster.heal.side_effect = refresh_metadata
        cluster.stop.return_value = self.future_value(None)

    def add_broker(self, host, port, broker_id):
        broker = Mock()

        @gen.coroutine
        def mock_send(request):
            self.requests_by_broker[broker_id].append(request)
            response = self.responses[broker_id][request.api].pop(0)
            if isinstance(response, Exception):
                raise response
            response.correlation_id = request.correlation_id
            raise gen.Return(response)

        broker.send.side_effect = mock_send

        self.mock_brokers[broker_id] = broker


# Path: tests/cases/async.py
# class AsyncTestCase(testing.AsyncTestCase):
# 
#     def future_value(self, value):
#         f = concurrent.Future()
#         f.set_result(value)
#         return f
# 
#     def future_error(self, exc, instance=None, tb=None):
#         f = concurrent.Future()
#         if instance is None or tb is None:
#             f.set_exception(exc)
#         else:
#             f.set_exc_info((exc, instance, tb))
# 
#         return f


# Path: tests/cases/client.py
import collections
import six
from tornado import gen
from mock import patch, Mock
from .async import AsyncTestCase


        @gen.coroutine
        def refresh_metadata():
            cluster.topics.clear()
            cluster.leaders.clear()
            for topic, leaders in six.iteritems(self.topic_leaders):
                for partition in list(range(len(leaders))):
                    cluster.topics[topic].append(partition)
                    cluster.leaders[topic][partition] = leaders[partition]

        cluster.start.side_effect = refresh_metadata
        cluster.heal.side_effect = refresh_metadata
        cluster.stop.return_value = self.future_value(None)

    def add_broker(self, host, port, broker_id):
        broker = Mock()

        @gen.coroutine
        def mock_send(request):
            self.requests_by_broker[broker_id].append(request)
            response = self.responses[broker_id][request.api].pop(0)
            if isinstance(response, Exception):
                raise response
            response.correlation_id = request.correlation_id
            raise gen.Return(response)

        broker.send.side_effect = mock_send

        self.mock_brokers[broker_id] = broker


    def add_topic(self, topic_name, leaders):