GitHub PullRequestReviewCommentEvent at 2014-03-27 19:13:48
Repository: kombu
Actor: bmbouter
Payload: {"comment":{"url":"https://api.github.com/repos/pulp/kombu/pulls/comments/11039836","id":11039836,"diff_hunk":"@@ -0,0 +1,1161 @@\n+from __future__ import absolute_import\n+\n+import Queue\n+import socket\n+import threading\n+import time\n+\n+from itertools import count\n+\n+import qpid.messaging.exceptions\n+\n+import kombu.five\n+from kombu.transport.qpid import QpidMessagingExceptionHandler, QoS, Message\n+from kombu.transport.qpid import Channel, FDShimThread, FDShim\n+from kombu.transport.qpid import Connection, Transport\n+from kombu.transport.virtual import Base64\n+from kombu.utils.compat import OrderedDict\n+from kombu.tests.case import Case, Mock, SkipTest, mask_modules, patch\n+\n+\n+class TestQpidMessagingExceptionHandler(Case):\n+\n+    allowed_string = 'object in use'\n+    not_allowed_string = 'a different string'\n+\n+    def setUp(self):\n+        self.handler = QpidMessagingExceptionHandler(self.allowed_string)\n+\n+    def test_string_stored(self):\n+        \"\"\"Assert that the allowed_exception_string is stored correctly\"\"\"\n+        handler_string = self.handler.allowed_exception_string\n+        self.assertEqual(self.allowed_string, handler_string)\n+\n+    def test_exception_positive(self):\n+        \"\"\"Assert that an exception is silenced if it contains the\n+        allowed_string text\"\"\"\n+        exception_to_raise = Exception(self.allowed_string)\n+        def exception_raise_func():\n+            raise exception_to_raise\n+        decorated_func = self.handler(exception_raise_func)\n+        try:\n+            decorated_func()\n+        except:\n+            self.fail(\"QpidMessagingExceptionHandler allowed an exception \"\n+                      \"to be raised that should have been silenced!\")\n+\n+    def test_exception_negative(self):\n+        \"\"\"Assert that an exception that does not contain the\n+        allowed_string text is properly raised\"\"\"\n+        exception_to_raise = Exception(self.not_allowed_string)\n+        def exception_raise_func():\n+            raise exception_to_raise\n+        decorated_func = self.handler(exception_raise_func)\n+        self.assertRaises(Exception, decorated_func)\n+\n+\n+class TestQoS(Case):\n+\n+    def mock_message_factory(self):\n+        m_delivery_tag = self.delivery_tag_generator.next()\n+        m = 'message %s' % m_delivery_tag\n+        return (m, m_delivery_tag)\n+\n+    def add_n_messages_to_qos(self, n, qos):\n+        for i in range(n):\n+            self.add_message_to_qos(qos)\n+\n+    def add_message_to_qos(self, qos):\n+        m, m_delivery_tag = self.mock_message_factory()\n+        qos.append(m, m_delivery_tag)\n+\n+    def setUp(self):\n+        self.qos_no_limit = QoS()\n+        self.qos_limit_2 = QoS(prefetch_count=2)\n+        self.delivery_tag_generator = count(1)\n+\n+    def test_init_no_params(self):\n+        \"\"\"Check that internal state is correct after initialization\"\"\"\n+        self.assertEqual(self.qos_no_limit.prefetch_count, 0)\n+        self.assertIsInstance(self.qos_no_limit._not_yet_acked, OrderedDict)\n+\n+    def test_init_with_params(self):\n+        \"\"\"Check that internal state is correct after initialization with\n+        prefetch_count\"\"\"\n+        self.assertEqual(self.qos_limit_2.prefetch_count, 2)\n+\n+    def test_can_consume_no_limit(self):\n+        \"\"\"can_consume shall always return True with no prefetch limits\"\"\"\n+        self.assertTrue(self.qos_no_limit.can_consume())\n+        self.add_n_messages_to_qos(3, self.qos_no_limit)\n+        self.assertTrue(self.qos_no_limit.can_consume())\n+\n+    def test_can_consume_with_limit(self):\n+        \"\"\"can_consume shall return False only when the QoS holds\n+        prefetch_count number of messages\"\"\"\n+        self.assertTrue(self.qos_limit_2.can_consume())\n+        self.add_n_messages_to_qos(2, self.qos_limit_2)\n+        self.assertFalse(self.qos_limit_2.can_consume())\n+\n+    def test_can_consume_max_estimate_no_limit(self):\n+        \"\"\"can_consume_max_estimate shall always return 1 with no prefetch\n+        limits\"\"\"\n+        self.assertEqual(self.qos_no_limit.can_consume_max_estimate(), 1)\n+        self.add_n_messages_to_qos(3, self.qos_no_limit)\n+        self.assertEqual(self.qos_no_limit.can_consume_max_estimate(), 1)\n+\n+    def test_can_consume_max_estimate_with_limit(self):\n+        \"\"\"while prefetch limits are enabled, can_consume_max_estimate shall\n+        return (prefetch_limit - #messages) as the number of messages is\n+        incremented from 0 to prefetch_limit\"\"\"\n+        self.assertEqual(self.qos_limit_2.can_consume_max_estimate(), 2)\n+        self.add_message_to_qos(self.qos_limit_2)\n+        self.assertEqual(self.qos_limit_2.can_consume_max_estimate(), 1)\n+        self.add_message_to_qos(self.qos_limit_2)\n+        self.assertEqual(self.qos_limit_2.can_consume_max_estimate(), 0)\n+\n+    def test_append(self):\n+        \"\"\"Append two messages and check inside the QoS object that they\n+        were put into the internal data structures correctly\"\"\"\n+        qos = self.qos_no_limit\n+        m1, m1_tag = self.mock_message_factory()\n+        m2, m2_tag = self.mock_message_factory()\n+        qos.append(m1, m1_tag)\n+        length_not_yet_acked = len(qos._not_yet_acked)\n+        self.assertEqual(length_not_yet_acked, 1)\n+        checked_message1 = qos._not_yet_acked[m1_tag]\n+        self.assertIs(m1, checked_message1)\n+        qos.append(m2, m2_tag)\n+        length_not_yet_acked = len(qos._not_yet_acked)\n+        self.assertEqual(length_not_yet_acked, 2)\n+        checked_message2 = qos._not_yet_acked[m2_tag]\n+        self.assertIs(m2, checked_message2)\n+\n+    def test_get(self):\n+        \"\"\"Append two messages, and use get to receive them\"\"\"\n+        qos = self.qos_no_limit\n+        m1, m1_tag = self.mock_message_factory()\n+        m2, m2_tag = self.mock_message_factory()\n+        qos.append(m1, m1_tag)\n+        qos.append(m2, m2_tag)\n+        message1 = qos.get(m1_tag)\n+        message2 = qos.get(m2_tag)\n+        self.assertIs(m1, message1)\n+        self.assertIs(m2, message2)\n+\n+    def test_ack(self):\n+        \"\"\"Load a mock message, ack the message, and ensure the right\n+        call is made to the acknowledge method in the qpid.messaging client\n+        library\"\"\"\n+        message = Mock()\n+        qos = self.qos_no_limit\n+        qos.append(message, 1)\n+        qos.ack(1)\n+        message._receiver.session.acknowledge.assert_called_with(\n+            message=message)\n+\n+    @patch('qpid.messaging.Disposition')\n+    @patch('qpid.messaging.RELEASED')\n+    def test_ack_requeue_true(self, mock_RELEASED, mock_QpidDisposition):\n+        \"\"\"Load a mock message, reject the message with requeue=True,\n+        and ensure the right call to acknowledge is made\"\"\"\n+        message = Mock()\n+        mock_QpidDisposition.return_value='disposition'\n+        qos = self.qos_no_limit\n+        qos.append(message, 1)\n+        qos.reject(1, requeue=True)\n+        mock_QpidDisposition.assert_called_with(mock_RELEASED)\n+        message._receiver.session.acknowledge.assert_called_with(\n+            message=message, disposition='disposition')\n+\n+    @patch('qpid.messaging.Disposition')\n+    @patch('qpid.messaging.REJECTED')\n+    def test_ack_requeue_false(self, mock_REJECTED, mock_QpidDisposition):\n+        \"\"\"Load a mock message, reject the message with requeue=False,\n+        and ensure the right call to acknowledge is made\"\"\"\n+        message = Mock()\n+        mock_QpidDisposition.return_value='disposition'\n+        qos = self.qos_no_limit\n+        qos.append(message, 1)\n+        qos.reject(1, requeue=False)\n+        mock_QpidDisposition.assert_called_with(mock_REJECTED)\n+        message._receiver.session.acknowledge.assert_called_with(\n+            message=message, disposition='disposition')\n+\n+\n+class TestFDShimThread(Case):\n+\n+    def setUp(self):\n+        self.mock_create_qpid_connection = Mock()\n+        self.mock_session = Mock()\n+        self.mock_create_qpid_connection().session = Mock(\n+            return_value=self.mock_session)\n+        self.mock_receiver = Mock()\n+        self.mock_session.receiver = Mock(return_value=self.mock_receiver)\n+        self.mock_queue = Mock()\n+        self.mock_delivery_queue = Mock()\n+        self.my_thread = FDShimThread(self.mock_create_qpid_connection,\n+                                      self.mock_queue,\n+                                      self.mock_delivery_queue)\n+        self.my_thread.daemon = True\n+\n+    def test_init_variables(self):\n+        \"\"\"Test that all simple init params are internally stored\n+        correctly\"\"\"\n+        self.assertIs(self.my_thread._queue, self.mock_queue)\n+        self.assertIs(self.my_thread._delivery_queue,\n+                      self.mock_delivery_queue)\n+        self.assertFalse(self.my_thread._is_killed)\n+\n+    def test_session_gets_made(self):\n+        \"\"\"Test that a session is created\"\"\"\n+        self.mock_create_qpid_connection().session.assert_called_with()\n+\n+    def test_session_gets_stored(self):\n+        \"\"\"Test that a session is saved properly after being created\"\"\"\n+        self.assertIs(self.mock_session, self.my_thread._session)\n+\n+    def test_receiver_gets_made(self):\n+        \"\"\"Test that a receiver is created listening on the queue\"\"\"\n+        self.mock_session.receiver.assert_called_with(self.mock_queue)\n+\n+    def test_receiver_gets_stored(self):\n+        \"\"\"Test that a receiver is saved properly after being created\"\"\"\n+        self.assertIs(self.mock_receiver, self.my_thread._receiver)\n+\n+    def test_kill(self):\n+        \"\"\"Start a thread, and then kill it using the kill() method. Ensure\n+        that it exits properly within the expected amount of time.\"\"\"\n+        self.my_thread.start()\n+        self.my_thread.kill()\n+        self.my_thread.join(timeout=FDShimThread.block_timeout)\n+        self.assertFalse(self.my_thread.is_alive())\n+\n+    def test_receiver_cleanup(self):\n+        \"\"\"Ensure that when a thread exits normally after a call to kill()\n+        that close() is called on the receiver.\n+        \"\"\"\n+        self.my_thread.start()\n+        self.my_thread.kill()\n+        self.my_thread.join(timeout=FDShimThread.block_timeout)\n+        self.mock_receiver.close.assert_called_with()\n+\n+    def test_session_cleanup(self):\n+        \"\"\"Ensure that when a thread exits normally after a call to kill()\n+        that close() is called on the session.\n+        \"\"\"\n+        self.my_thread.start()\n+        self.my_thread.kill()\n+        self.my_thread.join(timeout=FDShimThread.block_timeout)\n+        self.mock_session.close.assert_called_with()\n+\n+    def test_call_to_fetch_raise_empty(self):\n+        \"\"\"Ensure the call to fetch() occurs, and with the proper timeout.\n+        Raises an Empty exception.\"\"\"\n+        QpidEmpty = qpid.messaging.exceptions.Empty\n+        self.mock_receiver.fetch = Mock(side_effect=QpidEmpty())\n+        self.my_thread.start()\n+        time.sleep(1)\n+        self.mock_receiver.fetch.assert_called_with(\n+            timeout=FDShimThread.block_timeout)\n+\n+    def test_call_to_fetch_return_message(self):\n+        \"\"\"Ensure the call to fetch() occurs, that the response bundle is\n+        built correctly is called as the only argument to receiver.put()\"\"\"\n+        mock_response = Mock()\n+        mock_source = Mock()\n+        mock_put = Mock()\n+        response_bundle = (mock_source, mock_response)\n+        self.mock_receiver.fetch = Mock(return_value=mock_response)\n+        self.mock_receiver.source = mock_source\n+        self.mock_delivery_queue.put = mock_put\n+        self.my_thread.start()\n+        time.sleep(1)\n+        self.mock_receiver.fetch.assert_called_with(\n+            timeout=FDShimThread.block_timeout)\n+        mock_put.assert_called_with(response_bundle)\n+\n+\n+class TestFDShim(Case):\n+    def setUp(self):\n+        self.mock_queue_from_fdshim = Mock()\n+        self.mock_delivery_queue = Mock()\n+        self.my_fdshim = FDShim(self.mock_queue_from_fdshim,\n+                               self.mock_delivery_queue)\n+        self.my_thread = threading.Thread(\n+            target=self.my_fdshim.monitor_consumers)\n+        self.my_thread.daemon = True\n+\n+    def test_init_variables(self):\n+        \"\"\"Test that all simple init params are internally stored\n+        correctly\"\"\"\n+        self.assertIs(self.my_fdshim.queue_from_fdshim,\n+                      self.mock_queue_from_fdshim)\n+        self.assertIs(self.my_fdshim.delivery_queue,\n+                      self.mock_delivery_queue)\n+        self.assertFalse(self.my_fdshim._is_killed)\n+\n+    def test_kill(self):\n+        \"\"\"Start a thread, and then kill it using the kill() method. Ensure\n+        that it exits properly within the expected amount of time.\"\"\"\n+        self.my_thread.start()\n+        self.my_fdshim.kill()\n+        self.my_thread.join(timeout=3)\n+        self.assertFalse(self.my_thread.is_alive())\n+\n+    def test_call_to_get_raise_empty(self):\n+        \"\"\"Ensure the call to delivery_queue.get() occurs, and with\n+        block=True.  Raises an Queue.Empty exception.\"\"\"\n+        self.mock_delivery_queue.get = Mock(side_effect=Queue.Empty())\n+        self.my_thread.start()\n+        time.sleep(1)\n+        self.mock_delivery_queue.get.assert_called_with(block=True)\n+\n+    @patch('os.write')\n+    def test_call_to_get_return_response_bundle(self, write_method):\n+        \"\"\"Ensure the call to get() occurs, and when the response bundle\n+        is received from delivery_queue, that the response bundle is put\n+        into queue_from_fdshim using put().\n+\n+        This method patches os.write to ensure that a '0' is written to the\n+        _w file descriptor attribute of FDShim.\"\"\"\n+        response_bundle = Mock()\n+        self.mock_delivery_queue.get = Mock(return_value=response_bundle)\n+        self.my_thread.start()\n+        time.sleep(1)\n+        self.mock_delivery_queue.get.assert_called_with(block=True)\n+        self.mock_queue_from_fdshim.put.assert_called_with(\n+            response_bundle)\n+        write_method.assert_called_with(self.my_fdshim._w, '0')\n+\n+\n+class TestConnection(Case):\n+\n+    def setUp(self):\n+        self.connection_options = {'host': 'localhost',\n+                                   'port': 5672,\n+                                   'username': 'guest',\n+                                   'password': 'guest',\n+                                   'transport': 'tcp',\n+                                   'timeout': 10,\n+                                   'sasl_mechanisms': 'PLAIN'}\n+        self.my_connection = Connection(**self.connection_options)\n+\n+    def test_init_variables(self):\n+        \"\"\"Test that all simple init params are internally stored\n+        correctly\"\"\"\n+        self.assertDictEqual(self.connection_options,\n+                             self.my_connection.connection_options)\n+        self.assertIsInstance(self.my_connection.channels, list)\n+        self.assertIsInstance(self.my_connection._callbacks, dict)\n+\n+    def test_verify_connection_class_attributes(self):\n+        \"\"\"Verify that Channel class attribute is set correctly\"\"\"\n+        self.assertEqual(Channel, Connection.Channel)\n+\n+    @patch('qpid.messaging.Connection')\n+    def test_create_qpid_connection(self, QpidConnection):\n+        \"\"\"Test that create_qpid_connection calls establish with the\n+        connection_options, and then returns the result.\"\"\"\n+        new_connection = 'connection'\n+        QpidConnection.establish = Mock(return_value=new_connection)\n+        conn_from_func = self.my_connection.create_qpid_connection()\n+        QpidConnection.establish.assert_called_with(\n+            **self.connection_options)\n+        self.assertEqual(new_connection, conn_from_func)\n+\n+    def test_close_channel_exists(self):\n+        \"\"\"Test that calling close_channel() with a valid channel removes\n+        the channel from self.channels and sets channel.connection to None\"\"\"\n+        mock_channel = Mock()\n+        self.my_connection.channels = [mock_channel]\n+        mock_channel.connection = True\n+        self.my_connection.close_channel(mock_channel)\n+        self.assertEqual(self.my_connection.channels, [])\n+        self.assertIsNone(mock_channel.connection)\n+\n+\n+    def test_close_channel_does_not_exist(self):\n+        \"\"\"Test that calling close_channel() with an invalid channel does\n+        not raise a ValueError and sets channel.connection to None.\"\"\"\n+        self.my_connection.channels = Mock()\n+        self.my_connection.channels.remove = Mock(side_effect=ValueError())\n+        mock_channel = Mock()\n+        mock_channel.connection = True\n+        self.my_connection.close_channel(mock_channel)\n+        self.assertIsNone(mock_channel.connection)\n+\n+\n+class TestChannel(Case):\n+\n+    @patch('qpidtoollibs.BrokerAgent')\n+    def setUp(self, mock_BrokerAgent):\n+        self.mock_connection = Mock()\n+        self.mock_qpid_connection = Mock()\n+        self.mock_qpid_session = Mock()\n+        self.mock_qpid_connection.session = \\\n+            Mock(return_value=self.mock_qpid_session)\n+        self.mock_connection.create_qpid_connection = \\\n+            Mock(return_value=self.mock_qpid_connection)\n+        self.mock_transport = Mock()\n+        self.mock_delivery_queue = Mock()\n+        self.mock_broker = Mock()\n+        self.mock_Message = Mock()\n+        self.mock_BrokerAgent = mock_BrokerAgent\n+        mock_BrokerAgent.return_value = self.mock_broker\n+        self.my_channel = Channel(self.mock_connection,\n+                                  self.mock_transport,\n+                                  self.mock_delivery_queue)\n+        self.my_channel.Message = self.mock_Message\n+\n+    def test_verify_QoS_class_attribute(self):\n+        \"\"\"Verify that the class attribute QoS refers to the QoS object\"\"\"\n+        self.assertIs(QoS, Channel.QoS)\n+\n+    def test_verify_Message_class_attribute(self):\n+        \"\"\"Verify that the class attribute Message refers to the Message\n+        object\"\"\"\n+        self.assertIs(Message, Channel.Message)\n+\n+    def test_body_encoding_class_attribute(self):\n+        \"\"\"Verify that the class attribute body_encoding is set to base64\"\"\"\n+        self.assertEqual('base64', Channel.body_encoding)\n+\n+    def test_codecs_class_attribute(self):\n+        \"\"\"Verify that the codecs class attribute has a correct key and\n+        value\"\"\"\n+        self.assertIsInstance(Channel.codecs, dict)\n+        self.assertIn('base64', Channel.codecs)\n+        self.assertIsInstance(Channel.codecs['base64'], Base64)\n+\n+    def test_delivery_tags(self):\n+        \"\"\"Test that _delivery_tags is using itertools\"\"\"\n+        self.assertIsInstance(Channel._delivery_tags, count)\n+\n+    def test_init_variables(self):\n+        \"\"\"Test init variables\"\"\"\n+        self.assertIs(self.mock_connection, self.my_channel.connection)\n+        self.assertIs(self.mock_transport, self.my_channel.transport)\n+        self.assertIs(self.mock_delivery_queue,\n+                      self.my_channel.delivery_queue)\n+        self.assertIsInstance(self.my_channel._tag_to_queue, dict)\n+        self.assertIsInstance(self.my_channel._consumer_threads, dict)\n+        self.assertIs(self.mock_qpid_session, self.my_channel._qpid_session)\n+        self.assertIs(self.mock_broker, self.my_channel._broker)\n+        self.assertIsNone(self.my_channel._qos)\n+        self.assertIsInstance(self.my_channel._consumers, set)\n+        self.assertFalse(self.my_channel.closed)\n+        self.mock_connection.create_qpid_connection.assert_called_with()\n+        self.mock_qpid_connection.session.assert_called_with()\n+        self.mock_BrokerAgent.assert_called_with(self.mock_qpid_connection)\n+\n+    def test_get(self):\n+        \"\"\"Test _get() for return value from call to receiver, and check\n+        that receiver has closed() called on it.\"\"\"\n+        mock_queue = Mock()\n+        mock_rx = Mock()\n+        mock_message = Mock()\n+        mock_rx.fetch.return_value = mock_message\n+        self.mock_qpid_session.receiver.return_value = mock_rx\n+        result = self.my_channel._get(mock_queue)\n+        self.mock_qpid_session.receiver.assert_called_with(mock_queue)\n+        mock_rx.close.assert_called_with()\n+        self.assertIs(mock_message, result)\n+\n+    @patch('qpid.messaging.Message')\n+    def test_put_queue(self, mock_qpid_Message_obj):\n+        \"\"\"Test putting a messages directly into a queue.\"\"\"\n+        mock_routing_key = 'routingkey'\n+        mock_message = Mock()\n+        mock_sender = Mock()\n+        mock_new_qpid_message_obj = Mock()\n+        self.mock_qpid_session.sender.return_value = mock_sender\n+        mock_qpid_Message_obj.return_value = mock_new_qpid_message_obj\n+        self.my_channel._put(mock_routing_key, mock_message)\n+        address_string = '%s; {assert: always, node: {type: queue}}' % \\\n+                         mock_routing_key\n+        self.mock_qpid_session.sender.assert_called_with(address_string)\n+        mock_qpid_Message_obj.assert_called_with(content=mock_message,\n+                                                 subject=None)\n+        mock_sender.send.assert_called_with(mock_new_qpid_message_obj,\n+                                            sync=True)\n+        mock_sender.close.assert_called_with()\n+\n+    @patch('qpid.messaging.Message')\n+    def test_put_exchange(self, mock_qpid_Message_obj):\n+        \"\"\"Test putting a messages directly into an exchange.\"\"\"\n+        mock_routing_key = 'routingkey'\n+        mock_exchange_name = 'myexchange'\n+        mock_message = Mock()\n+        mock_sender = Mock()\n+        mock_new_qpid_message_obj = Mock()\n+        self.mock_qpid_session.sender.return_value = mock_sender\n+        mock_qpid_Message_obj.return_value = mock_new_qpid_message_obj\n+        self.my_channel._put(mock_routing_key, mock_message,\n+                             mock_exchange_name)\n+        address_string = '%s/%s; {assert: always, node: {type: topic}}' % \\\n+                         (mock_exchange_name, mock_routing_key)\n+        self.mock_qpid_session.sender.assert_called_with(address_string)\n+        mock_qpid_Message_obj.assert_called_with(content=mock_message,\n+                                                 subject=mock_routing_key)\n+        mock_sender.send.assert_called_with(mock_new_qpid_message_obj,\n+                                            sync=True)\n+        mock_sender.close.assert_called_with()\n+\n+    def test_purge(self):\n+        \"\"\"Test purging a queue that has messages, and verify the return\n+        value\"\"\"\n+        message_count = 5\n+        mock_queue = Mock()\n+        mock_queue_to_purge = Mock()\n+        mock_queue_to_purge.values = {'msgDepth': message_count}\n+        self.mock_broker.getQueue.return_value = mock_queue_to_purge\n+        result = self.my_channel._purge(mock_queue)\n+        self.mock_broker.getQueue.assert_called_with(mock_queue)\n+        mock_queue_to_purge.purge.assert_called_with(message_count)\n+        self.assertEqual(message_count, result)\n+\n+    def test_size(self):\n+        \"\"\"Test getting the number of messages in a queue specified by\n+        name and returning them.\"\"\"\n+        message_count = 5\n+        mock_queue = Mock()\n+        mock_queue_to_check = Mock()\n+        mock_queue_to_check.values = {'msgDepth': message_count}\n+        self.mock_broker.getQueue.return_value = mock_queue_to_check\n+        result = self.my_channel._size(mock_queue)\n+        self.mock_broker.getQueue.assert_called_with(mock_queue)\n+        self.assertEqual(message_count, result)\n+\n+    def test_delete(self):\n+        \"\"\"Test deleting a queue calls purge and delQueue with queue name\"\"\"\n+        mock_queue = Mock()\n+        self.my_channel._purge = Mock()\n+        result = self.my_channel._delete(mock_queue)\n+        self.my_channel._purge.assert_called_with(mock_queue)\n+        self.mock_broker.delQueue.assert_called_with(mock_queue)\n+        self.assertIsNone(result)\n+\n+    def test_new_queue_raises_exception_and_silenced(self):\n+        \"\"\"Test new queue, where an exception is raised and then silenced\"\"\"\n+        mock_queue = Mock()\n+        self.mock_broker.addQueue.side_effect = \\\n+            Exception('The foo object already exists.')\n+        result = self.my_channel._new_queue(mock_queue)\n+        self.mock_broker.addQueue.assert_called_with(mock_queue)\n+        self.assertIsNone(result)\n+\n+    def test_new_queue_raises_exception_not_silenced(self):\n+        \"\"\"Test new queue, where an exception is raised and not silenced\"\"\"\n+        unique_exception = Exception('This exception should not be silenced')\n+        mock_queue = Mock()\n+        self.mock_broker.addQueue.side_effect = unique_exception\n+        self.assertRaises(unique_exception.__class__,\n+                          self.my_channel._new_queue,\n+                          mock_queue)\n+        self.mock_broker.addQueue.assert_called_with(mock_queue)\n+\n+    def test_new_queue_no_exception(self):\n+        \"\"\"Test creation of a new queue, where an exception is NOT raised\"\"\"\n+        mock_queue = Mock()\n+        result = self.my_channel._new_queue(mock_queue)\n+        self.mock_broker.addQueue.assert_called_with(mock_queue)\n+        self.assertIsNone(result)\n+\n+    def test_has_queue_true(self):\n+        \"\"\"Test checking if a queue exists, and it does\"\"\"\n+        mock_queue = Mock()\n+        self.mock_broker.getQueue.return_value = True\n+        result = self.my_channel._has_queue(mock_queue)\n+        self.assertTrue(result)\n+\n+    def test_has_queue_false(self):\n+        \"\"\"Test checking if a queue exists, and it does not\"\"\"\n+        mock_queue = Mock()\n+        self.mock_broker.getQueue.return_value = False\n+        result = self.my_channel._has_queue(mock_queue)\n+        self.assertFalse(result)\n+\n+    @patch('amqp.protocol.queue_declare_ok_t')\n+    def test_queue_declare(self, mock_queue_declare_ok_t):\n+        \"\"\"Test the declaration of a queue, and the return value\"\"\"\n+        mock_queue = Mock()\n+        message_count = 5\n+        self.my_channel._new_queue = Mock()\n+        self.my_channel._size = Mock(return_value=message_count)\n+        mock_queue_declare_ok_t.return_value = Mock()\n+        result = self.my_channel.queue_declare(mock_queue)\n+        self.my_channel._new_queue.assert_called_with(mock_queue)\n+        self.my_channel._size.assert_called_with(mock_queue)\n+        mock_queue_declare_ok_t.assert_called_with(mock_queue,\n+                                                   message_count, 0)\n+        self.assertIs(mock_queue_declare_ok_t.return_value, result)\n+\n+    def test_queue_delete_if_empty_param(self):\n+        \"\"\"Test the deletion of a queue with if_empty=True\"\"\"\n+        mock_queue = Mock()\n+        self.my_channel._has_queue = Mock(return_value=True)\n+        self.my_channel._size = Mock(return_value=5)\n+        result = self.my_channel.queue_delete(mock_queue, if_empty=True)\n+        self.my_channel._has_queue.assert_called_with(mock_queue)\n+        self.my_channel._size.assert_called_with(mock_queue)\n+        self.assertIsNone(result)\n+\n+    def test_queue_delete_if_unused_param(self):\n+        \"\"\"Test the deletion of a queue with if_unused=True\"\"\"\n+        mock_queue = Mock()\n+        mock_queue_obj = Mock()\n+        mock_queue_attributes = {'consumerCount': 5}\n+        mock_queue_obj.getAttributes.return_value = mock_queue_attributes\n+        self.my_channel._has_queue = Mock(return_value=True)\n+        self.my_channel._size = Mock(return_value=5)\n+        self.mock_broker.getQueue.return_value = mock_queue_obj\n+        result = self.my_channel.queue_delete(mock_queue, if_unused=True)\n+        self.assertIsNone(result)\n+\n+    def test_queue_delete(self):\n+        \"\"\"Test the deletion of a queue\"\"\"\n+        mock_queue = Mock()\n+        mock_queue_obj = Mock()\n+        mock_queue_attributes = {'consumerCount': 5}\n+        mock_queue_obj.getAttributes.return_value = mock_queue_attributes\n+        self.my_channel._has_queue = Mock(return_value=True)\n+        self.my_channel._size = Mock(return_value=5)\n+        self.my_channel._delete = Mock()\n+        self.mock_broker.getQueue.return_value = mock_queue_obj\n+        result = self.my_channel.queue_delete(mock_queue)\n+        self.my_channel._delete.assert_called_with(mock_queue)\n+        self.assertIsNone(result)\n+\n+    def test_exchange_declare_raises_exception_and_silenced(self):\n+        \"\"\"Create exchange where an exception is raised and then silenced\"\"\"\n+        mock_exchange = Mock()\n+        self.mock_broker.addExchange.side_effect = \\\n+            Exception('The foo object already exists.')\n+        self.my_channel.exchange_declare()\n+\n+    def test_exchange_declare_raises_exception_not_silenced(self):\n+        \"\"\"Create Exchange where an exception is raised and not silenced\"\"\"\n+        mock_exchange = Mock()\n+        unique_exception = Exception('This exception should not be silenced')\n+        self.mock_broker.addExchange.side_effect = unique_exception\n+        self.assertRaises(unique_exception.__class__,\n+                          self.my_channel.exchange_declare)\n+\n+    def test_exchange_declare(self):\n+        \"\"\"Create Exchange where an exception is NOT raised\"\"\"\n+        mock_exchange = Mock()\n+        mock_type = Mock()\n+        mock_durable = Mock()\n+        options = {'durable': mock_durable}\n+        result = self.my_channel.exchange_declare(mock_exchange,\n+                                                  mock_type,\n+                                                  mock_durable)\n+        self.mock_broker.addExchange.assert_called_with(mock_type,\n+                                                        mock_exchange,\n+                                                        options)\n+        self.assertIsNone(result)\n+\n+    def test_exchange_delete(self):\n+        \"\"\"Test the deletion of an exchange by name\"\"\"\n+        mock_exchange = Mock()\n+        result = self.my_channel.exchange_delete(mock_exchange)\n+        self.mock_broker.delExchange.assert_called_with(mock_exchange)\n+        self.assertIsNone(result)\n+\n+    def test_after_reply_message_received_exception_and_silenced(self):\n+        \"\"\"Call after_reply_message_received exception raised and silenced\"\"\"\n+        mock_queue = Mock()\n+        normal_exception = Exception('The queue in use is foo')\n+        self.my_channel._delete = Mock(side_effect=normal_exception)\n+        self.my_channel.after_reply_message_received(mock_queue)\n+\n+    def test_after_reply_message_received_exception_not_silenced(self):\n+        \"\"\"Call after_reply_message_received exception raised not silenced\"\"\"\n+        mock_queue = Mock()\n+        abnormal_exception = Exception('This exception should pass through')\n+        self.my_channel._delete = Mock(side_effect=abnormal_exception)\n+        self.assertRaises(abnormal_exception.__class__,\n+                          self.my_channel.after_reply_message_received,\n+                          mock_queue)\n+\n+    def test_after_reply_message_received(self):\n+        \"\"\"Call after_reply_message_received\"\"\"\n+        mock_queue = Mock()\n+        self.my_channel._delete = Mock()\n+        self.my_channel.after_reply_message_received(mock_queue)\n+        self.my_channel._delete.assert_called_with(mock_queue)\n+\n+    def test_queue_bind(self):\n+        \"\"\"Test binding a queue to an exchange using a routing key\"\"\"\n+        mock_queue = Mock()\n+        mock_exchange = Mock()\n+        mock_routing_key = Mock()\n+        self.my_channel.queue_bind(mock_queue,\n+                                   mock_exchange,\n+                                   mock_routing_key)\n+        self.mock_broker.bind.assert_called_with(mock_exchange,\n+                                                 mock_queue,\n+                                                 mock_routing_key)\n+\n+    def test_queue_unbind(self):\n+        \"\"\"Test unbinding a queue from an exchange using a routing key\"\"\"\n+        mock_queue = Mock()\n+        mock_exchange = Mock()\n+        mock_routing_key = Mock()\n+        self.my_channel.queue_unbind(mock_queue,\n+                                     mock_exchange,\n+                                     mock_routing_key)\n+        self.mock_broker.unbind.assert_called_with(mock_exchange,\n+                                                   mock_queue,\n+                                                   mock_routing_key)\n+\n+    def test_queue_purge(self):\n+        \"\"\"Test purging a queue by name\"\"\"\n+        mock_queue = Mock()\n+        purge_result = Mock()\n+        self.my_channel._purge = Mock(return_value=purge_result)\n+        result = self.my_channel.queue_purge(mock_queue)\n+        self.my_channel._purge.assert_called_with(mock_queue)\n+        self.assertIs(purge_result, result)\n+\n+    def test_basic_get_happy_path(self):\n+        \"\"\"Test a basic_get with the most common case\"\"\"\n+        mock_queue = Mock()\n+        mock_qpid_message = Mock()\n+        new_message = Mock()\n+        self.my_channel._get = Mock(return_value=mock_qpid_message)\n+        self.mock_Message.return_value = new_message\n+        result = self.my_channel.basic_get(mock_queue)\n+        self.mock_qpid_session.acknowledge.assert_called_with(\n+            message=mock_qpid_message)\n+        self.assertIs(new_message, result)\n+\n+    def test_basic_get_no_ack_equals_True(self):\n+        \"\"\"Test a basic_get where no_ack equals True\"\"\"\n+        mock_queue = Mock()\n+        mock_qpid_message = Mock()\n+        new_message = Mock()\n+        self.my_channel._get = Mock(return_value=mock_qpid_message)\n+        self.mock_Message.return_value = new_message\n+        self.mock_qpid_session.acknowledge.side_effect = \\\n+            Exception('Acknowledge should not be called')","path":"kombu/tests/transport/test_qpid.py","position":null,"original_position":742,"commit_id":"96b97e79dec625c6884bfc743a9eba098c7f4d3b","original_commit_id":"e53f7bee35b8ea2a23bf13689a77c5538a0d4728","user":{"login":"bmbouter","id":180319,"avatar_url":"https://avatars.githubusercontent.com/u/180319?","gravatar_id":"9045966c5a1abf470dcd0d316ae32d35","url":"https://api.github.com/users/bmbouter","html_url":"https://github.com/bmbouter","followers_url":"https://api.github.com/users/bmbouter/followers","following_url":"https://api.github.com/users/bmbouter/following{/other_user}","gists_url":"https://api.github.com/users/bmbouter/gists{/gist_id}","starred_url":"https://api.github.com/users/bmbouter/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/bmbouter/subscriptions","organizations_url":"https://api.github.com/users/bmbouter/orgs","repos_url":"https://api.github.com/users/bmbouter/repos","events_url":"https://api.github.com/users/bmbouter/events{/privacy}","received_events_url":"https://api.github.com/users/bmbouter/received_events","type":"User","site_admin":false},"body":"This is a great point, and it works!  Fixed with 96b97e79dec625c6884bfc743a9eba098c7f4d3b","created_at":"2014-03-27T19:13:48Z","updated_at":"2014-03-27T19:13:48Z","html_url":"https://github.com/pulp/kombu/pull/1#discussion_r11039836","pull_request_url":"https://api.github.com/repos/pulp/kombu/pulls/1","_links":{"self":{"href":"https://api.github.com/repos/pulp/kombu/pulls/comments/11039836"},"html":{"href":"https://github.com/pulp/kombu/pull/1#discussion_r11039836"},"pull_request":{"href":"https://api.github.com/repos/pulp/kombu/pulls/1"}}}}