'BlockingChannel.basic_publish, publish, basic_consume, QoS,         Basic.Cancel from broker'

def test(self):

    connection = self._connect()
    ch = connection.channel()
    q_name = ('TestPublishAndConsumeAndQos_q' + uuid.uuid1().hex)
    exg_name = ('TestPublishAndConsumeAndQos_exg_' + uuid.uuid1().hex)
    routing_key = 'TestPublishAndConsumeAndQos'
    res = ch.confirm_delivery()
    self.assertIsNone(res)
    ch.exchange_declare(exg_name, exchange_type='direct')
    self.addCleanup(connection.channel().exchange_delete, exg_name)
    ch.queue_declare(q_name, auto_delete=True)
    self.addCleanup(self._connect().channel().queue_delete, q_name)
    ch.queue_bind(q_name, exchange=exg_name, routing_key=routing_key)
    msg1_headers = dict(test_name='TestPublishAndConsumeWithPubacksAndQosOfOne')
    msg1_properties = pika.spec.BasicProperties(headers=msg1_headers)
    res = ch.basic_publish(exg_name, routing_key=routing_key, body='via-basic_publish', properties=msg1_properties, mandatory=True)
    self.assertEqual(res, True)
    ch.publish(exg_name, routing_key, body='via-publish', mandatory=True)
    frame = ch.queue_declare(q_name, passive=True)
    self.assertEqual(frame.method.message_count, 2)
    ch.basic_qos(prefetch_size=0, prefetch_count=1, all_channels=False)
    rx_messages = []
    consumer_tag = ch.basic_consume((lambda *args: rx_messages.append(args)), q_name, no_ack=False, exclusive=False, arguments=None)
    while (not rx_messages):
        connection.process_data_events(time_limit=None)
    self.assertEqual(len(rx_messages), 1)
    msg = rx_messages[0]
    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.assertEqual(rx_method.delivery_tag, 1)
    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_properties.headers, msg1_headers)
    self.assertEqual(rx_body, as_bytes('via-basic_publish'))
    self.assertFalse(ch._pending_events)
    ch.basic_ack(delivery_tag=rx_method.delivery_tag, multiple=False)
    while (len(rx_messages) < 2):
        connection.process_data_events(time_limit=None)
    self.assertEqual(len(rx_messages), 2)
    msg = rx_messages[1]
    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.assertEqual(rx_method.delivery_tag, 2)
    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('via-publish'))
    self.assertFalse(ch._pending_events)
    ch.basic_ack(delivery_tag=rx_method.delivery_tag, multiple=False)
    self._assert_exact_message_count_with_retries(channel=ch, queue=q_name, expected_count=0)
    connection.process_data_events(time_limit=0.005)
    self.assertEqual(len(rx_messages), 2)
    rx_cancellations = []
    ch.add_on_cancel_callback(rx_cancellations.append)
    ch.queue_delete(q_name)
    ch.start_consuming()
    self.assertEqual(len(rx_cancellations), 1)
    (frame,) = rx_cancellations
    self.assertEqual(frame.method.consumer_tag, consumer_tag)
