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!
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)
 

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?
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()

pip install -e .[dev]
parse_log_pytest
requirements.txt
yapf pytest
apt-get update apt-get install -y gcc libev-dev
3.5
test-requirements.txt
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
attrs==22.2.0
certifi==2021.5.30
charset-normalizer==2.0.12
codecov==2.1.13
coverage==6.2
idna==3.10
importlib-metadata==4.8.3
iniconfig==1.1.1
mock==5.2.0
nose==1.3.7
packaging==21.3
-e git+https://github.com/pika/pika.git@d8a782d97579cd96ed67ccfb55f63ca8fdafa199#egg=pika
pluggy==1.0.0
py==1.11.0
pyparsing==3.1.4
pytest==7.0.1
requests==2.27.1
tomli==1.2.3
tornado==6.1
Twisted==15.3.0
typing_extensions==4.1.1
urllib3==1.26.20
yapf==0.32.0
zipp==3.6.0
zope.interface==5.5.2

name: pika
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - ca-certificates=2025.2.25=h06a4308_0
  - certifi=2021.5.30=py36h06a4308_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.3=he6710b0_2
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - ncurses=6.4=h6a678d5_0
  - openssl=1.1.1w=h7f8727e_0
  - pip=21.2.2=py36h06a4308_0
  - python=3.6.13=h12debd9_1
  - readline=8.2=h5eee18b_0
  - setuptools=58.0.4=py36h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - wheel=0.37.1=pyhd3eb1b0_0
  - xz=5.6.4=h5eee18b_1
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - attrs==22.2.0
      - charset-normalizer==2.0.12
      - codecov==2.1.13
      - coverage==6.2
      - idna==3.10
      - importlib-metadata==4.8.3
      - iniconfig==1.1.1
      - mock==5.2.0
      - nose==1.3.7
      - packaging==21.3
      - pluggy==1.0.0
      - py==1.11.0
      - pyparsing==3.1.4
      - pytest==7.0.1
      - requests==2.27.1
      - tomli==1.2.3
      - tornado==6.1
      - twisted==15.3.0
      - typing-extensions==4.1.1
      - urllib3==1.26.20
      - yapf==0.32.0
      - zipp==3.6.0
      - zope-interface==5.5.2
prefix: /opt/conda/envs/pika

