Repository: pika/pika

Issue:
BlockingChannel.basic_qos hanging on non-integer parameters
Under Python 3.6.4 and Pika 0.11.2, the `BlockingChannel.basic_qos` method hangs when setting its `prefetch_size` or `prefetch_count` parameter to a non-integer value.

Input:

```
$ python3 <<EOF
import pika
with pika.BlockingConnection() as connection:
    channel = connection.channel()
    channel.basic_qos(prefetch_size="hello")
EOF
```

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

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

```diff
@@            Coverage Diff             @@
##           master     #948      +/-   ##
==========================================
+ Coverage   82.48%   82.51%   +0.02%     
==========================================
  Files          20       20              
  Lines        3665     3671       +6     
  Branches      544      545       +1     
==========================================
+ Hits         3023     3029       +6     
  Misses        497      497              
  Partials      145      145
```


| [Impacted Files](https://codecov.io/gh/pika/pika/pull/948?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [pika/channel.py](https://codecov.io/gh/pika/pika/pull/948/diff?src=pr&el=tree#diff-cGlrYS9jaGFubmVsLnB5) | `94.11% <100%> (+0.09%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/pika/pika/pull/948?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/948?src=pr&el=footer). Last update [1b76018...c0d0b3a](https://codecov.io/gh/pika/pika/pull/948?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

lukebakken: @vitaly-krugl - this should be all set. I will address #951 next.

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..5299d15 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
 
@@ -442,16 +449,16 @@ class Channel(object):
                                    falls into other prefetch limits). May be set
                                    to zero, meaning "no specific limit",
                                    although other prefetch limits may still
-                                   apply. The prefetch-size is ignored if the
-                                   no-ack option is set.
+                                   apply. The prefetch-size is ignored by
+                                   consumers who have enabled the no-ack option.
         :param int prefetch_count: Specifies a prefetch window in terms of whole
                                    messages. This field may be used in
                                    combination with the prefetch-size field; a
                                    message will only be sent in advance if both
                                    prefetch windows (and those at the channel
                                    and connection level) allow it. The
-                                   prefetch-count is ignored if the no-ack
-                                   option is set.
+                                   prefetch-count is ignored by consumers who
+                                   have enabled the no-ack option.
         :param bool all_channels: Should the QoS apply to all channels
         :param callable callback: The callback to call for Basic.QosOk response
         :raises ValueError:
@@ -459,6 +466,8 @@ class Channel(object):
         """
         self._validate_channel()
         self._validate_rpc_completion_callback(callback)
+        self._validate_zero_or_greater('prefetch_size', prefetch_size)
+        self._validate_zero_or_greater('prefetch_count', prefetch_count)
         return self._rpc(spec.Basic.Qos(prefetch_size, prefetch_count,
                                         all_channels),
                          callback, [spec.Basic.QosOk])
@@ -1411,6 +1420,11 @@ class Channel(object):
             raise TypeError(
                 'Completion callback must be callable if not None')
 
+    def _validate_zero_or_greater(self, name, value):
+        if int(value) < 0:
+            errmsg = '{} must be >= 0, but got {}'.format(name, value)
+            raise ValueError(errmsg)
+
 
 class ContentFrameAssembler(object):
     """Handle content related frames, building a message and return the message


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..67e0325 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)
@@ -392,6 +435,26 @@ class ChannelTests(unittest.TestCase):
         self.assertRaises(exceptions.ChannelClosed, self.obj.basic_qos, 0,
                           False, True)
 
+    def test_basic_qos_invalid_prefetch_size_raises_error(self):
+        self.obj._set_state(self.obj.OPEN)
+        with self.assertRaises(ValueError) as ex:
+            self.obj.basic_qos('foo', 123)
+        self.assertEqual("invalid literal for int() with base 10: 'foo'",
+                         ex.exception.args[0])
+        with self.assertRaises(ValueError) as ex:
+            self.obj.basic_qos(-1, 123)
+        self.assertIn('prefetch_size', ex.exception.args[0])
+
+    def test_basic_qos_invalid_prefetch_count_raises_error(self):
+        self.obj._set_state(self.obj.OPEN)
+        with self.assertRaises(ValueError) as ex:
+            self.obj.basic_qos(123, 'foo')
+        self.assertEqual("invalid literal for int() with base 10: 'foo'",
+                         ex.exception.args[0])
+        with self.assertRaises(ValueError) as ex:
+            self.obj.basic_qos(123, -1)
+        self.assertIn('prefetch_count', ex.exception.args[0])
+
     @mock.patch('pika.spec.Basic.Qos')
     @mock.patch('pika.channel.Channel._rpc')
     def test_basic_qos_rpc_request(self, rpc, _unused):
@@ -514,12 +577,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 +717,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 +746,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 +773,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 +885,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 +917,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 +943,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 +969,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 +1518,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')