Repository: pika/pika

Issue:
Regression: Cannot receive long running messages when in use in tornado app
Hi there,

we just noticed a regression from pika 0.9.14 to 0.10.0. We haven't been able to track down the problem further than this (sorry) but will instead stop for now by pinning the old version.

The problem seems to be that pika (used via tornado-celery) is not able to receive answer messages from rabbitmq for longish running (>4-5 seconds) requests to a background worker.

The answer message instead seems to time out after the message timeout period in rabbitmq after which it is lost.

Do you need more information to fix this?

Discussion:
vitaly-krugl: CC @gmr: Hi Gavin, please review this fix.
vitaly-krugl: @gmr, the failed test `BlockingChannel.basic_nack single message` is unrelated to the changes in this PR; it's a known issue described in #677. 

Please re-start the build. Thanks!

Patch:
diff --git a/pika/adapters/libev_connection.py b/pika/adapters/libev_connection.py
index 2a27055..26a6547 100644
--- a/pika/adapters/libev_connection.py
+++ b/pika/adapters/libev_connection.py
@@ -127,7 +127,7 @@ class LibevConnection(BaseConnection):
             if self._on_signal_callback and not global_sigterm_watcher:
                 global_sigterm_watcher = \
                     self.ioloop.signal(signal.SIGTERM,
-                                                                                  self._handle_sigterm)
+                                       self._handle_sigterm)
 
             if self._on_signal_callback and not global_sigint_watcher:
                 global_sigint_watcher = self.ioloop.signal(signal.SIGINT,
@@ -136,8 +136,8 @@ class LibevConnection(BaseConnection):
             if not self._io_watcher:
                 self._io_watcher = \
                     self.ioloop.io(self.socket.fileno(),
-                                                                        self._PIKA_TO_LIBEV_ARRAY[self.event_state],
-                                                                        self._handle_events)
+                                   self._PIKA_TO_LIBEV_ARRAY[self.event_state],
+                                   self._handle_events)
 
             self.async = pyev.Async(self.ioloop, self._noop_callable)
             self.async.start()
@@ -209,8 +209,9 @@ class LibevConnection(BaseConnection):
                     self._PIKA_TO_LIBEV_ARRAY[self.event_state])
 
                 break
-            except:  # sometimes the stop() doesn't complete in time
-                if retries > 5: raise
+            except Exception:  # sometimes the stop() doesn't complete in time
+                if retries > 5:
+                    raise
                 self._io_watcher.stop()  # so try it again
                 retries += 1
 
@@ -268,7 +269,7 @@ class LibevConnection(BaseConnection):
         :rtype: timer instance handle.
 
         """
-        LOGGER.debug('deadline: {0}'.format(deadline))
+        LOGGER.debug('deadline: %s', deadline)
         timer = self._get_timer(deadline)
         self._active_timers[timer] = (callback_method, callback_timeout,
                                       callback_kwargs)
diff --git a/pika/channel.py b/pika/channel.py
index 641e469..5c67c49 100644
--- a/pika/channel.py
+++ b/pika/channel.py
@@ -436,7 +436,10 @@ class Channel(object):
         For more information see:
             http://www.rabbitmq.com/extensions.html#confirms
 
-        :param method callback: The callback for delivery confirmations
+        :param method callback: The callback for delivery confirmations that has
+          the following signature: callback(pika.frame.Method), where
+          method_frame contains either method `spec.Basic.Ack` or
+          `spec.Basic.Nack`
         :param bool nowait: Do not send a reply frame (Confirm.SelectOk)
 
         """
@@ -674,7 +677,8 @@ class Channel(object):
 
         Leave the queue name empty for a auto-named queue in RabbitMQ
 
-        :param method callback: The method to call on Queue.DeclareOk
+        :param method callback: callback(pika.frame.Method) for method
+          Queue.DeclareOk
         :param queue: The queue name
         :type queue: str or unicode
         :param bool passive: Only check to see if the queue exists
