'BlockingChannel: two basic_consume consumers on same channel'

def test(self):

    connection = self._connect()
    ch = connection.channel()
    exg_name = ('TestPublishAndConsumeAndQos_exg_' + uuid.uuid1().hex)
    q1_name = ('TestTwoBasicConsumersOnSameChannel_q1' + uuid.uuid1().hex)
    q2_name = ('TestTwoBasicConsumersOnSameChannel_q2' + uuid.uuid1().hex)
    q1_routing_key = 'TestTwoBasicConsumersOnSameChannel1'
    q2_routing_key = 'TestTwoBasicConsumersOnSameChannel2'
    ch.confirm_delivery()
    ch.exchange_declare(exg_name, exchange_type='direct')
    self.addCleanup(connection.channel().exchange_delete, exg_name)
    ch.queue_declare(q1_name, auto_delete=True)
    self.addCleanup(self._connect().channel().queue_delete, q1_name)
    ch.queue_bind(q1_name, exchange=exg_name, routing_key=q1_routing_key)
    ch.queue_declare(q2_name, auto_delete=True)
    self.addCleanup(self._connect().channel().queue_delete, q2_name)
    ch.queue_bind(q2_name, exchange=exg_name, routing_key=q2_routing_key)
    q1_tx_message_bodies = [('q1_message+%s' % (i,)) for i in pika.compat.xrange(100)]
    for message_body in q1_tx_message_bodies:
        ch.publish(exg_name, q1_routing_key, body=message_body, mandatory=True)
    q2_tx_message_bodies = [('q2_message+%s' % (i,)) for i in pika.compat.xrange(150)]
    for message_body in q2_tx_message_bodies:
        ch.publish(exg_name, q2_routing_key, body=message_body, mandatory=True)
    q1_rx_messages = []
    q1_consumer_tag = ch.basic_consume((lambda *args: q1_rx_messages.append(args)), q1_name, no_ack=False, exclusive=False, arguments=None)
    q2_rx_messages = []
    q2_consumer_tag = ch.basic_consume((lambda *args: q2_rx_messages.append(args)), q2_name, no_ack=False, exclusive=False, arguments=None)
    while ((len(q1_rx_messages) < len(q1_tx_message_bodies)) or (len(q2_rx_messages) < len(q2_tx_message_bodies))):
        connection.process_data_events(time_limit=None)
    self.assertEqual(len(q2_rx_messages), len(q2_tx_message_bodies))
    def validate_messages(rx_messages, routing_key, consumer_tag, tx_message_bodies):
        self.assertEqual(len(rx_messages), len(tx_message_bodies))
        for (msg, expected_body) in zip(rx_messages, tx_message_bodies):
            self.assertIsInstance(msg, tuple)
            (rx_ch, rx_method, rx_properties, rx_body) = msg
            self.assertIs(rx_ch, ch)
            self.assertIsInstance(rx_method, pika.spec.Basic.Deliver)
            self.assertEqual(rx_method.consumer_tag, consumer_tag)
            self.assertFalse(rx_method.redelivered)
            self.assertEqual(rx_method.exchange, exg_name)
            self.assertEqual(rx_method.routing_key, routing_key)
            self.assertIsInstance(rx_properties, pika.BasicProperties)
            self.assertEqual(rx_body, as_bytes(expected_body))
    validate_messages(rx_messages=q1_rx_messages, routing_key=q1_routing_key, consumer_tag=q1_consumer_tag, tx_message_bodies=q1_tx_message_bodies)
    validate_messages(rx_messages=q2_rx_messages, routing_key=q2_routing_key, consumer_tag=q2_consumer_tag, tx_message_bodies=q2_tx_message_bodies)
    self.assertFalse(ch._pending_events)