tests/unit/channel_tests.py::ChannelTests::test_confirm_delivery_callback_with_nowait_raises_value_error tests/unit/channel_tests.py::ChannelTests::test_rpc_not_blocking_and_no_on_synchronous_complete_when_no_replies
tests/unit/channel_tests.py::ChannelTests::test_add_callback tests/unit/channel_tests.py::ChannelTests::test_add_callback_multiple_replies tests/unit/channel_tests.py::ChannelTests::test_add_callbacks_basic_cancel_empty_added tests/unit/channel_tests.py::ChannelTests::test_add_callbacks_basic_get_empty_added tests/unit/channel_tests.py::ChannelTests::test_add_callbacks_channel_close_added tests/unit/channel_tests.py::ChannelTests::test_add_callbacks_channel_flow_added tests/unit/channel_tests.py::ChannelTests::test_add_on_cancel_callback tests/unit/channel_tests.py::ChannelTests::test_add_on_close_callback tests/unit/channel_tests.py::ChannelTests::test_add_on_flow_callback tests/unit/channel_tests.py::ChannelTests::test_add_on_return_callback tests/unit/channel_tests.py::ChannelTests::test_basic_ack_channel_closed tests/unit/channel_tests.py::ChannelTests::test_basic_cancel_callback_appended tests/unit/channel_tests.py::ChannelTests::test_basic_cancel_calls_validate tests/unit/channel_tests.py::ChannelTests::test_basic_cancel_channel_cancelled_appended tests/unit/channel_tests.py::ChannelTests::test_basic_cancel_no_consumer_tag tests/unit/channel_tests.py::ChannelTests::test_basic_cancel_on_cancel_appended tests/unit/channel_tests.py::ChannelTests::test_basic_cancel_raises_value_error tests/unit/channel_tests.py::ChannelTests::test_basic_cancel_then_close tests/unit/channel_tests.py::ChannelTests::test_basic_consume_calls_validate tests/unit/channel_tests.py::ChannelTests::test_basic_consume_channel_closed tests/unit/channel_tests.py::ChannelTests::test_basic_consume_consumer_tag tests/unit/channel_tests.py::ChannelTests::test_basic_consume_consumer_tag_cancelled_full tests/unit/channel_tests.py::ChannelTests::test_basic_consume_consumer_tag_in_consumers tests/unit/channel_tests.py::ChannelTests::test_basic_consume_consumers_callback_value tests/unit/channel_tests.py::ChannelTests::test_basic_consume_consumers_pending_list_is_empty tests/unit/channel_tests.py::ChannelTests::test_basic_consume_consumers_rpc_called tests/unit/channel_tests.py::ChannelTests::test_basic_consume_duplicate_consumer_tag_raises tests/unit/channel_tests.py::ChannelTests::test_basic_consume_has_pending_list tests/unit/channel_tests.py::ChannelTests::test_basic_get_callback tests/unit/channel_tests.py::ChannelTests::test_basic_get_calls_validate tests/unit/channel_tests.py::ChannelTests::test_basic_get_send_method_called tests/unit/channel_tests.py::ChannelTests::test_basic_nack_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_basic_nack_send_method_request tests/unit/channel_tests.py::ChannelTests::test_basic_publish_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_basic_publish_send_method_request tests/unit/channel_tests.py::ChannelTests::test_basic_qos_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_basic_qos_rpc_request tests/unit/channel_tests.py::ChannelTests::test_basic_recover_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_basic_recover_rpc_request tests/unit/channel_tests.py::ChannelTests::test_basic_reject_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_basic_reject_send_method_request tests/unit/channel_tests.py::ChannelTests::test_basic_send_method_calls_rpc tests/unit/channel_tests.py::ChannelTests::test_channel_open_add_callbacks_called tests/unit/channel_tests.py::ChannelTests::test_cleanup tests/unit/channel_tests.py::ChannelTests::test_close_basic_cancel_called tests/unit/channel_tests.py::ChannelTests::test_close_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_close_state tests/unit/channel_tests.py::ChannelTests::test_confirm_delivery_callback_basic_ack tests/unit/channel_tests.py::ChannelTests::test_confirm_delivery_callback_basic_nack tests/unit/channel_tests.py::ChannelTests::test_confirm_delivery_callback_without_nowait_selectok tests/unit/channel_tests.py::ChannelTests::test_confirm_delivery_no_callback_callback_call_count tests/unit/channel_tests.py::ChannelTests::test_confirm_delivery_no_callback_no_basic_ack_callback tests/unit/channel_tests.py::ChannelTests::test_confirm_delivery_no_callback_no_basic_nack_callback tests/unit/channel_tests.py::ChannelTests::test_confirm_delivery_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_confirm_delivery_raises_method_not_implemented_for_confirms tests/unit/channel_tests.py::ChannelTests::test_confirm_delivery_raises_method_not_implemented_for_nack tests/unit/channel_tests.py::ChannelTests::test_consumer_tags tests/unit/channel_tests.py::ChannelTests::test_exchange_bind_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_exchange_bind_raises_value_error_on_invalid_callback tests/unit/channel_tests.py::ChannelTests::test_exchange_bind_rpc_request tests/unit/channel_tests.py::ChannelTests::test_exchange_bind_rpc_request_nowait tests/unit/channel_tests.py::ChannelTests::test_exchange_declare_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_exchange_declare_raises_value_error_on_invalid_callback tests/unit/channel_tests.py::ChannelTests::test_exchange_declare_rpc_request tests/unit/channel_tests.py::ChannelTests::test_exchange_declare_rpc_request_nowait tests/unit/channel_tests.py::ChannelTests::test_exchange_declare_with_type_arg_assigns_to_exchange_type tests/unit/channel_tests.py::ChannelTests::test_exchange_declare_with_type_arg_raises_deprecation_warning tests/unit/channel_tests.py::ChannelTests::test_exchange_delete_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_exchange_delete_raises_value_error_on_invalid_callback tests/unit/channel_tests.py::ChannelTests::test_exchange_delete_rpc_request tests/unit/channel_tests.py::ChannelTests::test_exchange_delete_rpc_request_nowait tests/unit/channel_tests.py::ChannelTests::test_exchange_unbind_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_exchange_unbind_raises_value_error_on_invalid_callback tests/unit/channel_tests.py::ChannelTests::test_exchange_unbind_rpc_request tests/unit/channel_tests.py::ChannelTests::test_exchange_unbind_rpc_request_nowait tests/unit/channel_tests.py::ChannelTests::test_flow_off_rpc_request tests/unit/channel_tests.py::ChannelTests::test_flow_on_flowok_callback tests/unit/channel_tests.py::ChannelTests::test_flow_on_rpc_request tests/unit/channel_tests.py::ChannelTests::test_flow_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_flow_raises_invalid_callback tests/unit/channel_tests.py::ChannelTests::test_get_pending_message tests/unit/channel_tests.py::ChannelTests::test_get_pending_message_item_popped tests/unit/channel_tests.py::ChannelTests::test_handle_content_frame_basic_deliver_called tests/unit/channel_tests.py::ChannelTests::test_handle_content_frame_basic_get_called tests/unit/channel_tests.py::ChannelTests::test_handle_content_frame_basic_return_called tests/unit/channel_tests.py::ChannelTests::test_handle_content_frame_method_returns_none tests/unit/channel_tests.py::ChannelTests::test_handle_content_frame_sets_header_frame tests/unit/channel_tests.py::ChannelTests::test_handle_content_frame_sets_method_frame tests/unit/channel_tests.py::ChannelTests::test_has_content_false tests/unit/channel_tests.py::ChannelTests::test_has_content_true tests/unit/channel_tests.py::ChannelTests::test_immediate_called_logger_warning tests/unit/channel_tests.py::ChannelTests::test_init_blocked tests/unit/channel_tests.py::ChannelTests::test_init_blocking tests/unit/channel_tests.py::ChannelTests::test_init_callbacks tests/unit/channel_tests.py::ChannelTests::test_init_cancelled tests/unit/channel_tests.py::ChannelTests::test_init_channel_number tests/unit/channel_tests.py::ChannelTests::test_init_connection tests/unit/channel_tests.py::ChannelTests::test_init_consumers tests/unit/channel_tests.py::ChannelTests::test_init_frame_dispatcher tests/unit/channel_tests.py::ChannelTests::test_init_has_on_flow_callback tests/unit/channel_tests.py::ChannelTests::test_init_invalid_channel_number tests/unit/channel_tests.py::ChannelTests::test_init_on_flowok_callback tests/unit/channel_tests.py::ChannelTests::test_init_on_getok_callback tests/unit/channel_tests.py::ChannelTests::test_init_on_openok_callback tests/unit/channel_tests.py::ChannelTests::test_init_pending tests/unit/channel_tests.py::ChannelTests::test_init_state tests/unit/channel_tests.py::ChannelTests::test_is_closed_false tests/unit/channel_tests.py::ChannelTests::test_is_closed_true tests/unit/channel_tests.py::ChannelTests::test_is_closing_false tests/unit/channel_tests.py::ChannelTests::test_is_closing_true tests/unit/channel_tests.py::ChannelTests::test_on_cancel_not_appended_cancelled tests/unit/channel_tests.py::ChannelTests::test_on_cancel_removed_consumer tests/unit/channel_tests.py::ChannelTests::test_on_cancelok_removed_consumer tests/unit/channel_tests.py::ChannelTests::test_on_cancelok_removed_pending tests/unit/channel_tests.py::ChannelTests::test_on_close_warning tests/unit/channel_tests.py::ChannelTests::test_on_confirm_selectok tests/unit/channel_tests.py::ChannelTests::test_on_deliver_callback_called tests/unit/channel_tests.py::ChannelTests::test_on_deliver_pending_callbacks_called tests/unit/channel_tests.py::ChannelTests::test_on_deliver_pending_called tests/unit/channel_tests.py::ChannelTests::test_on_eventok tests/unit/channel_tests.py::ChannelTests::test_on_flow tests/unit/channel_tests.py::ChannelTests::test_on_flow_with_callback tests/unit/channel_tests.py::ChannelTests::test_on_flowok tests/unit/channel_tests.py::ChannelTests::test_on_flowok_callback_reset tests/unit/channel_tests.py::ChannelTests::test_on_flowok_calls_callback tests/unit/channel_tests.py::ChannelTests::test_on_getempty tests/unit/channel_tests.py::ChannelTests::test_on_getok_callback_called tests/unit/channel_tests.py::ChannelTests::test_on_getok_callback_reset tests/unit/channel_tests.py::ChannelTests::test_on_getok_no_callback tests/unit/channel_tests.py::ChannelTests::test_on_openok_callback_called tests/unit/channel_tests.py::ChannelTests::test_on_openok_no_callback tests/unit/channel_tests.py::ChannelTests::test_on_synchronous_complete tests/unit/channel_tests.py::ChannelTests::test_onreturn tests/unit/channel_tests.py::ChannelTests::test_onreturn_warning tests/unit/channel_tests.py::ChannelTests::test_queue_bind_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_queue_bind_raises_value_error_on_invalid_callback tests/unit/channel_tests.py::ChannelTests::test_queue_bind_rpc_request tests/unit/channel_tests.py::ChannelTests::test_queue_bind_rpc_request_nowait tests/unit/channel_tests.py::ChannelTests::test_queue_declare_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_queue_declare_raises_value_error_on_invalid_callback tests/unit/channel_tests.py::ChannelTests::test_queue_declare_rpc_request tests/unit/channel_tests.py::ChannelTests::test_queue_declare_rpc_request_nowait tests/unit/channel_tests.py::ChannelTests::test_queue_delete_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_queue_delete_raises_value_error_on_invalid_callback tests/unit/channel_tests.py::ChannelTests::test_queue_delete_rpc_request tests/unit/channel_tests.py::ChannelTests::test_queue_delete_rpc_request_nowait tests/unit/channel_tests.py::ChannelTests::test_queue_purge_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_queue_purge_raises_value_error_on_invalid_callback tests/unit/channel_tests.py::ChannelTests::test_queue_purge_rpc_request tests/unit/channel_tests.py::ChannelTests::test_queue_purge_rpc_request_nowait tests/unit/channel_tests.py::ChannelTests::test_queue_unbind_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_queue_unbind_raises_value_error_on_invalid_callback tests/unit/channel_tests.py::ChannelTests::test_queue_unbind_rpc_request tests/unit/channel_tests.py::ChannelTests::test_rpc_adds_callback tests/unit/channel_tests.py::ChannelTests::test_rpc_enters_blocking_and_adds_on_synchronous_complete tests/unit/channel_tests.py::ChannelTests::test_rpc_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_rpc_throws_type_error_with_invalid_callback tests/unit/channel_tests.py::ChannelTests::test_rpc_throws_value_error_with_unacceptable_replies tests/unit/channel_tests.py::ChannelTests::test_rpc_while_blocking_appends_blocked_collection tests/unit/channel_tests.py::ChannelTests::test_send_method tests/unit/channel_tests.py::ChannelTests::test_set_state tests/unit/channel_tests.py::ChannelTests::test_tx_commit_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_tx_commit_rpc_request tests/unit/channel_tests.py::ChannelTests::test_tx_rollback_rpc_request tests/unit/channel_tests.py::ChannelTests::test_tx_select_rpc_request tests/unit/channel_tests.py::ChannelTests::test_validate_channel_and_callback_raises_channel_closed tests/unit/channel_tests.py::ChannelTests::test_validate_channel_and_callback_raises_value_error_not_callable
pika/channel.py pika/adapters/libev_connection.py
pika/channel.py pika/adapters/libev_connection.py