@@ -694,7 +698,8 @@ class Channel(object):
         self._validate_channel_and_callback(callback)
         return self._rpc(spec.Queue.Declare(0, queue, passive, durable,
                                             exclusive, auto_delete, nowait,
-                                            arguments or dict()), callback,
+                                            arguments or dict()),
+                         callback,
                          replies)
 
     def queue_delete(self,
@@ -1087,52 +1092,74 @@ class Channel(object):
         """
         LOGGER.debug('%i blocked frames', len(self._blocked))
         self._blocking = None
-        while len(self._blocked) > 0 and self._blocking is None:
+        while self._blocked and self._blocking is None:
             self._rpc(*self._blocked.popleft())
 
     def _rpc(self, method_frame, callback=None, acceptable_replies=None):
-        """Shortcut wrapper to the Connection's rpc command using its callback
-        stack, passing in our channel number.
+        """Make a syncronous channel RPC call for a synchronous method frame. If
+        the channel is already in the blocking state, then enqueue the request,
+        but don't send it at this time; it will be eventually sent by
+        `_on_synchronous_complete` after the prior blocking request receives a
+        resposne. If the channel is not in the blocking state and
+        `acceptable_replies` is not empty, transition the channel to the
+        blocking state and register for `_on_synchronous_complete` before
+        sending the request.
+
+        NOTE: A populated callback must be accompanied by populated
+        acceptable_replies.
 
         :param pika.amqp_object.Method method_frame: The method frame to call
         :param method callback: The callback for the RPC response
         :param list acceptable_replies: The replies this RPC call expects
 
         """
+        assert method_frame.synchronous, (
+            'Only synchronous-capable frames may be used with _rpc: %r'
+            % (method_frame,))
+
+        # Validate we got None or a list of acceptable_replies
+        if not isinstance(acceptable_replies, (type(None), list)):
+            raise TypeError('acceptable_replies should be list or None')
+
+        # Validate the callback is callable
+        if callback is not None and not is_callable(callback):
+            raise TypeError('callback should be None, a function or method.')
+
+        if callback is not None and not acceptable_replies:
+            raise ValueError('A populated callback must be accompanied by '
+                             'populated acceptable_replies')
+
         # Make sure the channel is open
         if self.is_closed:
             raise exceptions.ChannelClosed
 
         # If the channel is blocking, add subsequent commands to our stack
         if self._blocking:
+            LOGGER.debug('Already in blocking state, so enqueueing frame %s; '
+                         'acceptable_replies=%r',
+                         method_frame, acceptable_replies)
             return self._blocked.append([method_frame, callback,
                                          acceptable_replies])
 
-        # Validate we got None or a list of acceptable_replies
-        if acceptable_replies and not isinstance(acceptable_replies, list):
-            raise TypeError("acceptable_replies should be list or None")
-
-        # Validate the callback is callable
-        if callback and not is_callable(callback):
-            raise TypeError("callback should be None, a function or method.")
-
-        # Block until a response frame is received for synchronous frames
-        if method_frame.synchronous:
-            self._blocking = method_frame.NAME
-
         # If acceptable replies are set, add callbacks
         if acceptable_replies:
-            for reply in acceptable_replies or list():
+            # Block until a response frame is received for synchronous frames
+            self._blocking = method_frame.NAME
+            LOGGER.debug(
+                'Entering blocking state on frame %s; acceptable_replies=%r',
+                method_frame, acceptable_replies)
+
+            for reply in acceptable_replies:
                 if isinstance(reply, tuple):
                     reply, arguments = reply
                 else:
                     arguments = None
-                LOGGER.debug('Adding in on_synchronous_complete callback')
+                LOGGER.debug('Adding on_synchronous_complete callback')
                 self.callbacks.add(self.channel_number, reply,
                                    self._on_synchronous_complete,
                                    arguments=arguments)
-                if callback:
-                    LOGGER.debug('Adding passed in callback')
+                if callback is not None:
+                    LOGGER.debug('Adding passed-in callback')
                     self.callbacks.add(self.channel_number, reply, callback,
                                        arguments=arguments)
 


Test patch:
diff --git a/tests/acceptance/async_adapter_tests.py b/tests/acceptance/async_adapter_tests.py
index 9d7448e..9a44208 100644
--- a/tests/acceptance/async_adapter_tests.py
+++ b/tests/acceptance/async_adapter_tests.py
@@ -1,13 +1,25 @@
+# Suppress pylint messages concerning missing class and method docstrings
+# pylint: disable=C0111
+
+# Suppress pylint warning about attribute defined outside __init__
+# pylint: disable=W0201
+
+# Suppress pylint warning about access to protected member
+# pylint: disable=W0212
+
+# Suppress pylint warning about unused argument
+# pylint: disable=W0613
+
 import time
 import uuid
 
-from pika import spec, URLParameters
+from pika import spec
 from pika.compat import as_bytes
 
 from async_test_base import (AsyncTestCase, BoundQueueTestCase, AsyncAdapters)
 
 
-class TestA_Connect(AsyncTestCase, AsyncAdapters):
+class TestA_Connect(AsyncTestCase, AsyncAdapters):  # pylint: disable=C0103
     DESCRIPTION = "Connect, open channel and disconnect"
 
     def begin(self, channel):
@@ -26,11 +38,49 @@ class TestConfirmSelect(AsyncTestCase, AsyncAdapters):
         self.stop()
 
 
+class TestBlockingNonBlockingBlockingRPCWontStall(AsyncTestCase, AsyncAdapters):
+    DESCRIPTION = ("Verify that a sequence of blocking, non-blocking, blocking "
+                   "RPC requests won't stall")
+
+    def begin(self, channel):
+        # Queue declaration params table: queue name, nowait value
+        self._expected_queue_params = (
+            ("blocking-non-blocking-stall-check-" + uuid.uuid1().hex, False),
+            ("blocking-non-blocking-stall-check-" + uuid.uuid1().hex, True),
+            ("blocking-non-blocking-stall-check-" + uuid.uuid1().hex, False)
+        )
+
+        self._declared_queue_names = []
+
+        for queue, nowait in self._expected_queue_params:
+            channel.queue_declare(callback=self._queue_declare_ok_cb
+                                  if not nowait else None,
+                                  queue=queue,
+                                  auto_delete=True,
+                                  nowait=nowait,
+                                  arguments={'x-expires': self.TIMEOUT * 1000})
+
+    def _queue_declare_ok_cb(self, declare_ok_frame):
+        self._declared_queue_names.append(declare_ok_frame.method.queue)
+
+        if len(self._declared_queue_names) == 2:
+            # Initiate check for creation of queue declared with nowait=True
+            self.channel.queue_declare(callback=self._queue_declare_ok_cb,
+                                       queue=self._expected_queue_params[1][0],
+                                       passive=True,
+                                       nowait=False)
+        elif len(self._declared_queue_names) == 3:
+            self.assertSequenceEqual(
+                sorted(self._declared_queue_names),
+                sorted(item[0] for item in self._expected_queue_params))
+            self.stop()
+
+
 class TestConsumeCancel(AsyncTestCase, AsyncAdapters):
     DESCRIPTION = "Consume and cancel"
 
     def begin(self, channel):
-        self.queue_name = str(uuid.uuid4())
+        self.queue_name = self.__class__.__name__ + ':' + uuid.uuid1().hex
         channel.queue_declare(self.on_queue_declared, queue=self.queue_name)
 
     def on_queue_declared(self, frame):
@@ -58,7 +108,7 @@ class TestExchangeDeclareAndDelete(AsyncTestCase, AsyncAdapters):
     X_TYPE = 'direct'
 
     def begin(self, channel):
-        self.name = self.__class__.__name__ + ':' + str(id(self))
+        self.name = self.__class__.__name__ + ':' + uuid.uuid1().hex
         channel.exchange_declare(self.on_exchange_declared, self.name,
                                  exchange_type=self.X_TYPE,
                                  passive=False,
@@ -81,7 +131,7 @@ class TestExchangeRedeclareWithDifferentValues(AsyncTestCase, AsyncAdapters):
     X_TYPE2 = 'topic'
 
     def begin(self, channel):
-        self.name = self.__class__.__name__ + ':' + str(id(self))
+        self.name = self.__class__.__name__ + ':' + uuid.uuid1().hex
         self.channel.add_on_close_callback(self.on_channel_closed)
         channel.exchange_declare(self.on_exchange_declared, self.name,
                                  exchange_type=self.X_TYPE1,
@@ -97,7 +147,7 @@ class TestExchangeRedeclareWithDifferentValues(AsyncTestCase, AsyncAdapters):
         self.connection.channel(self.on_cleanup_channel)
 
     def on_exchange_declared(self, frame):
-        self.channel.exchange_declare(self.on_exchange_declared, self.name,
+        self.channel.exchange_declare(self.on_bad_result, self.name,
                                       exchange_type=self.X_TYPE2,
                                       passive=False,
                                       durable=False,
@@ -134,7 +184,8 @@ class TestQueueNameDeclareAndDelete(AsyncTestCase, AsyncAdapters):
     DESCRIPTION = "Create and delete a named queue"
 
     def begin(self, channel):
-        channel.queue_declare(self.on_queue_declared, str(id(self)),
+        self._q_name = self.__class__.__name__ + ':' + uuid.uuid1().hex
+        channel.queue_declare(self.on_queue_declared, self._q_name,
                               passive=False,
                               durable=False,
                               exclusive=True,
@@ -143,10 +194,9 @@ class TestQueueNameDeclareAndDelete(AsyncTestCase, AsyncAdapters):
                               arguments={'x-expires': self.TIMEOUT * 1000})
 
     def on_queue_declared(self, frame):
-        queue = str(id(self))
         self.assertIsInstance(frame.method, spec.Queue.DeclareOk)
         # Frame's method's queue is encoded (impl detail)
-        self.assertEqual(frame.method.queue, queue)
+        self.assertEqual(frame.method.queue, self._q_name)
         self.channel.queue_delete(self.on_queue_delete, frame.method.queue)
 
     def on_queue_delete(self, frame):
@@ -159,8 +209,9 @@ class TestQueueRedeclareWithDifferentValues(AsyncTestCase, AsyncAdapters):
     DESCRIPTION = "Should close chan: re-declared queue w/ diff params"
 
     def begin(self, channel):
+        self._q_name = self.__class__.__name__ + ':' + uuid.uuid1().hex
         self.channel.add_on_close_callback(self.on_channel_closed)
-        channel.queue_declare(self.on_queue_declared, str(id(self)),
+        channel.queue_declare(self.on_queue_declared, self._q_name,
                               passive=False,
                               durable=False,
                               exclusive=True,
@@ -172,7 +223,7 @@ class TestQueueRedeclareWithDifferentValues(AsyncTestCase, AsyncAdapters):
         self.stop()
 
     def on_queue_declared(self, frame):
-        self.channel.queue_declare(self.on_bad_result, str(id(self)),
+        self.channel.queue_declare(self.on_bad_result, self._q_name,
                                    passive=False,
                                    durable=True,
                                    exclusive=False,
@@ -181,13 +232,13 @@ class TestQueueRedeclareWithDifferentValues(AsyncTestCase, AsyncAdapters):
                                    arguments={'x-expires': self.TIMEOUT * 1000})
 
     def on_bad_result(self, frame):
-        self.channel.queue_delete(None, str(id(self)), nowait=True)
+        self.channel.queue_delete(None, self._q_name, nowait=True)
         raise AssertionError("Should not have received a Queue.DeclareOk")
 
 
 
-class TestTX1_Select(AsyncTestCase, AsyncAdapters):
-    DESCRIPTION="Receive confirmation of Tx.Select"
+class TestTX1_Select(AsyncTestCase, AsyncAdapters):  # pylint: disable=C0103
+    DESCRIPTION = "Receive confirmation of Tx.Select"
 
     def begin(self, channel):
         channel.tx_select(self.on_complete)
@@ -198,8 +249,8 @@ class TestTX1_Select(AsyncTestCase, AsyncAdapters):
 
 
 
-class TestTX2_Commit(AsyncTestCase, AsyncAdapters):
-    DESCRIPTION="Start a transaction, and commit it"
+class TestTX2_Commit(AsyncTestCase, AsyncAdapters):  # pylint: disable=C0103
+    DESCRIPTION = "Start a transaction, and commit it"
 
     def begin(self, channel):
         channel.tx_select(self.on_selectok)
@@ -213,7 +264,7 @@ class TestTX2_Commit(AsyncTestCase, AsyncAdapters):
         self.stop()
 
 
-class TestTX2_CommitFailure(AsyncTestCase, AsyncAdapters):
+class TestTX2_CommitFailure(AsyncTestCase, AsyncAdapters):  # pylint: disable=C0103
     DESCRIPTION = "Close the channel: commit without a TX"
 
     def begin(self, channel):
@@ -226,11 +277,12 @@ class TestTX2_CommitFailure(AsyncTestCase, AsyncAdapters):
     def on_selectok(self, frame):
         self.assertIsInstance(frame.method, spec.Tx.SelectOk)
 
-    def on_commitok(self, frame):
+    @staticmethod
+    def on_commitok(frame):
         raise AssertionError("Should not have received a Tx.CommitOk")
 
 
-class TestTX3_Rollback(AsyncTestCase, AsyncAdapters):
+class TestTX3_Rollback(AsyncTestCase, AsyncAdapters):  # pylint: disable=C0103
     DESCRIPTION = "Start a transaction, then rollback"
 
     def begin(self, channel):
@@ -246,7 +298,7 @@ class TestTX3_Rollback(AsyncTestCase, AsyncAdapters):
 
 
 
-class TestTX3_RollbackFailure(AsyncTestCase, AsyncAdapters):
+class TestTX3_RollbackFailure(AsyncTestCase, AsyncAdapters):  # pylint: disable=C0103
     DESCRIPTION = "Close the channel: rollback without a TX"
 
     def begin(self, channel):
@@ -256,12 +308,12 @@ class TestTX3_RollbackFailure(AsyncTestCase, AsyncAdapters):
     def on_channel_closed(self, channel, reply_code, reply_text):
         self.stop()
 
-    def on_commitok(self, frame):
+    @staticmethod
+    def on_commitok(frame):
         raise AssertionError("Should not have received a Tx.RollbackOk")
 
 
-
-class TestZ_PublishAndConsume(BoundQueueTestCase, AsyncAdapters):
+class TestZ_PublishAndConsume(BoundQueueTestCase, AsyncAdapters):  # pylint: disable=C0103
     DESCRIPTION = "Publish a message and consume it"
 
     def on_ready(self, frame):
@@ -282,10 +334,11 @@ class TestZ_PublishAndConsume(BoundQueueTestCase, AsyncAdapters):
 
 
 
-class TestZ_PublishAndConsumeBig(BoundQueueTestCase, AsyncAdapters):
+class TestZ_PublishAndConsumeBig(BoundQueueTestCase, AsyncAdapters):  # pylint: disable=C0103
     DESCRIPTION = "Publish a big message and consume it"
 
-    def _get_msg_body(self):
+    @staticmethod
+    def _get_msg_body():
         return '\n'.join(["%s" % i for i in range(0, 2097152)])
 
     def on_ready(self, frame):
@@ -305,7 +358,7 @@ class TestZ_PublishAndConsumeBig(BoundQueueTestCase, AsyncAdapters):
         self.channel.basic_cancel(self.on_cancelled, self.ctag)
 
 
-class TestZ_PublishAndGet(BoundQueueTestCase, AsyncAdapters):
+class TestZ_PublishAndGet(BoundQueueTestCase, AsyncAdapters):  # pylint: disable=C0103
     DESCRIPTION = "Publish a message and get it"
 
     def on_ready(self, frame):
@@ -321,13 +374,14 @@ class TestZ_PublishAndGet(BoundQueueTestCase, AsyncAdapters):
         self.stop()
 
 
-class TestZ_AccessDenied(AsyncTestCase, AsyncAdapters):
+class TestZ_AccessDenied(AsyncTestCase, AsyncAdapters):  # pylint: disable=C0103
     DESCRIPTION = "Verify that access denied invokes on open error callback"
 
     def start(self, *args, **kwargs):
         self.parameters.virtual_host = str(uuid.uuid4())
         self.error_captured = False
         super(TestZ_AccessDenied, self).start(*args, **kwargs)
+        self.assertTrue(self.error_captured)
 
     def on_open_error(self, connection, error):
         self.error_captured = True
@@ -336,7 +390,3 @@ class TestZ_AccessDenied(AsyncTestCase, AsyncAdapters):
     def on_open(self, connection):
         super(TestZ_AccessDenied, self).on_open(connection)
         self.stop()
-
-    def tearDown(self):
-        self.assertTrue(self.error_captured)
-        super(TestZ_AccessDenied, self).tearDown()
diff --git a/tests/acceptance/async_test_base.py b/tests/acceptance/async_test_base.py
index 26c3e1a..eafe72f 100644
--- a/tests/acceptance/async_test_base.py
+++ b/tests/acceptance/async_test_base.py
@@ -1,3 +1,10 @@
+# Suppress pylint warnings concerning attribute defined outside __init__
+# pylint: disable=W0201
+
+# Suppress pylint messages concerning missing docstrings
+# pylint: disable=C0111
+
+from datetime import datetime
 import select
 import logging
 try:
@@ -6,7 +13,9 @@ except ImportError:
     import unittest
 
 import platform
-target = platform.python_implementation()
+_TARGET = platform.python_implementation()
+
+import uuid
 
 import pika
 from pika import adapters
@@ -24,6 +33,9 @@ class AsyncTestCase(unittest.TestCase):
             'amqp://guest:guest@localhost:5672/%2F')
         super(AsyncTestCase, self).setUp()
 
+    def tearDown(self):
+        self._stop()
+
     def shortDescription(self):
         method_desc = super(AsyncTestCase, self).shortDescription()
         if self.DESCRIPTION:
@@ -31,11 +43,12 @@ class AsyncTestCase(unittest.TestCase):
         else:
             return method_desc
 
-    def begin(self, channel):
+    def begin(self, channel):  # pylint: disable=R0201,W0613
         """Extend to start the actual tests on the channel"""
-        raise AssertionError("AsyncTestCase.begin_test not extended")
+        self.fail("AsyncTestCase.begin_test not extended")
 
     def start(self, adapter=None):
+        self.logger.info('start at %s', datetime.utcnow())
         self.adapter = adapter or self.ADAPTER
 
         self.connection = self.adapter(self.parameters, self.on_open,
@@ -53,19 +66,18 @@ class AsyncTestCase(unittest.TestCase):
 
     def _stop(self):
         if hasattr(self, 'timeout') and self.timeout:
+            self.logger.info("Removing timeout")
             self.connection.remove_timeout(self.timeout)
             self.timeout = None
         if hasattr(self, 'connection') and self.connection:
+            self.logger.info("Stopping ioloop")
             self.connection.ioloop.stop()
             self.connection = None
 
-    def tearDown(self):
-        self._stop()
-
     def on_closed(self, connection, reply_code, reply_text):
         """called when the connection has finished closing"""
-        self.logger.debug('on_closed: %r %r %r', connection,
-                          reply_code, reply_text)
+        self.logger.info('on_closed: %r %r %r', connection,
+                         reply_code, reply_text)
         self._stop()
 
     def on_open(self, connection):
@@ -73,29 +85,25 @@ class AsyncTestCase(unittest.TestCase):
         self.channel = connection.channel(self.begin)
 
     def on_open_error(self, connection, error):
-        self.logger.debug('on_open_error: %r %r', connection, error)
+        self.logger.error('on_open_error: %r %r', connection, error)
         connection.ioloop.stop()
         raise AssertionError('Error connecting to RabbitMQ')
 
     def on_timeout(self):
         """called when stuck waiting for connection to close"""
+        self.logger.info('on_timeout at %s', datetime.utcnow())
         # force the ioloop to stop
-        self.logger.debug('on_timeout')
+        self.logger.debug('on_timeout called')
         self.connection.ioloop.stop()
         raise AssertionError('Test timed out')
 
 
 class BoundQueueTestCase(AsyncTestCase):
 
-    def tearDown(self):
-        """Cleanup auto-declared queue and exchange"""
-        self._cconn = self.adapter(self.parameters, self._on_cconn_open,
-                                   self._on_cconn_error, self._on_cconn_closed)
-
     def start(self, adapter=None):
         # PY3 compat encoding
-        self.exchange = 'e' + str(id(self))
-        self.queue = 'q' + str(id(self))
+        self.exchange = 'e-' + self.__class__.__name__ + ':' + uuid.uuid1().hex
+        self.queue = 'q-' + self.__class__.__name__ + ':' + uuid.uuid1().hex
         self.routing_key = self.__class__.__name__
         super(BoundQueueTestCase, self).start(adapter)
 
@@ -106,82 +114,70 @@ class BoundQueueTestCase(AsyncTestCase):
                                       durable=False,
                                       auto_delete=True)
 
-    def on_exchange_declared(self, frame):
+    def on_exchange_declared(self, frame):  # pylint: disable=W0613
         self.channel.queue_declare(self.on_queue_declared, self.queue,
                                    passive=False,
                                    durable=False,
                                    exclusive=True,
                                    auto_delete=True,
                                    nowait=False,
-                                   arguments={'x-expires': self.TIMEOUT * 1000}
-                                   )
+                                   arguments={'x-expires': self.TIMEOUT * 1000})
 
-    def on_queue_declared(self, frame):
+    def on_queue_declared(self, frame):  # pylint: disable=W0613
         self.channel.queue_bind(self.on_ready, self.queue, self.exchange,
                                 self.routing_key)
 
     def on_ready(self, frame):
         raise NotImplementedError
 
-    def _on_cconn_closed(self, cconn, *args, **kwargs):
-        cconn.ioloop.stop()
-        self._cconn = None
-
-    def _on_cconn_error(self, connection):
-        connection.ioloop.stop()
-        raise AssertionError('Error connecting to RabbitMQ')
-
-    def _on_cconn_open(self, connection):
-        connection.channel(self._on_cconn_channel)
-
-    def _on_cconn_channel(self, channel):
-        channel.exchange_delete(None, self.exchange, nowait=True)
-        channel.queue_delete(None, self.queue, nowait=True)
-        self._cconn.close()
 
 #
 # In order to write test cases that will tested using all the Async Adapters
-# write a class that inherits both from one of TestCase classes above and 
+# write a class that inherits both from one of TestCase classes above and
 # from the AsyncAdapters class below. This allows you to avoid duplicating the
 # test methods for each adapter in each test class.
 #
 
 class AsyncAdapters(object):
 
+    def start(self, adapter_class):
+        raise NotImplementedError
+
     def select_default_test(self):
         "SelectConnection:DefaultPoller"
-        select_connection.POLLER_TYPE=None
+        select_connection.POLLER_TYPE = None
         self.start(adapters.SelectConnection)
 
     def select_select_test(self):
         "SelectConnection:select"
-        select_connection.POLLER_TYPE='select'
+        select_connection.POLLER_TYPE = 'select'
         self.start(adapters.SelectConnection)
 
-    @unittest.skipIf(not hasattr(select, 'poll') 
-        or not hasattr(select.poll(), 'modify'), "poll not supported")
+    @unittest.skipIf(
+        not hasattr(select, 'poll') or
+        not hasattr(select.poll(), 'modify'), "poll not supported")  # pylint: disable=E1101
     def select_poll_test(self):
         "SelectConnection:poll"
-        select_connection.POLLER_TYPE='poll'
+        select_connection.POLLER_TYPE = 'poll'
         self.start(adapters.SelectConnection)
 
     @unittest.skipIf(not hasattr(select, 'epoll'), "epoll not supported")
     def select_epoll_test(self):
         "SelectConnection:epoll"
-        select_connection.POLLER_TYPE='epoll'
+        select_connection.POLLER_TYPE = 'epoll'
         self.start(adapters.SelectConnection)
 
     @unittest.skipIf(not hasattr(select, 'kqueue'), "kqueue not supported")
     def select_kqueue_test(self):
         "SelectConnection:kqueue"
-        select_connection.POLLER_TYPE='kqueue'
+        select_connection.POLLER_TYPE = 'kqueue'
         self.start(adapters.SelectConnection)
 
     def tornado_test(self):
         "TornadoConnection"
         self.start(adapters.TornadoConnection)
 
-    @unittest.skipIf(target == 'PyPy', 'PyPy is not supported')
+    @unittest.skipIf(_TARGET == 'PyPy', 'PyPy is not supported')
     @unittest.skipIf(adapters.LibevConnection is None, 'pyev is not installed')
     def libev_test(self):
         "LibevConnection"
diff --git a/tests/unit/channel_tests.py b/tests/unit/channel_tests.py
index a80996a..57c5178 100644
--- a/tests/unit/channel_tests.py
+++ b/tests/unit/channel_tests.py
@@ -2,6 +2,10 @@
 Tests for pika.channel.ContentFrameDispatcher
 
 """
+
+# Disable pylint warning about Access to a protected member
+# pylint: disable=W0212
+
 import collections
 import logging
 
@@ -16,17 +20,35 @@ except ImportError:
     import unittest
 import warnings
 
+from pika import callback
 from pika import channel
+from pika import connection
 from pika import exceptions
 from pika import frame
 from pika import spec
 
 
+class ConnectionTemplate(connection.Connection):
+    """Template for using as mock spec_set for the pika Connection class. It
+    defines members accessed by the code under test that would be defined in the
+    base class's constructor.
+    """
+    callbacks = None
+
+    # Suppress pylint warnings about specific abstract methods not being
+    # overridden
+    _adapter_connect = connection.Connection._adapter_connect
+    _adapter_disconnect = connection.Connection._adapter_disconnect
+    _flush_outbound = connection.Connection._flush_outbound
+    add_timeout = connection.Connection.add_timeout
+    remove_timeout = connection.Connection.remove_timeout
+
+
 class ChannelTests(unittest.TestCase):
 
-    @mock.patch('pika.connection.Connection')
-    def _create_connection(self, connection=None):
-        return connection
+    @mock.patch('pika.connection.Connection', autospec=ConnectionTemplate)
+    def _create_connection(self, connectionClassMock=None):
+        return connectionClassMock()
 
     def setUp(self):
         self.connection = self._create_connection()
@@ -440,14 +462,12 @@ class ChannelTests(unittest.TestCase):
         self.obj.confirm_delivery(logging.debug)
         self.obj.callbacks.add.assert_called_with(*expectation, arguments=None)
 
-    def test_confirm_delivery_callback_with_nowait(self):
+    def test_confirm_delivery_callback_with_nowait_raises_value_error(self):
         self.obj._set_state(self.obj.OPEN)
         expectation = [self.obj.channel_number, spec.Confirm.SelectOk,
                        self.obj._on_selectok]
-        self.obj.confirm_delivery(logging.debug, True)
-        self.assertNotIn(mock.call(*expectation,
-                                   arguments=None),
-                         self.obj.callbacks.add.call_args_list)
+        with self.assertRaises(ValueError):
+            self.obj.confirm_delivery(logging.debug, True)
 
     def test_confirm_delivery_callback_basic_ack(self):
         self.obj._set_state(self.obj.OPEN)
@@ -847,7 +867,6 @@ class ChannelTests(unittest.TestCase):
 
     def test_add_callbacks_basic_get_empty_added(self):
         self.obj._add_callbacks()
-        print(self.obj.callbacks.add.__dict__)
         self.obj.callbacks.add.assert_any_call(self.obj.channel_number,
                                                spec.Basic.GetEmpty,
                                                self.obj._on_getempty, False)
@@ -1153,20 +1172,22 @@ class ChannelTests(unittest.TestCase):
 
     def test_rpc_raises_channel_closed(self):
         self.assertRaises(exceptions.ChannelClosed, self.obj._rpc,
-                          frame.Method(self.obj.channel_number,
-                                       spec.Basic.Ack(1)))
+                          spec.Basic.Cancel('tag_abc'))
 
     def test_rpc_while_blocking_appends_blocked_collection(self):
         self.obj._set_state(self.obj.OPEN)
         self.obj._blocking = spec.Confirm.Select()
-        expectation = [frame.Method(self.obj.channel_number, spec.Basic.Ack(1)),
-                       'Foo', None]
+        acceptable_replies = [
+            (spec.Basic.CancelOk, {'consumer_tag': 'tag_abc'})]
+        expectation = [spec.Basic.Cancel('tag_abc'), lambda *args: None,
+                       acceptable_replies]
         self.obj._rpc(*expectation)
         self.assertIn(expectation, self.obj._blocked)
 
     def test_rpc_throws_value_error_with_unacceptable_replies(self):
         self.obj._set_state(self.obj.OPEN)
-        self.assertRaises(TypeError, self.obj._rpc, spec.Basic.Ack(1),
+        self.assertRaises(TypeError, self.obj._rpc,
+                          spec.Basic.Cancel('tag_abc'),
                           logging.debug, 'Foo')
 
     def test_rpc_throws_type_error_with_invalid_callback(self):
@@ -1174,15 +1195,27 @@ class ChannelTests(unittest.TestCase):
         self.assertRaises(TypeError, self.obj._rpc, spec.Channel.Open(1),
                           ['foo'], [spec.Channel.OpenOk])
 
-    def test_rpc_adds_on_synchronous_complete(self):
+    def test_rpc_enters_blocking_and_adds_on_synchronous_complete(self):
         self.obj._set_state(self.obj.OPEN)
         method_frame = spec.Channel.Open()
         self.obj._rpc(method_frame, None, [spec.Channel.OpenOk])
+        self.assertEqual(self.obj._blocking, method_frame.NAME)
         self.obj.callbacks.add.assert_called_with(
             self.obj.channel_number, spec.Channel.OpenOk,
             self.obj._on_synchronous_complete,
             arguments=None)
 
+    def test_rpc_not_blocking_and_no_on_synchronous_complete_when_no_replies(self):
+        self.obj._set_state(self.obj.OPEN)
+        method_frame = spec.Channel.Open()
+        self.obj._rpc(method_frame, None, acceptable_replies=[])
+        self.assertIsNone(self.obj._blocking)
+        with self.assertRaises(AssertionError):
+            self.obj.callbacks.add.assert_called_with(
+                mock.ANY, mock.ANY,
+                self.obj._on_synchronous_complete,
+                arguments=mock.ANY)
+
     def test_rpc_adds_callback(self):
         self.obj._set_state(self.obj.OPEN)
         method_frame = spec.Channel.Open()