Repository: pika/pika

Issue:
Channel.basic_consume doesn't support consume-ok/completion callback
[Channel.basic_consume](https://github.com/pika/pika/blob/3d3b95d31b67dfeaf5ef43650c162e25169336e6/pika/channel.py#L263) doesn't provide an argument for specifying the completion callback to be called upon [consume-ok](https://www.rabbitmq.com/amqp-0-9-1-reference.html#basic.consume-ok). (the callback arg that it presently supports is for consuming incoming messages).

A client's logic might need to know that the `basic-consume` request succeeded before initiating other operations (e.g., when requesting exclusive access). Lack of the completion callback prevents the application from doing so easily.

Implementation notes:
1. Rename `Channel.basic_consume`'s consumer callback (presently named `callback`) to `consumer_callback`, `message_callback`, `message_rx_callback`, or similar, and name the completion callback `callback` to follow the recently-established convention for naming of completion callbacks. Also rename the consumer callback arg in `BlockingChannel.basic_consume` for consistency with `Channel.basic_consume`.
2. If completion callback isn't specified, auto-generate and return the consumer tag  following the legacy behavior. If completion callback is specified, return None from `Channel.basic_consume` and let the server generate the consumer tag and return it via [consume-ok](https://www.rabbitmq.com/amqp-0-9-1-reference.html#basic.consume-ok). Update the docstring in `Channel.basic_consume` to reflect this.
3. Make sure pika's [Channel._generate_consumer_tag](https://github.com/pika/pika/blob/3d3b95d31b67dfeaf5ef43650c162e25169336e6/pika/channel.py#L323) doesn't generate consumer tags that can potentially collide with server-generated ones by prefixing "pika" into the consumer tag name that `Channel._generate_consumer_tag` generates.

Discussion:
codecov[bot]: # [Codecov](https://codecov.io/gh/pika/pika/pull/954?src=pr&el=h1) Report
> Merging [#954](https://codecov.io/gh/pika/pika/pull/954?src=pr&el=desc) into [master](https://codecov.io/gh/pika/pika/commit/3d3b95d31b67dfeaf5ef43650c162e25169336e6?src=pr&el=desc) will **increase** coverage by `0.01%`.
> The diff coverage is `78.57%`.

[![Impacted file tree graph](https://codecov.io/gh/pika/pika/pull/954/graphs/tree.svg?src=pr&token=cJFWQg66l4&width=650&height=150)](https://codecov.io/gh/pika/pika/pull/954?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##           master     #954      +/-   ##
==========================================
+ Coverage   82.33%   82.34%   +0.01%     
==========================================
  Files          19       19              
  Lines        3690     3694       +4     
  Branches      548      549       +1     
==========================================
+ Hits         3038     3042       +4     
  Misses        502      502              
  Partials      150      150
```


| [Impacted Files](https://codecov.io/gh/pika/pika/pull/954?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [pika/channel.py](https://codecov.io/gh/pika/pika/pull/954/diff?src=pr&el=tree#diff-cGlrYS9jaGFubmVsLnB5) | `94.34% <100%> (+0.05%)` | :arrow_up: |
| [pika/adapters/blocking\_connection.py](https://codecov.io/gh/pika/pika/pull/954/diff?src=pr&el=tree#diff-cGlrYS9hZGFwdGVycy9ibG9ja2luZ19jb25uZWN0aW9uLnB5) | `84.72% <62.5%> (ø)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/pika/pika/pull/954?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/pika/pika/pull/954?src=pr&el=footer). Last update [3d3b95d...8ac57ae](https://codecov.io/gh/pika/pika/pull/954?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/pika/adapters/base_connection.py b/pika/adapters/base_connection.py
index ffeeadc..cecd3b7 100644
--- a/pika/adapters/base_connection.py
+++ b/pika/adapters/base_connection.py
@@ -219,7 +219,7 @@ class BaseConnection(connection.Connection):
             error = 'Connection to %s:%s failed: %s' % (sock_addr_tuple[4][0],
                                                         sock_addr_tuple[4][1],
                                                         error)
-            LOGGER.warning(error)
+            LOGGER.error(error)
             return error
 
         # Handle SSL Connection Negotiation
diff --git a/pika/adapters/blocking_connection.py b/pika/adapters/blocking_connection.py
index ea9b6c1..d021e94 100644
--- a/pika/adapters/blocking_connection.py
+++ b/pika/adapters/blocking_connection.py
@@ -705,7 +705,8 @@ class BlockingConnection(object):
         with self._acquire_event_dispatch() as dispatch_acquired:
             # Check if we can actually process pending events
             common_terminator = lambda: bool(dispatch_acquired and
-                (self._channels_pending_dispatch or self._ready_events))
+                                             (self._channels_pending_dispatch or
+                                              self._ready_events))
             if time_limit is None:
                 self._flush_output(common_terminator)
             else:
@@ -954,7 +955,7 @@ class ReturnedMessage(object):
 class _ConsumerInfo(object):
     """Information about an active consumer"""
 
-    __slots__ = ('consumer_tag', 'no_ack', 'callback',
+    __slots__ = ('consumer_tag', 'no_ack', 'on_message_callback',
                  'alternate_event_sink', 'state')
 
     # Consumer states
@@ -963,16 +964,16 @@ class _ConsumerInfo(object):
     TEARING_DOWN = 3
     CANCELLED_BY_BROKER = 4
 
-    def __init__(self, consumer_tag, no_ack, callback=None,
+    def __init__(self, consumer_tag, no_ack, on_message_callback=None,
                  alternate_event_sink=None):
         """
         NOTE: exactly one of callback/alternate_event_sink musts be non-None.
 
         :param str consumer_tag:
         :param bool no_ack: the no-ack value for the consumer
-        :param callable callback: The function for dispatching messages to
+        :param callable on_message_callback: The function for dispatching messages to
             user, having the signature:
-            callback(channel, method, properties, body)
+            on_message_callback(channel, method, properties, body)
                 channel: BlockingChannel
                 method: spec.Basic.Deliver
                 properties: spec.BasicProperties
@@ -983,12 +984,12 @@ class _ConsumerInfo(object):
             `_pending_events` container. Signature:
             alternate_event_sink(evt)
         """
-        assert (callback is None) != (alternate_event_sink is None), (
-            'exactly one of callback/alternate_event_sink must be non-None',
-            callback, alternate_event_sink)
+        assert (on_message_callback is None) != (alternate_event_sink is None), (
+            'exactly one of on_message_callback/alternate_event_sink must be non-None',
+            on_message_callback, alternate_event_sink)
         self.consumer_tag = consumer_tag
         self.no_ack = no_ack
-        self.callback = callback
+        self.on_message_callback = on_message_callback
         self.alternate_event_sink = alternate_event_sink
         self.state = self.SETTING_UP
 
@@ -1316,7 +1317,7 @@ class BlockingChannel(object):
         self._cleanup()
         method = method_frame.method
         raise exceptions.ChannelClosed(method.reply_code,
-                 method.reply_text)
+                                       method.reply_text)
 
     def _on_consumer_cancelled_by_broker(self, method_frame):
         """Called by impl when broker cancels consumer via Basic.Cancel.
@@ -1402,8 +1403,8 @@ class BlockingChannel(object):
 
             if type(evt) is _ConsumerDeliveryEvt:
                 consumer_info = self._consumer_infos[evt.method.consumer_tag]
-                consumer_info.callback(self, evt.method,
-                                       evt.properties, evt.body)
+                consumer_info.on_message_callback(self, evt.method,
+                                                  evt.properties, evt.body)
 
             elif type(evt) is _ConsumerCancellationEvt:
                 del self._consumer_infos[evt.method_frame.method.consumer_tag]
@@ -1500,7 +1501,7 @@ class BlockingChannel(object):
 
     def basic_consume(self,
                       queue,
-                      callback,
+                      on_message_callback,
                       no_ack=False,
                       exclusive=False,
                       consumer_tag=None,
@@ -1520,9 +1521,9 @@ class BlockingChannel(object):
 
         :param queue: The queue from which to consume
         :type queue: str or unicode
-        :param callable callback: Required function for dispatching messages
+        :param callable on_message_callback: Required function for dispatching messages
             to user, having the signature:
-            callback(channel, method, properties, body)
+            on_message_callback(channel, method, properties, body)
                 channel: BlockingChannel
                 method: spec.Basic.Deliver
                 properties: spec.BasicProperties
@@ -1541,17 +1542,17 @@ class BlockingChannel(object):
             consumer_tag is already present.
 
         """
-        if not callable(callback):
-            raise ValueError('callback callback must be callable; got %r'
-                             % callback)
+        if not callable(on_message_callback):
+            raise ValueError('callback on_message_callback must be callable; got %r'
+                             % on_message_callback)
 
         return self._basic_consume_impl(
             queue=queue,
+            on_message_callback=on_message_callback,
             no_ack=no_ack,
             exclusive=exclusive,
             consumer_tag=consumer_tag,
-            arguments=arguments,
-            callback=callback)
+            arguments=arguments)
 
     def _basic_consume_impl(self,
                             queue,
@@ -1559,12 +1560,12 @@ class BlockingChannel(object):
                             exclusive,
                             consumer_tag,
                             arguments=None,
-                            callback=None,
+                            on_message_callback=None,
                             alternate_event_sink=None):
         """The low-level implementation used by `basic_consume` and `consume`.
         See `basic_consume` docstring for more info.
 
-        NOTE: exactly one of callback/alternate_event_sink musts be
+        NOTE: exactly one of on_message_callback/alternate_event_sink musts be
         non-None.
 
         This method has one additional parameter alternate_event_sink over the
@@ -1580,10 +1581,10 @@ class BlockingChannel(object):
             consumer_tag is already present.
 
         """
-        if (callback is None) == (alternate_event_sink is None):
+        if (on_message_callback is None) == (alternate_event_sink is None):
             raise ValueError(
-                ('exactly one of callback/alternate_event_sink must '
-                 'be non-None', callback, alternate_event_sink))
+                ('exactly one of on_message_callback/alternate_event_sink must '
+                 'be non-None', on_message_callback, alternate_event_sink))
 
         if not consumer_tag:
             # Need a consumer tag to register consumer info before sending
@@ -1599,13 +1600,13 @@ class BlockingChannel(object):
         self._consumer_infos[consumer_tag] = _ConsumerInfo(
             consumer_tag,
             no_ack=no_ack,
-            callback=callback,
+            on_message_callback=on_message_callback,
             alternate_event_sink=alternate_event_sink)
 
         try:
             with self._basic_consume_ok_result as ok_result:
                 tag = self._impl.basic_consume(
-                    callback=self._on_consumer_message_delivery,
+                    on_message_callback=self._on_consumer_message_delivery,
                     queue=queue,
                     no_ack=no_ack,
                     exclusive=exclusive,
diff --git a/pika/channel.py b/pika/channel.py
index 2ec16f6..b1ff5cc 100644
--- a/pika/channel.py
+++ b/pika/channel.py
@@ -262,11 +262,12 @@ class Channel(object):
 
     def basic_consume(self,
                       queue,
-                      callback,
+                      on_message_callback,
                       no_ack=False,
                       exclusive=False,
                       consumer_tag=None,
-                      arguments=None):
+                      arguments=None,
+                      callback=None):
         """Sends the AMQP 0-9-1 command Basic.Consume to the broker and binds messages
         for the consumer_tag to the consumer callback. If you do not pass in
         a consumer_tag, one will be automatically generated for you. Returns
@@ -280,8 +281,8 @@ class Channel(object):
         :param queue: The queue to consume from. Use the empty string to specify the
                       most recent server-named queue for this channel.
         :type queue: str or unicode
-        :param callable callback: The function to call when consuming
-            with the signature callback(channel, method, properties, body), where
+        :param callable on_message_callback: The function to call when consuming
+            with the signature on_message_callback(channel, method, properties, body), where
                                 channel: pika.Channel
                                 method: pika.spec.Basic.Deliver
                                 properties: pika.spec.BasicProperties
@@ -292,12 +293,15 @@ class Channel(object):
         :param consumer_tag: Specify your own consumer tag
         :type consumer_tag: str or unicode
         :param dict arguments: Custom key/value pair arguments for the consumer
+        :param callable callback: callback(pika.frame.Method) for method
+          Basic.ConsumeOk.
         :rtype: str
         :raises ValueError:
 
         """
-        self._require_callback(callback)
+        self._require_callback(on_message_callback)
         self._validate_channel()
+        self._validate_rpc_completion_callback(callback)
 
         # If a consumer tag was not passed, create one
         if not consumer_tag:
@@ -309,14 +313,17 @@ class Channel(object):
         if no_ack:
             self._consumers_with_noack.add(consumer_tag)
 
-        self._consumers[consumer_tag] = callback
+        self._consumers[consumer_tag] = on_message_callback
+
+        rpc_callback = self._on_eventok if callback is None else callback
+
         self._rpc(spec.Basic.Consume(queue=queue,
                                      consumer_tag=consumer_tag,
                                      no_ack=no_ack,
                                      exclusive=exclusive,
                                      arguments=arguments or dict()),
-                  self._on_eventok, [(spec.Basic.ConsumeOk,
-                                      {'consumer_tag': consumer_tag})])
+                  rpc_callback, [(spec.Basic.ConsumeOk,
+                                 {'consumer_tag': consumer_tag})])
 
         return consumer_tag
 


Test patch:
diff --git a/tests/acceptance/async_adapter_tests.py b/tests/acceptance/async_adapter_tests.py
index 48be9a6..81a8c3c 100644
--- a/tests/acceptance/async_adapter_tests.py
+++ b/tests/acceptance/async_adapter_tests.py
@@ -340,7 +340,6 @@ class TestZ_PublishAndConsume(BoundQueueTestCase, AsyncAdapters):  # pylint: dis
         self.channel.basic_cancel(self.ctag, callback=self.on_cancelled)
 
 
-
 class TestZ_PublishAndConsumeBig(BoundQueueTestCase, AsyncAdapters):  # pylint: disable=C0103
     DESCRIPTION = "Publish a big message and consume it"
 
diff --git a/tests/acceptance/blocking_adapter_test.py b/tests/acceptance/blocking_adapter_test.py
index 6f7120b..7628f89 100644
--- a/tests/acceptance/blocking_adapter_test.py
+++ b/tests/acceptance/blocking_adapter_test.py
@@ -1768,14 +1768,12 @@ class TestBasicCancelPurgesPendingConsumerCancellationEvt(BlockingTestCaseBase):
 
         ch.publish('', routing_key=q_name, body='via-publish', mandatory=True)
 
-        # Create a consumer
+        # Create a consumer. Not passing a 'callback' to test client-generated
+        # consumer tags
         rx_messages = []
         consumer_tag = ch.basic_consume(
             q_name,
-            lambda *args: rx_messages.append(args),
-            no_ack=False,
-            exclusive=False,
-            arguments=None)
+            lambda *args: rx_messages.append(args))
 
         # Wait for the published message to arrive, but don't consume it
         while not ch._pending_events:
