'BlockingChannel user cancels non-ackable consumer via basic_cancel'

def test(self):

    connection = self._connect()
    ch = connection.channel()
    q_name = ('TestBasicCancelWithNonAckableConsumer_q' + uuid.uuid1().hex)
    body1 = ('a' * 1024)
    body2 = ('b' * 2048)
    ch.queue_declare(q_name, auto_delete=False)
    self.addCleanup(self._connect().channel().queue_delete, q_name)
    ch.publish(exchange='', routing_key=q_name, body=body1)
    ch.publish(exchange='', routing_key=q_name, body=body2)
    self._assert_exact_message_count_with_retries(channel=ch, queue=q_name, expected_count=2)
    consumer_tag = ch.basic_consume((lambda *x: None), q_name, no_ack=True, exclusive=False, arguments=None)
    self._assert_exact_message_count_with_retries(channel=ch, queue=q_name, expected_count=0)
    messages = ch.basic_cancel(consumer_tag)
    self.assertEqual(len(messages), 2)
    (_, _, rx_body1) = messages[0]
    self.assertEqual(rx_body1, as_bytes(body1))
    (_, _, rx_body2) = messages[1]
    self.assertEqual(rx_body2, as_bytes(body2))
    ch.close()
    ch = connection.channel()
    frame = ch.queue_declare(q_name, passive=True)
    self.assertEqual(frame.method.message_count, 0)
