Method test

tests/acceptance/blocking_adapter_test.py:1260–1299

BlockingChannel.basic_nack with requeue=False

(self)

class TestBasicNackNoRequeue(BlockingTestCaseBase):

    def test(self):
        """BlockingChannel.basic_nack with requeue=False"""
        connection = self._connect()

        ch = connection.channel()

        q_name = 'TestBasicNackNoRequeue_q' + uuid.uuid1().hex

        # Place channel in publisher-acknowledgments mode so that the message
        # may be delivered synchronously to the queue by publishing it with
        # mandatory=True
        ch.confirm_delivery()

        # Declare a new queue
        ch.queue_declare(q_name, exclusive=True)

        # Deposit two messages in the queue via default exchange
        ch.basic_publish(exchange='',
                         routing_key=q_name,
                         body='TestBasicNackNoRequeue1',
                         mandatory=True)
        ch.basic_publish(exchange='',
                         routing_key=q_name,
                         body='TestBasicNackNoRequeue2',
                         mandatory=True)

        # Get the messages
        (rx_method, _, rx_body) = ch.basic_get(q_name, auto_ack=False)
        self.assertEqual(rx_body, as_bytes('TestBasicNackNoRequeue1'))

        (rx_method, _, rx_body) = ch.basic_get(q_name, auto_ack=False)
        self.assertEqual(rx_body, as_bytes('TestBasicNackNoRequeue2'))

        # Nack the second message
        ch.basic_nack(rx_method.delivery_tag, requeue=False)

        # Verify that no messages are present in the queue
        self._assert_exact_message_count_with_retries(channel=ch,
                                                      queue=q_name,
                                                      expected_count=0)


class TestBasicNackMultiple(BlockingTestCaseBase):