@@ -1848,7 +1846,8 @@ class TestBasicPublishWithoutPubacks(BlockingTestCaseBase):
                                                       queue=q_name,
                                                       expected_count=2)
 
-        # Create a consumer
+        # Create a consumer. Not passing a 'callback' to test client-generated
+        # consumer tags
         rx_messages = []
         consumer_tag = ch.basic_consume(
             q_name,
diff --git a/tests/unit/channel_tests.py b/tests/unit/channel_tests.py
index eba7b70..c1f8b67 100644
--- a/tests/unit/channel_tests.py
+++ b/tests/unit/channel_tests.py
@@ -1,6 +1,5 @@
 """
 Tests for pika.channel.Channel
-
 """
 import collections
 import logging
@@ -12,6 +11,9 @@ import mock
 
 from pika import channel, connection, exceptions, frame, spec
 
+# Disable protected-access, missing-docstring, and invalid-name,
+# too-many-public-methods, too-many-lines
+# pylint: disable=W0212,C0111,C0103,R0904,C0302
 
 class ConnectionTemplate(connection.Connection):
     """Template for using as mock spec_set for the pika Connection class. It
@@ -243,67 +245,88 @@ class ChannelTests(unittest.TestCase):
 
     def test_basic_consume_channel_closed(self):
         mock_callback = mock.Mock()
+        mock_on_msg_callback = mock.Mock()
         self.assertRaises(exceptions.ChannelClosed, self.obj.basic_consume,
-                          'test-queue', mock_callback)
+                          'test-queue', mock_on_msg_callback,
+                          callback=mock_callback)
 
     @mock.patch('pika.channel.Channel._validate_channel')
-    def test_basic_consume_calls_validate(self, validate):
+    @mock.patch('pika.channel.Channel._require_callback')
+    def test_basic_consume_calls_validate(self, require, validate):
         self.obj._set_state(self.obj.OPEN)
         mock_callback = mock.Mock()
-        self.obj.basic_consume('test-queue', mock_callback)
+        mock_on_msg_callback = mock.Mock()
+        self.obj.basic_consume('test-queue', mock_on_msg_callback,
+                               callback=mock_callback)
+        require.assert_called_once_with(mock_on_msg_callback)
         validate.assert_called_once()
 
-    def test_basic_consume_consumer_tag(self):
+    def test_basic_consume_consumer_tag_no_completion_callback(self):
+        self.obj._set_state(self.obj.OPEN)
+        expectation = 'ctag1.'
+        mock_on_msg_callback = mock.Mock()
+        consumer_tag = self.obj.basic_consume('test-queue',
+                                              mock_on_msg_callback)[:6]
+        self.assertEqual(consumer_tag, expectation)
+
+    def test_basic_consume_consumer_tag_with_completion_callback(self):
         self.obj._set_state(self.obj.OPEN)
         expectation = 'ctag1.'
         mock_callback = mock.Mock()
-        self.assertEqual(
-            self.obj.basic_consume('test-queue', mock_callback)[:6],
-            expectation)
+        mock_on_msg_callback = mock.Mock()
+        consumer_tag = self.obj.basic_consume('test-queue',
+                                              mock_on_msg_callback,
+                                              callback=mock_callback)[:6]
+        self.assertEqual(consumer_tag, expectation)
 
     def test_basic_consume_consumer_tag_cancelled_full(self):
         self.obj._set_state(self.obj.OPEN)
         expectation = 'ctag1.'
-        mock_callback = mock.Mock()
+        mock_on_msg_callback = mock.Mock()
         for ctag in ['ctag1.%i' % ii for ii in range(11)]:
             self.obj._cancelled.add(ctag)
         self.assertEqual(
-            self.obj.basic_consume('test-queue', mock_callback)[:6],
+            self.obj.basic_consume('test-queue', mock_on_msg_callback)[:6],
             expectation)
 
     def test_basic_consume_consumer_tag_in_consumers(self):
         self.obj._set_state(self.obj.OPEN)
         consumer_tag = 'ctag1.0'
+        mock_on_msg_callback = mock.Mock()
         mock_callback = mock.Mock()
         self.obj.basic_consume(
-            'test-queue', mock_callback, consumer_tag=consumer_tag)
+            'test-queue', mock_on_msg_callback,
+            consumer_tag=consumer_tag, callback=mock_callback)
         self.assertIn(consumer_tag, self.obj._consumers)
 
     def test_basic_consume_duplicate_consumer_tag_raises(self):
         self.obj._set_state(self.obj.OPEN)
         consumer_tag = 'ctag1.0'
+        mock_on_msg_callback = mock.Mock()
         mock_callback = mock.Mock()
         self.obj._consumers[consumer_tag] = logging.debug
         self.assertRaises(exceptions.DuplicateConsumerTag,
                           self.obj.basic_consume, 'test-queue',
-                          mock_callback, False, False, consumer_tag)
+                          mock_on_msg_callback, False, False,
+                          consumer_tag, None, mock_callback)
 
     def test_basic_consume_consumers_callback_value(self):
         self.obj._set_state(self.obj.OPEN)
         consumer_tag = 'ctag1.0'
-        mock_callback = mock.Mock()
+        mock_on_msg_callback = mock.Mock()
         self.obj.basic_consume(
-            'test-queue', mock_callback, consumer_tag=consumer_tag)
-        self.assertEqual(self.obj._consumers[consumer_tag], mock_callback)
+            'test-queue', mock_on_msg_callback, consumer_tag=consumer_tag)
+        self.assertEqual(self.obj._consumers[consumer_tag], mock_on_msg_callback)
 
     @mock.patch('pika.spec.Basic.Consume')
     @mock.patch('pika.channel.Channel._rpc')
-    def test_basic_consume_consumers_rpc_called(self, rpc, _unused):
+    def test_basic_consume_consumers_rpc_with_no_completion_callback(self, rpc, _unused):
         self.obj._set_state(self.obj.OPEN)
         consumer_tag = 'ctag1.0'
-        mock_callback = mock.Mock()
+        mock_on_msg_callback = mock.Mock()
         self.obj.basic_consume(
-            'test-queue', mock_callback, consumer_tag=consumer_tag)
+            'test-queue', mock_on_msg_callback,
+            consumer_tag=consumer_tag)
         expectation = spec.Basic.Consume(
             queue='test-queue',
             consumer_tag=consumer_tag,
@@ -314,6 +337,26 @@ class ChannelTests(unittest.TestCase):
                                         'consumer_tag': consumer_tag
                                     })])
 
+    @mock.patch('pika.spec.Basic.Consume')
+    @mock.patch('pika.channel.Channel._rpc')
+    def test_basic_consume_consumers_rpc_with_completion_callback(self, rpc, _unused):
+        self.obj._set_state(self.obj.OPEN)
+        consumer_tag = 'ctag1.0'
+        mock_on_msg_callback = mock.Mock()
+        mock_callback = mock.Mock()
+        self.obj.basic_consume(
+            'test-queue', mock_on_msg_callback,
+            consumer_tag=consumer_tag, callback=mock_callback)
+        expectation = spec.Basic.Consume(
+            queue='test-queue',
+            consumer_tag=consumer_tag,
+            no_ack=False,
+            exclusive=False)
+        rpc.assert_called_once_with(expectation, mock_callback,
+                                    [(spec.Basic.ConsumeOk, {
+                                        'consumer_tag': consumer_tag
+                                    })])
+
     @mock.patch('pika.channel.Channel._require_callback')
     def test_basic_get_calls_require_callback(self, require):
         self.obj._set_state(self.obj.OPEN)
@@ -514,12 +557,13 @@ class ChannelTests(unittest.TestCase):
 
     def test_confirm_delivery_with_bad_callback_raises_value_error(self):
         self.assertRaises(ValueError,
-                self.obj.confirm_delivery, 'bad-callback')
+                          self.obj.confirm_delivery,
+                          'bad-callback')
 
     def test_confirm_delivery_raises_channel_closed(self):
         cb = mock.Mock()
         self.assertRaises(exceptions.ChannelClosed,
-            self.obj.confirm_delivery, cb)
+                          self.obj.confirm_delivery, cb)
 
     def test_confirm_delivery_raises_method_not_implemented_for_confirms(self):
         self.obj._set_state(self.obj.OPEN)
@@ -653,7 +697,6 @@ class ChannelTests(unittest.TestCase):
     @mock.patch('pika.channel.Channel._rpc')
     def test_exchange_bind_rpc_request_nowait(self, rpc, _unused):
         self.obj._set_state(self.obj.OPEN)
-        mock_callback = mock.Mock()
         self.obj.exchange_bind('foo', 'bar', 'baz', callback=None)
         rpc.assert_called_once_with(
             spec.Exchange.Bind(0, 'foo', 'bar', 'baz'), None, [])
@@ -683,7 +726,6 @@ class ChannelTests(unittest.TestCase):
     @mock.patch('pika.channel.Channel._rpc')
     def test_exchange_declare_rpc_request_nowait(self, rpc, _unused):
         self.obj._set_state(self.obj.OPEN)
-        mock_callback = mock.Mock()
         self.obj.exchange_declare('foo', callback=None)
         rpc.assert_called_once_with(
             spec.Exchange.Declare(0, 'foo'), None, [])
@@ -711,7 +753,6 @@ class ChannelTests(unittest.TestCase):
     @mock.patch('pika.channel.Channel._rpc')
     def test_exchange_delete_rpc_request_nowait(self, rpc, _unused):
         self.obj._set_state(self.obj.OPEN)
-        mock_callback = mock.Mock()
         self.obj.exchange_delete('foo', callback=None)
         rpc.assert_called_once_with(
             spec.Exchange.Delete(0, 'foo'), None, [])
@@ -824,7 +865,6 @@ class ChannelTests(unittest.TestCase):
     @mock.patch('pika.channel.Channel._rpc')
     def test_queue_bind_rpc_request_nowait(self, rpc, _unused):
         self.obj._set_state(self.obj.OPEN)
-        mock_callback = mock.Mock()
         self.obj.queue_bind('foo', 'bar', 'baz', callback=None)
         rpc.assert_called_once_with(
             spec.Queue.Bind(0, 'foo', 'bar', 'baz'), None, [])
@@ -857,7 +897,6 @@ class ChannelTests(unittest.TestCase):
     @mock.patch('pika.channel.Channel._rpc')
     def test_queue_declare_rpc_request_nowait(self, rpc, _unused):
         self.obj._set_state(self.obj.OPEN)
-        mock_callback = mock.Mock()
         self.obj.queue_declare('foo', callback=None)
         rpc.assert_called_once_with(
             spec.Queue.Declare(0, 'foo'), None, [])
@@ -884,7 +923,6 @@ class ChannelTests(unittest.TestCase):
     @mock.patch('pika.channel.Channel._rpc')
     def test_queue_delete_rpc_request_nowait(self, rpc, _unused):
         self.obj._set_state(self.obj.OPEN)
-        mock_callback = mock.Mock()
         self.obj.queue_delete('foo', callback=None)
         rpc.assert_called_once_with(
             spec.Queue.Delete(0, 'foo'), None, [])
@@ -911,7 +949,6 @@ class ChannelTests(unittest.TestCase):
     @mock.patch('pika.channel.Channel._rpc')
     def test_queue_purge_rpc_request_nowait(self, rpc, _unused):
         self.obj._set_state(self.obj.OPEN)
-        mock_callback = mock.Mock()
         self.obj.queue_purge('foo', callback=None)
         rpc.assert_called_once_with(
             spec.Queue.Purge(0, 'foo'), None, [])
@@ -1461,4 +1498,5 @@ class ChannelTests(unittest.TestCase):
             self):
         self.obj._set_state(self.obj.OPEN)
         self.assertRaises(TypeError,
-                self.obj._validate_rpc_completion_callback, 'foo')
+                          self.obj._validate_rpc_completion_callback,
+                          'foo')