Add qpid to master (#617)

* Updates master with the latest from 3.0

The Qpid transport on the 3.0 branch is working well
but has not been updated on the master branch. This
brings the version on master up to date.

* Tests are now running as expected

3 files changed, 1451 insertions, 1606 deletions

diff --git a/kombu/tests/transport/test_qpid.py b/kombu/tests/transport/test_qpid.py
index d25ed0b1..0e913d84 100644
--- a/kombu/tests/transport/test_qpid.py
+++ b/kombu/tests/transport/test_qpid.py
@@ -1,97 +1,143 @@
 from __future__ import absolute_import, unicode_literals
 
-import datetime
 import select
 import ssl
 import socket
 import sys
-import threading
 import time
+import uuid
 
-from collections import OrderedDict
+from collections import Callable, OrderedDict
 from itertools import count
+from functools import wraps
 
-from kombu.five import Empty, bytes_if_py2
-from kombu.transport.qpid import AuthenticationFailure, QoS, Message
-from kombu.transport.qpid import QpidMessagingExceptionHandler, Channel
-from kombu.transport.qpid import Connection, ReceiversMonitor, Transport
-from kombu.transport.qpid import ConnectionError
+from mock import call
+from nose import SkipTest
+
+from kombu.five import Empty, keys, range, monotonic
+from kombu.transport.qpid import (AuthenticationFailure, Channel, Connection,
+                                  ConnectionError, Message, NotFound, QoS,
+                                  Transport)
 from kombu.transport.virtual import Base64
-from kombu.tests.case import Case, Mock, call, skip
+from kombu.tests.case import Case, Mock
 from kombu.tests.case import patch
 
+
 QPID_MODULE = 'kombu.transport.qpid'
+PY3 = sys.version_info[0] == 3
 
 
-@skip.if_python3()
-@skip.if_pypy()
-class QPidCase(Case):
-    pass
+def case_no_pypy(cls):
+    setup = cls.setUp
 
+    @wraps(setup)
+    def around_setup(self):
+        if getattr(sys, 'pypy_version_info', None):
+            raise SkipTest('pypy incompatible')
+        setup(self)
+    cls.setUp = around_setup
+    return cls
 
-class MockException(Exception):
-    pass
 
+def case_no_python3(cls):
+    setup = cls.setUp
 
-class BreakOutException(Exception):
-    pass
+    @wraps(setup)
+    def around_setup(self):
+        if PY3:
+            raise SkipTest('Python3 incompatible')
+        setup(self)
+    cls.setUp = around_setup
+    return cls
 
 
-class test_QPidMessagingExceptionHandler(QPidCase):
-    allowed_string = 'object in use'
-    not_allowed_string = 'a different string'
+def disable_runtime_dependency_check(cls):
+    """A decorator to disable runtime dependency checking"""
+    setup = cls.setUp
+    teardown = cls.tearDown
+    dependency_is_none_patcher = patch(QPID_MODULE + '.dependency_is_none')
 
-    def setup(self):
-        # Create a mock ExceptionHandler for testing by this object.
-        self.handler = QpidMessagingExceptionHandler(self.allowed_string)
+    @wraps(setup)
+    def around_setup(self):
+        mock_dependency_is_none = dependency_is_none_patcher.start()
+        mock_dependency_is_none.return_value = False
+        setup(self)
 
-    def test_string_stored(self):
-        handler_string = self.handler.allowed_exception_string
-        self.assertEqual(self.allowed_string, handler_string)
+    @wraps(setup)
+    def around_teardown(self):
+        dependency_is_none_patcher.stop()
+        teardown(self)
 
-    def test_exception_positive(self):
-        exception_to_raise = Exception(self.allowed_string)
+    cls.setUp = around_setup
+    cls.tearDown = around_teardown
+    return cls
 
-        def exception_raise_fun():
-            raise exception_to_raise
-        decorated_fun = self.handler(exception_raise_fun)
-        try:
-            decorated_fun()
-        except:
-            self.fail('QpidMessagingExceptionHandler allowed an exception '
-                      'to be raised that should have been silenced!')
 
-    def test_exception_negative(self):
-        exception_to_raise = Exception(self.not_allowed_string)
+class ExtraAssertionsMixin(object):
+    """A mixin class adding assertDictEqual and assertDictContainsSubset"""
+
+    def assertDictEqual(self, a, b, msg=None):
+        """
+        Test that two dictionaries are equal.
+
+        Implemented here because this method was not available until Python
+        2.6. This asserts that the unique set of keys are the same in a and b.
+        Also asserts that the value of each key is the same in a and b using
+        the is operator.
+        """
+        self.assertEqual(set(keys(a)), set(keys(b)))
+        for key in keys(a):
+            self.assertEqual(a[key], b[key])
+
+    def assertDictContainsSubset(self, a, b, msg=None):
+        """
+        Assert that all the key/value pairs in a exist in b.
+        """
+        for key in keys(a):
+            self.assertIn(key, b)
+            self.assertEqual(a[key], b[key])
+
+
+class QpidException(Exception):
+    """
+    An object used to mock Exceptions provided by qpid.messaging.exceptions
+    """
+
+    def __init__(self, code=None, text=None):
+        super(Exception, self).__init__(self)
+        self.code = code
+        self.text = text
 
-        def exception_raise_fun():
-            raise exception_to_raise
-        decorated_fun = self.handler(exception_raise_fun)
-        with self.assertRaises(Exception):
-            decorated_fun()
 
+class BreakOutException(Exception):
+    pass
 
-class test_QoS__init__(QPidCase):
 
-    def setup(self):
-        self.session = Mock(name='session')
-        self.qos = QoS(self.session)
+@case_no_python3
+@case_no_pypy
+class TestQoS__init__(Case):
+
+    def setUp(self):
+        self.mock_session = Mock()
+        self.qos = QoS(self.mock_session)
 
-    def test_prefetch_default_set_correct_without_prefetch_value(self):
+    def test__init__prefetch_default_set_correct_without_prefetch_value(self):
         self.assertEqual(self.qos.prefetch_count, 1)
 
-    def test_prefetch_is_hard_set_to_one(self):
-        qos_limit_two = QoS(self.session)
+    def test__init__prefetch_is_hard_set_to_one(self):
+        qos_limit_two = QoS(self.mock_session)
         self.assertEqual(qos_limit_two.prefetch_count, 1)
 
-    def test_not_yet_acked_is_initialized(self):
+    def test__init___not_yet_acked_is_initialized(self):
         self.assertIsInstance(self.qos._not_yet_acked, OrderedDict)
 
 
-class test_QoS__can_consume(QPidCase):
+@case_no_python3
+@case_no_pypy
+class TestQoSCanConsume(Case):
 
-    def setup(self):
-        session = Mock(name='session')
+    def setUp(self):
+        session = Mock()
         self.qos = QoS(session)
 
     def test_True_when_prefetch_limit_is_zero(self):
@@ -110,11 +156,13 @@ class test_QoS__can_consume(QPidCase):
         self.assertFalse(self.qos.can_consume())
 
 
-class test_QoS__can_consume_max_estimate(QPidCase):
+@case_no_python3
+@case_no_pypy
+class TestQoSCanConsumeMaxEstimate(Case):
 
-    def setup(self):
-        self.session = Mock(name='session')
-        self.qos = QoS(self.session)
+    def setUp(self):
+        self.mock_session = Mock()
+        self.qos = QoS(self.mock_session)
 
     def test_return_one_when_prefetch_count_eq_zero(self):
         self.qos.prefetch_count = 0
@@ -126,94 +174,104 @@ class test_QoS__can_consume_max_estimate(QPidCase):
         self.assertEqual(self.qos.can_consume_max_estimate(), 2)
 
 
-class test_QoS__ack(QPidCase):
+@case_no_python3
+@case_no_pypy
+class TestQoSAck(Case):
 
-    def setup(self):
-        self.session = Mock(name='session')
-        self.qos = QoS(self.session)
+    def setUp(self):
+        self.mock_session = Mock()
+        self.qos = QoS(self.mock_session)
 
-    def test_pops__not_yet_acked(self):
-        message = Mock(name='message')
+    def test_ack_pops__not_yet_acked(self):
+        message = Mock()
         self.qos.append(message, 1)
         self.assertIn(1, self.qos._not_yet_acked)
         self.qos.ack(1)
         self.assertNotIn(1, self.qos._not_yet_acked)
 
-    def test_calls_session_acknowledge_with_message(self):
-        message = Mock(name='message')
+    def test_ack_calls_session_acknowledge_with_message(self):
+        message = Mock()
         self.qos.append(message, 1)
         self.qos.ack(1)
         self.qos.session.acknowledge.assert_called_with(message=message)
 
 
-class test_QoS__reject(QPidCase):
+@case_no_python3
+@case_no_pypy
+class TestQoSReject(Case):
 
-    def setup(self):
-        self.session = Mock(name='session')
-        self.message = Mock(name='message')
-        self.qos = QoS(self.session)
+    def setUp(self):
+        self.mock_session = Mock()
+        self.mock_message = Mock()
+        self.qos = QoS(self.mock_session)
         self.patch_qpid = patch(QPID_MODULE + '.qpid')
-        self.qpid = self.patch_qpid.start()
-        self.Disposition = self.qpid.messaging.Disposition
-        self.RELEASED = self.qpid.messaging.RELEASED
-        self.REJECTED = self.qpid.messaging.REJECTED
+        self.mock_qpid = self.patch_qpid.start()
+        self.mock_Disposition = self.mock_qpid.messaging.Disposition
+        self.mock_RELEASED = self.mock_qpid.messaging.RELEASED
+        self.mock_REJECTED = self.mock_qpid.messaging.REJECTED
 
-    def teardown(self):
+    def tearDown(self):
         self.patch_qpid.stop()
 
-    def test_pops__not_yet_acked(self):
-        self.qos.append(self.message, 1)
+    def test_reject_pops__not_yet_acked(self):
+        self.qos.append(self.mock_message, 1)
         self.assertIn(1, self.qos._not_yet_acked)
         self.qos.reject(1)
         self.assertNotIn(1, self.qos._not_yet_acked)
 
-    def test_requeue_true(self):
-        self.qos.append(self.message, 1)
+    def test_reject_requeue_true(self):
+        self.qos.append(self.mock_message, 1)
         self.qos.reject(1, requeue=True)
-        self.Disposition.assert_called_with(self.RELEASED)
+        self.mock_Disposition.assert_called_with(self.mock_RELEASED)
         self.qos.session.acknowledge.assert_called_with(
-            message=self.message,
-            disposition=self.Disposition.return_value)
+            message=self.mock_message,
+            disposition=self.mock_Disposition.return_value,
+        )
 
-    def test_requeue_false(self):
-        message = Mock(name='message')
+    def test_reject_requeue_false(self):
+        message = Mock()
         self.qos.append(message, 1)
         self.qos.reject(1, requeue=False)
-        self.Disposition.assert_called_with(self.REJECTED)
+        self.mock_Disposition.assert_called_with(self.mock_REJECTED)
         self.qos.session.acknowledge.assert_called_with(
-            message=message, disposition=self.Disposition.return_value)
-
+            message=message, disposition=self.mock_Disposition.return_value,
+        )
 
-class test_QoS(QPidCase):
 
-    def setup(self):
-        self.session = Mock(name='session')
-        self.qos_no_limit = QoS(self.session)
-        self.qos_limit_2 = QoS(self.session, prefetch_count=2)
-        self.delivery_tag_generator = count(1)
+@case_no_python3
+@case_no_pypy
+class TestQoS(Case):
 
     def mock_message_factory(self):
-        # Create and return a mock message tag and delivery_tag.
+        """Create and return a mock message tag and delivery_tag."""
         m_delivery_tag = self.delivery_tag_generator.next()
-        m = 'message %s' % m_delivery_tag
-        return (m, m_delivery_tag)
+        m = 'message %s' % (m_delivery_tag, )
+        return m, m_delivery_tag
 
     def add_n_messages_to_qos(self, n, qos):
-        # Add N mock messages into the passed in qos object.
+        """Add N mock messages into the passed in qos object"""
         for i in range(n):
             self.add_message_to_qos(qos)
 
     def add_message_to_qos(self, qos):
-        # Add a single mock message into the passed in qos object.
-        #
-        # Uses the mock_message_factory() to create the message and
-        # delivery_tag.
+        """Add a single mock message into the passed in qos object.
+
+        Uses the mock_message_factory() to create the message and
+        delivery_tag.
+        """
         m, m_delivery_tag = self.mock_message_factory()
         qos.append(m, m_delivery_tag)
 
+    def setUp(self):
+        self.mock_session = Mock()
+        self.qos_no_limit = QoS(self.mock_session)
+        self.qos_limit_2 = QoS(self.mock_session, prefetch_count=2)
+        self.delivery_tag_generator = count(1)
+
     def test_append(self):
-        # Append two messages and check inside the QoS object that
-        # were put into the internal data structures correctly.
+        """Append two messages and check inside the QoS object that they
+        were put into the internal data structures correctly
+        """
         qos = self.qos_no_limit
         m1, m1_tag = self.mock_message_factory()
         m2, m2_tag = self.mock_message_factory()
@@ -229,7 +287,7 @@ class test_QoS(QPidCase):
         self.assertIs(m2, checked_message2)
 
     def test_get(self):
-        # Append two messages, and use get to receive them.
+        """Append two messages, and use get to receive them"""
         qos = self.qos_no_limit
         m1, m1_tag = self.mock_message_factory()
         m2, m2_tag = self.mock_message_factory()
@@ -241,694 +299,763 @@ class test_QoS(QPidCase):
         self.assertIs(m2, message2)
 
 
-@skip.if_python3()
-@skip.if_pypy()
-class ConnectionCase(QPidCase):
+@case_no_python3
+@case_no_pypy
+class ConnectionTestBase(Case):
 
     @patch(QPID_MODULE + '.qpid')
-    def setUp(self, qpid):
+    def setUp(self, mock_qpid):
         self.connection_options = {
             'host': 'localhost',
             'port': 5672,
-            'username': 'guest',
-            'password': '',
             'transport': 'tcp',
             'timeout': 10,
-            'sasl_mechanisms': 'ANONYMOUS PLAIN',
+            'sasl_mechanisms': 'ANONYMOUS',
         }
-        self.qpid_connection = qpid.messaging.Connection
+        self.mock_qpid_connection = mock_qpid.messaging.Connection
         self.conn = Connection(**self.connection_options)
-        super(ConnectionCase, self).setUp()
 
 
-class test_Connection__init__(ConnectionCase):
+@case_no_python3
+@case_no_pypy
+class TestConnectionInit(ExtraAssertionsMixin, ConnectionTestBase):
 
     def test_stores_connection_options(self):
         # ensure that only one mech was passed into connection. The other
         # options should all be passed through as-is
         modified_conn_opts = self.connection_options
-        modified_conn_opts['sasl_mechanisms'] = 'PLAIN'
-        self.assertDictEqual(modified_conn_opts,
-                             self.conn.connection_options)
+        self.assertDictEqual(
+            modified_conn_opts, self.conn.connection_options,
+        )
 
-    def test_variables(self):
+    def test_class_variables(self):
         self.assertIsInstance(self.conn.channels, list)
         self.assertIsInstance(self.conn._callbacks, dict)
 
     def test_establishes_connection(self):
         modified_conn_opts = self.connection_options
-        modified_conn_opts['sasl_mechanisms'] = 'PLAIN'
-        self.qpid_connection.establish.assert_called_with(
-            **modified_conn_opts)
+        self.mock_qpid_connection.establish.assert_called_with(
+            **modified_conn_opts
+        )
 
     def test_saves_established_connection(self):
-        created_conn = self.qpid_connection.establish.return_value
+        created_conn = self.mock_qpid_connection.establish.return_value
         self.assertIs(self.conn._qpid_conn, created_conn)
 
-    @patch(QPID_MODULE + '.ConnectionError', new=(MockException,))
+    @patch(QPID_MODULE + '.ConnectionError', new=(QpidException, ))
     @patch(QPID_MODULE + '.sys.exc_info')
     @patch(QPID_MODULE + '.qpid')
-    def test_mutates_ConnError_by_message(self, qpid, exc_info):
-        my_conn_error = MockException()
-        my_conn_error.text = bytes_if_py2(
-            'connection-forced: Authentication failed(320)')
-        qpid.messaging.Connection.establish.side_effect = my_conn_error
-        exc_info.return_value = (bytes_if_py2('a'), bytes_if_py2('b'), None)
+    def test_mutates_ConnError_by_message(self, mock_qpid, mock_exc_info):
+        text = 'connection-forced: Authentication failed(320)'
+        my_conn_error = QpidException(text=text)
+        mock_qpid.messaging.Connection.establish.side_effect = my_conn_error
+        mock_exc_info.return_value = 'a', 'b', None
         try:
             self.conn = Connection(**self.connection_options)
         except AuthenticationFailure as error:
             exc_info = sys.exc_info()
-            self.assertNotIsInstance(error, MockException)
-            self.assertEqual(exc_info[1], 'b')
+            self.assertNotIsInstance(error, QpidException)
+            self.assertIs(exc_info[1], 'b')
             self.assertIsNone(exc_info[2])
         else:
             self.fail('ConnectionError type was not mutated correctly')
 
-    @patch(QPID_MODULE + '.ConnectionError', new=(MockException,))
+    @patch(QPID_MODULE + '.ConnectionError', new=(QpidException, ))
     @patch(QPID_MODULE + '.sys.exc_info')
     @patch(QPID_MODULE + '.qpid')
-    def test_mutates_ConnError_by_code(self, qpid, exc_info):
-        my_conn_error = MockException()
-        my_conn_error.code = 320
-        my_conn_error.text = bytes_if_py2('someothertext')
-        qpid.messaging.Connection.establish.side_effect = my_conn_error
-        exc_info.return_value = (bytes_if_py2('a'), bytes_if_py2('b'), None)
+    def test_mutates_ConnError_by_code(self, mock_qpid, mock_exc_info):
+        my_conn_error = QpidException(code=320, text='someothertext')
+        mock_qpid.messaging.Connection.establish.side_effect = my_conn_error
+        mock_exc_info.return_value = 'a', 'b', None
         try:
             self.conn = Connection(**self.connection_options)
         except AuthenticationFailure as error:
             exc_info = sys.exc_info()
-            self.assertNotIsInstance(error, MockException)
-            self.assertEqual(exc_info[1], bytes_if_py2('b'))
+            self.assertNotIsInstance(error, QpidException)
+            self.assertIs(exc_info[1], 'b')
             self.assertIsNone(exc_info[2])
         else:
             self.fail('ConnectionError type was not mutated correctly')
 
-    @patch(QPID_MODULE + '.ConnectionError', new=(MockException,))
+    @patch(QPID_MODULE + '.ConnectionError', new=(QpidException, ))
+    @patch(QPID_MODULE + '.sys.exc_info')
+    @patch(QPID_MODULE + '.qpid')
+    def test_connection__init__mutates_ConnError_by_message2(self, mock_qpid,
+                                                             mock_exc_info):
+        """
+        Test for PLAIN connection via python-saslwrapper, sans cyrus-sasl-plain
+
+        This test is specific for what is returned when we attempt to connect
+        with PLAIN mech and python-saslwrapper is installed, but
+        cyrus-sasl-plain is not installed.
+        """
+        my_conn_error = QpidException()
+        my_conn_error.text = 'Error in sasl_client_start (-4) SASL(-4): no '\
+                             'mechanism available'
+        mock_qpid.messaging.Connection.establish.side_effect = my_conn_error
+        mock_exc_info.return_value = ('a', 'b', None)
+        try:
+            self.conn = Connection(**self.connection_options)
+        except AuthenticationFailure as error:
+            exc_info = sys.exc_info()
+            self.assertTrue(not isinstance(error, QpidException))
+            self.assertTrue(exc_info[1] is 'b')
+            self.assertTrue(exc_info[2] is None)
+        else:
+            self.fail('ConnectionError type was not mutated correctly')
+
+    @patch(QPID_MODULE + '.ConnectionError', new=(QpidException, ))
     @patch(QPID_MODULE + '.sys.exc_info')
     @patch(QPID_MODULE + '.qpid')
-    def test_unknown_connection_error(self, qpid, exc_info):
+    def test_unknown_connection_error(self, mock_qpid, mock_exc_info):
         # If we get a connection error that we don't understand,
         # bubble it up as-is
-        my_conn_error = MockException()
-        my_conn_error.code = 999
-        my_conn_error.text = bytes_if_py2('someothertext')
-        qpid.messaging.Connection.establish.side_effect = my_conn_error
-        exc_info.return_value = ('a', 'b', None)
+        my_conn_error = QpidException(code=999, text='someothertext')
+        mock_qpid.messaging.Connection.establish.side_effect = my_conn_error
+        mock_exc_info.return_value = 'a', 'b', None
         try:
             self.conn = Connection(**self.connection_options)
         except Exception as error:
-            self.assertEqual(error.code, 999)
+            self.assertTrue(error.code == 999)
         else:
             self.fail('Connection should have thrown an exception')
 
-    @patch.object(Transport, 'channel_errors', new=(MockException,))
+    @patch.object(Transport, 'channel_errors', new=(QpidException, ))
     @patch(QPID_MODULE + '.qpid')
     @patch(QPID_MODULE + '.ConnectionError', new=IOError)
-    def test_non_qpid_error_raises(self, qpid):
-        Qpid_Connection = qpid.messaging.Connection
+    def test_non_qpid_error_raises(self, mock_qpid):
+        mock_Qpid_Connection = mock_qpid.messaging.Connection
         my_conn_error = SyntaxError()
-        my_conn_error.text = bytes_if_py2(
-            'some non auth related error message')
-        Qpid_Connection.establish.side_effect = my_conn_error
+        my_conn_error.text = 'some non auth related error message'
+        mock_Qpid_Connection.establish.side_effect = my_conn_error
         with self.assertRaises(SyntaxError):
             Connection(**self.connection_options)
 
     @patch(QPID_MODULE + '.qpid')
     @patch(QPID_MODULE + '.ConnectionError', new=IOError)
-    def test_non_auth_conn_error_raises(self, qpid):
-        Qpid_Connection = qpid.messaging.Connection
+    def test_non_auth_conn_error_raises(self, mock_qpid):
+        mock_Qpid_Connection = mock_qpid.messaging.Connection
         my_conn_error = IOError()
-        my_conn_error.text = bytes_if_py2(
-            'some non auth related error message')
-        Qpid_Connection.establish.side_effect = my_conn_error
+        my_conn_error.text = 'some non auth related error message'
+        mock_Qpid_Connection.establish.side_effect = my_conn_error
         with self.assertRaises(IOError):
             Connection(**self.connection_options)
 
 
-class test_Connection__class_attributes(ConnectionCase):
+@case_no_python3
+@case_no_pypy
+class TestConnectionClassAttributes(ConnectionTestBase):
 
-    def test_class_attributes(self):
+    def test_connection_verify_class_attributes(self):
         self.assertEqual(Channel, Connection.Channel)
 
 
-class test_Connection__get_qpid_connection(ConnectionCase):
+@case_no_python3
+@case_no_pypy
+class TestConnectionGetQpidConnection(ConnectionTestBase):
 
-    def test_get_qpid_connection(self):
-        self.conn._qpid_conn = Mock('qpid_conn')
+    def test_connection_get_qpid_connection(self):
+        self.conn._qpid_conn = Mock()
         returned_connection = self.conn.get_qpid_connection()
         self.assertIs(self.conn._qpid_conn, returned_connection)
 
 
-class test_Connection__close_channel(ConnectionCase):
+@case_no_python3
+@case_no_pypy
+class TestConnectionClose(ConnectionTestBase):
 
-    def setup(self):
-        self.conn.channels = Mock(name='conn.channels')
+    def test_connection_close(self):
+        self.conn._qpid_conn = Mock()
+        self.conn.close()
+        self.conn._qpid_conn.close.assert_called_once_with()
 
-    def test_removes_channel_from_channel_list(self):
-        channel = Mock(name='channel')
-        self.conn.close_channel(channel)
-        self.conn.channels.remove.assert_called_once_with(channel)
 
-    def test_handles_ValueError_being_raised(self):
-        self.conn.channels.remove = Mock(
-            name='conn.channels', side_effect=ValueError(),
-        )
-        try:
-            self.conn.close_channel(Mock())
-        except ValueError:
-            self.fail('ValueError should not have been raised')
-
-    def test_set_channel_connection_to_None(self):
-        channel = Mock(name='channel')
-        channel.connection = False
-        self.conn.channels.remove = Mock(
-            name='conn.channels', side_effect=ValueError(),
-        )
-        self.conn.close_channel(channel)
-        self.assertIsNone(channel.connection)
+@case_no_python3
+@case_no_pypy
+class TestConnectionCloseChannel(ConnectionTestBase):
+
+    def setUp(self):
+        super(TestConnectionCloseChannel, self).setUp()
+        self.conn.channels = Mock()
 
+    def test_connection_close_channel_removes_channel_from_channel_list(self):
+        mock_channel = Mock()
+        self.conn.close_channel(mock_channel)
+        self.conn.channels.remove.assert_called_once_with(mock_channel)
 
-@skip.if_python3()
-@skip.if_pypy()
-class ChannelCase(QPidCase):
+    def test_connection_close_channel_handles_ValueError_being_raised(self):
+        self.conn.channels.remove = Mock(side_effect=ValueError())
+        self.conn.close_channel(Mock())
+
+    def test_connection_close_channel_set_channel_connection_to_None(self):
+        mock_channel = Mock()
+        mock_channel.connection = False
+        self.conn.channels.remove = Mock(side_effect=ValueError())
+        self.conn.close_channel(mock_channel)
+        self.assertIsNone(mock_channel.connection)
+
+
+@case_no_python3
+@case_no_pypy
+class ChannelTestBase(Case):
 
     def setUp(self):
         self.patch_qpidtoollibs = patch(QPID_MODULE + '.qpidtoollibs')
-        self.qpidtoollibs = self.patch_qpidtoollibs.start()
-        self.broker_agent = self.qpidtoollibs.BrokerAgent
-        self.conn = Mock(name='conn')
-        self.transport = Mock(name='transport')
+        self.mock_qpidtoollibs = self.patch_qpidtoollibs.start()
+        self.mock_broker_agent = self.mock_qpidtoollibs.BrokerAgent
+        self.conn = Mock()
+        self.transport = Mock()
         self.channel = Channel(self.conn, self.transport)
-        super(ChannelCase, self).setUp()
 
     def tearDown(self):
         self.patch_qpidtoollibs.stop()
-        super(ChannelCase, self).tearDown()
 
 
-class test_Channel__purge(ChannelCase):
+@case_no_python3
+@case_no_pypy
+class TestChannelPurge(ChannelTestBase):
 
-    def setup(self):
-        self.queue = Mock(name='queue')
+    def setUp(self):
+        super(TestChannelPurge, self).setUp()
+        self.mock_queue = Mock()
 
     def test_gets_queue(self):
-        self.channel._purge(self.queue)
-        getQueue = self.broker_agent.return_value.getQueue
-        getQueue.assert_called_once_with(self.queue)
+        self.channel._purge(self.mock_queue)
+        getQueue = self.mock_broker_agent.return_value.getQueue
+        getQueue.assert_called_once_with(self.mock_queue)
 
     def test_does_not_call_purge_if_message_count_is_zero(self):
         values = {'msgDepth': 0}
-        queue_obj = self.broker_agent.return_value.getQueue.return_value
+        queue_obj = self.mock_broker_agent.return_value.getQueue.return_value
         queue_obj.values = values
-        self.channel._purge(self.queue)
-        queue_obj.purge.assert_not_called()
+        self.channel._purge(self.mock_queue)
+        self.assertFalse(queue_obj.purge.called)
 
     def test_purges_all_messages_from_queue(self):
         values = {'msgDepth': 5}
-        queue_obj = self.broker_agent.return_value.getQueue.return_value
+        queue_obj = self.mock_broker_agent.return_value.getQueue.return_value
         queue_obj.values = values
-        self.channel._purge(self.queue)
+        self.channel._purge(self.mock_queue)
         queue_obj.purge.assert_called_with(5)
 
     def test_returns_message_count(self):
         values = {'msgDepth': 5}
-        queue_obj = self.broker_agent.return_value.getQueue.return_value
+        queue_obj = self.mock_broker_agent.return_value.getQueue.return_value
         queue_obj.values = values
-        result = self.channel._purge(self.queue)
+        result = self.channel._purge(self.mock_queue)
         self.assertEqual(result, 5)
 
+    @patch(QPID_MODULE + '.NotFound', new=QpidException)
+    def test_raises_channel_error_if_queue_does_not_exist(self):
+        self.mock_broker_agent.return_value.getQueue.return_value = None
+        self.assertRaises(QpidException, self.channel._purge, self.mock_queue)
+
 
-class test_Channel__put(ChannelCase):
+@case_no_python3
+@case_no_pypy
+class TestChannelPut(ChannelTestBase):
 
     @patch(QPID_MODULE + '.qpid')
-    def test_onto_queue(self, qpid_module):
+    def test_channel__put_onto_queue(self, mock_qpid):
         routing_key = 'routingkey'
-        message = Mock(name='message')
-        Message_cls = qpid_module.messaging.Message
+        mock_message = Mock()
+        mock_Message_cls = mock_qpid.messaging.Message
 
-        self.channel._put(routing_key, message)
+        self.channel._put(routing_key, mock_message)
 
-        address_string = '%s; {assert: always, node: {type: queue}}' % (
+        address_str = '{0}; {{assert: always, node: {{type: queue}}}}'.format(
             routing_key,
         )
-        self.transport.session.sender.assert_called_with(address_string)
-        Message_cls.assert_called_with(content=message, subject=None)
-        sender = self.transport.session.sender.return_value
-        sender.send.assert_called_with(Message_cls.return_value, sync=True)
-        sender.close.assert_called_with()
+        self.transport.session.sender.assert_called_with(address_str)
+        mock_Message_cls.assert_called_with(
+            content=mock_message, subject=None,
+        )
+        mock_sender = self.transport.session.sender.return_value
+        mock_sender.send.assert_called_with(
+            mock_Message_cls.return_value, sync=True,
+        )
+        mock_sender.close.assert_called_with()
 
     @patch(QPID_MODULE + '.qpid')
-    def test_onto_exchange(self, qpid_module):
-        routing_key = 'routingkey'
-        exchange_name = 'myexchange'
-        message = Mock(name='message')
-        Message_cls = qpid_module.messaging.Message
+    def test_channel__put_onto_exchange(self, mock_qpid):
+        mock_routing_key = 'routingkey'
+        mock_exchange_name = 'myexchange'
+        mock_message = Mock()
+        mock_Message_cls = mock_qpid.messaging.Message
 
-        self.channel._put(routing_key, message, exchange_name)
+        self.channel._put(mock_routing_key, mock_message, mock_exchange_name)
 
-        address_string = '%s/%s; {assert: always, node: {type: topic}}' % (
-            exchange_name, routing_key,
+        addrstr = '{0}/{1}; {{assert: always, node: {{type: topic}}}}'.format(
+            mock_exchange_name, mock_routing_key,
+        )
+        self.transport.session.sender.assert_called_with(addrstr)
+        mock_Message_cls.assert_called_with(
+            content=mock_message, subject=mock_routing_key,
+        )
+        mock_sender = self.transport.session.sender.return_value
+        mock_sender.send.assert_called_with(
+            mock_Message_cls.return_value, sync=True,
         )
-        self.transport.session.sender.assert_called_with(address_string)
-        Message_cls.assert_called_with(content=message, subject=routing_key)
-        sender = self.transport.session.sender.return_value
-        sender.send.assert_called_with(Message_cls.return_value, sync=True)
-        sender.close.assert_called_with()
+        mock_sender.close.assert_called_with()
 
 
-class test_Channel__get(ChannelCase):
+@case_no_python3
+@case_no_pypy
+class TestChannelGet(ChannelTestBase):
 
-    def test_get(self):
-        queue = Mock(name='queue')
+    def test_channel__get(self):
+        mock_queue = Mock()
 
-        result = self.channel._get(queue)
+        result = self.channel._get(mock_queue)
 
-        self.transport.session.receiver.assert_called_once_with(queue)
-        rx = self.transport.session.receiver.return_value
-        rx.fetch.assert_called_once_with(timeout=0)
-        rx.close.assert_called_once_with()
-        self.assertIs(rx.fetch.return_value, result)
+        self.transport.session.receiver.assert_called_once_with(mock_queue)
+        mock_rx = self.transport.session.receiver.return_value
+        mock_rx.fetch.assert_called_once_with(timeout=0)
+        mock_rx.close.assert_called_once_with()
+        self.assertIs(mock_rx.fetch.return_value, result)
 
 
-class test_Channel__close(ChannelCase):
+@case_no_python3
+@case_no_pypy
+class TestChannelClose(ChannelTestBase):
 
-    def setup(self):
+    def setUp(self):
+        super(TestChannelClose, self).setUp()
         self.patch_basic_cancel = patch.object(self.channel, 'basic_cancel')
-        self.basic_cancel = self.patch_basic_cancel.start()
-        self.receiver1 = Mock(name='receiver1')
-        self.receiver2 = Mock(name='receiver2')
-        self.channel._receivers = {1: self.receiver1,
-                                   2: self.receiver2}
+        self.mock_basic_cancel = self.patch_basic_cancel.start()
+        self.mock_receiver1 = Mock()
+        self.mock_receiver2 = Mock()
+        self.channel._receivers = {
+            1: self.mock_receiver1, 2: self.mock_receiver2,
+        }
         self.channel.closed = False
 
-    def teardown(self):
+    def tearDown(self):
         self.patch_basic_cancel.stop()
+        super(TestChannelClose, self).tearDown()
 
-    def test_sets_close_attribute(self):
+    def test_channel_close_sets_close_attribute(self):
         self.channel.close()
         self.assertTrue(self.channel.closed)
 
-    def test_calls_basic_cancel_on_all_receivers(self):
+    def test_channel_close_calls_basic_cancel_on_all_receivers(self):
         self.channel.close()
-        self.basic_cancel.assert_has_calls([call(1), call(2)])
+        self.mock_basic_cancel.assert_has_calls([call(1), call(2)])
 
-    def test_calls_close_channel_on_connection(self):
+    def test_channel_close_calls_close_channel_on_connection(self):
         self.channel.close()
         self.conn.close_channel.assert_called_once_with(self.channel)
 
-    def test_calls_close_on_broker_agent(self):
+    def test_channel_close_calls_close_on_broker_agent(self):
         self.channel.close()
         self.channel._broker.close.assert_called_once_with()
 
-    def test_does_nothing_if_already_closed(self):
+    def test_channel_close_does_nothing_if_already_closed(self):
         self.channel.closed = True
         self.channel.close()
-        self.basic_cancel.assert_not_called()
+        self.assertFalse(self.mock_basic_cancel.called)
 
-    def test_does_not_call_close_channel_if_conn_is_None(self):
+    def test_channel_close_does_not_call_close_channel_if_conn_is_None(self):
         self.channel.connection = None
         self.channel.close()
-        self.conn.close_channel.assert_not_called()
+        self.assertFalse(self.conn.close_channel.called)
 
 
-class test_Channel__basic_qos(ChannelCase):
+@case_no_python3
+@case_no_pypy
+class TestChannelBasicQoS(ChannelTestBase):
 
-    def test_always_returns_one(self):
+    def test_channel_basic_qos_always_returns_one(self):
         self.channel.basic_qos(2)
         self.assertEqual(self.channel.qos.prefetch_count, 1)
 
 
-class test_Channel_basic_get(ChannelCase):
-
-    def setup(self):
-        self.channel.Message = Mock(name='channel.Message')
-        self.channel._get = Mock(name='channel._get')
-
-    def test_calls__get_with_queue(self):
-        queue = Mock(name='queue')
-        self.channel.basic_get(queue)
-        self.channel._get.assert_called_once_with(queue)
+@case_no_python3
+@case_no_pypy
+class TestChannelBasicGet(ChannelTestBase):
 
-    def test_creates_Message_correctly(self):
-        queue = Mock(name='queue')
-        self.channel.basic_get(queue)
-        raw_message = self.channel._get.return_value.content
-        self.channel.Message.assert_called_once_with(self.channel,
-                                                     raw_message)
+    def setUp(self):
+        super(TestChannelBasicGet, self).setUp()
+        self.channel.Message = Mock()
+        self.channel._get = Mock()
+
+    def test_channel_basic_get_calls__get_with_queue(self):
+        mock_queue = Mock()
+        self.channel.basic_get(mock_queue)
+        self.channel._get.assert_called_once_with(mock_queue)
+
+    def test_channel_basic_get_creates_Message_correctly(self):
+        mock_queue = Mock()
+        self.channel.basic_get(mock_queue)
+        mock_raw_message = self.channel._get.return_value.content
+        self.channel.Message.assert_called_once_with(
+            self.channel, mock_raw_message,
+        )
 
-    def test_acknowledges_message_by_default(self):
-        queue = Mock(name='queue')
-        self.channel.basic_get(queue)
-        qpid_message = self.channel._get.return_value
+    def test_channel_basic_get_acknowledges_message_by_default(self):
+        mock_queue = Mock()
+        self.channel.basic_get(mock_queue)
+        mock_qpid_message = self.channel._get.return_value
         acknowledge = self.transport.session.acknowledge
-        acknowledge.assert_called_once_with(message=qpid_message)
+        acknowledge.assert_called_once_with(message=mock_qpid_message)
 
-    def test_acknowledges_message_with_no_ack_False(self):
-        queue = Mock(name='queue')
-        self.channel.basic_get(queue, no_ack=False)
-        qpid_message = self.channel._get.return_value
+    def test_channel_basic_get_acknowledges_message_with_no_ack_False(self):
+        mock_queue = Mock()
+        self.channel.basic_get(mock_queue, no_ack=False)
+        mock_qpid_message = self.channel._get.return_value
         acknowledge = self.transport.session.acknowledge
-        acknowledge.assert_called_once_with(message=qpid_message)
+        acknowledge.assert_called_once_with(message=mock_qpid_message)
 
-    def test_get_acknowledges_message_with_no_ack_True(self):
-        queue = Mock(name='queue')
-        self.channel.basic_get(queue, no_ack=True)
-        qpid_message = self.channel._get.return_value
+    def test_channel_basic_get_acknowledges_message_with_no_ack_True(self):
+        mock_queue = Mock()
+        self.channel.basic_get(mock_queue, no_ack=True)
+        mock_qpid_message = self.channel._get.return_value
         acknowledge = self.transport.session.acknowledge
-        acknowledge.assert_called_once_with(message=qpid_message)
+        acknowledge.assert_called_once_with(message=mock_qpid_message)
 
-    def test_get_returns_correct_message(self):
-        queue = Mock(name='queue')
-        basic_get_result = self.channel.basic_get(queue)
+    def test_channel_basic_get_returns_correct_message(self):
+        mock_queue = Mock()
+        basic_get_result = self.channel.basic_get(mock_queue)
         expected_message = self.channel.Message.return_value
         self.assertIs(expected_message, basic_get_result)
 
-    def test_returns_None_when_channel__get_raises_Empty(self):
-        queue = Mock(name='queue')
-        self.channel._get = Mock(
-            name='channel._get', side_effect=Empty,
-        )
-        basic_get_result = self.channel.basic_get(queue)
+    def test_basic_get_returns_None_when_channel__get_raises_Empty(self):
+        mock_queue = Mock()
+        self.channel._get = Mock(side_effect=Empty)
+        basic_get_result = self.channel.basic_get(mock_queue)
         self.assertEqual(self.channel.Message.call_count, 0)
         self.assertIsNone(basic_get_result)
 
 
-class test_Channel__basic_cancel(ChannelCase):
+@case_no_python3
+@case_no_pypy
+class TestChannelBasicCancel(ChannelTestBase):
 
-    def setup(self):
+    def setUp(self):
+        super(TestChannelBasicCancel, self).setUp()
         self.channel._receivers = {1: Mock()}
 
-    def test_no_error_if_consumer_tag_not_found(self):
+    def test_channel_basic_cancel_no_error_if_consumer_tag_not_found(self):
         self.channel.basic_cancel(2)
 
-    def test_pops_receiver(self):
+    def test_channel_basic_cancel_pops_receiver(self):
         self.channel.basic_cancel(1)
         self.assertNotIn(1, self.channel._receivers)
 
-    def test_closes_receiver(self):
-        receiver = self.channel._receivers[1]
+    def test_channel_basic_cancel_closes_receiver(self):
+        mock_receiver = self.channel._receivers[1]
         self.channel.basic_cancel(1)
-        receiver.close.assert_called_once_with()
+        mock_receiver.close.assert_called_once_with()
 
-    def test_pops__tag_to_queue(self):
-        self.channel._tag_to_queue = Mock(name='_tag_to_queue')
+    def test_channel_basic_cancel_pops__tag_to_queue(self):
+        self.channel._tag_to_queue = Mock()
         self.channel.basic_cancel(1)
         self.channel._tag_to_queue.pop.assert_called_once_with(1, None)
 
-    def test_pops_connection__callbacks(self):
-        self.channel._tag_to_queue = Mock(name='_tag_to_queue')
+    def test_channel_basic_cancel_pops_connection__callbacks(self):
+        self.channel._tag_to_queue = Mock()
         self.channel.basic_cancel(1)
-        queue = self.channel._tag_to_queue.pop.return_value
-        self.conn._callbacks.pop.assert_called_once_with(queue, None)
+        mock_queue = self.channel._tag_to_queue.pop.return_value
+        self.conn._callbacks.pop.assert_called_once_with(mock_queue, None)
 
 
-class test_Channel__init__(ChannelCase):
+@case_no_python3
+@case_no_pypy
+class TestChannelInit(ChannelTestBase, ExtraAssertionsMixin):
 
-    def test_sets_variables_as_expected(self):
+    def test_channel___init__sets_variables_as_expected(self):
         self.assertIs(self.conn, self.channel.connection)
         self.assertIs(self.transport, self.channel.transport)
         self.assertFalse(self.channel.closed)
         self.conn.get_qpid_connection.assert_called_once_with()
-        expected_broker_agent = self.broker_agent.return_value
+        expected_broker_agent = self.mock_broker_agent.return_value
         self.assertIs(self.channel._broker, expected_broker_agent)
         self.assertDictEqual(self.channel._tag_to_queue, {})
         self.assertDictEqual(self.channel._receivers, {})
-        self.assertIsNone(self.channel._qos)
+        self.assertIs(self.channel._qos, None)
 
 
-class test_Channel__basic_consume(ChannelCase):
+@case_no_python3
+@case_no_pypy
+class TestChannelBasicConsume(ChannelTestBase, ExtraAssertionsMixin):
 
-    def setup(self):
+    def setUp(self):
+        super(TestChannelBasicConsume, self).setUp()
         self.conn._callbacks = {}
 
-    def test_adds_queue_to__tag_to_queue(self):
-        tag = Mock(name='tag')
-        queue = Mock(name='queue')
-        self.channel.basic_consume(queue, Mock(), Mock(), tag)
-        expected_dict = {tag: queue}
+    def test_channel_basic_consume_adds_queue_to__tag_to_queue(self):
+        mock_tag = Mock()
+        mock_queue = Mock()
+        self.channel.basic_consume(mock_queue, Mock(), Mock(), mock_tag)
+        expected_dict = {mock_tag: mock_queue}
         self.assertDictEqual(expected_dict, self.channel._tag_to_queue)
 
-    def test_adds_entry_to_connection__callbacks(self):
-        queue = Mock(name='queue')
-        self.channel.basic_consume(queue, Mock(), Mock(), Mock())
-        self.assertIn(queue, self.conn._callbacks)
-        if not hasattr(self.conn._callbacks[queue], '__call__'):
-            self.fail('Callback stored must be callable')
-
-    def test_creates_new_receiver(self):
-        queue = Mock(name='queue')
-        self.channel.basic_consume(queue, Mock(), Mock(), Mock())
-        self.transport.session.receiver.assert_called_once_with(queue)
-
-    def test_saves_new_receiver(self):
-        tag = Mock(name='tag')
-        self.channel.basic_consume(Mock(), Mock(), Mock(), tag)
-        new_receiver = self.transport.session.receiver.return_value
-        expected_dict = {tag: new_receiver}
+    def test_channel_basic_consume_adds_entry_to_connection__callbacks(self):
+        mock_queue = Mock()
+        self.channel.basic_consume(mock_queue, Mock(), Mock(), Mock())
+        self.assertIn(mock_queue, self.conn._callbacks)
+        self.assertIsInstance(self.conn._callbacks[mock_queue], Callable)
+
+    def test_channel_basic_consume_creates_new_receiver(self):
+        mock_queue = Mock()
+        self.channel.basic_consume(mock_queue, Mock(), Mock(), Mock())
+        self.transport.session.receiver.assert_called_once_with(mock_queue)
+
+    def test_channel_basic_consume_saves_new_receiver(self):
+        mock_tag = Mock()
+        self.channel.basic_consume(Mock(), Mock(), Mock(), mock_tag)
+        new_mock_receiver = self.transport.session.receiver.return_value
+        expected_dict = {mock_tag: new_mock_receiver}
         self.assertDictEqual(expected_dict, self.channel._receivers)
 
-    def test_sets_capacity_on_new_receiver(self):
-        prefetch_count = Mock(name='prefetch_count')
-        self.channel.qos.prefetch_count = prefetch_count
+    def test_channel_basic_consume_sets_capacity_on_new_receiver(self):
+        mock_prefetch_count = Mock()
+        self.channel.qos.prefetch_count = mock_prefetch_count
         self.channel.basic_consume(Mock(), Mock(), Mock(), Mock())
         new_receiver = self.transport.session.receiver.return_value
-        self.assertIs(new_receiver.capacity, prefetch_count)
+        self.assertTrue(new_receiver.capacity is mock_prefetch_count)
 
     def get_callback(self, no_ack=Mock(), original_cb=Mock()):
-        self.channel.Message = Mock(name='Message')
-        queue = Mock(name='queue')
-        self.channel.basic_consume(queue, no_ack, original_cb, Mock())
-        return self.conn._callbacks[queue]
+        self.channel.Message = Mock()
+        mock_queue = Mock()
+        self.channel.basic_consume(mock_queue, no_ack, original_cb, Mock())
+        return self.conn._callbacks[mock_queue]
 
-    def test_callback_creates_Message_correctly(self):
+    def test_channel_basic_consume_callback_creates_Message_correctly(self):
         callback = self.get_callback()
-        qpid_message = Mock(name='qpid_message')
-        callback(qpid_message)
-        content = qpid_message.content
-        self.channel.Message.assert_called_once_with(self.channel,
-                                                     content)
-
-    def test_callback_adds_message_to_QoS(self):
-        self.channel._qos = Mock(name='_qos')
+        mock_qpid_message = Mock()
+        callback(mock_qpid_message)
+        mock_content = mock_qpid_message.content
+        self.channel.Message.assert_called_once_with(
+            self.channel, mock_content,
+        )
+
+    def test_channel_basic_consume_callback_adds_message_to_QoS(self):
+        self.channel._qos = Mock()
         callback = self.get_callback()
-        qpid_message = Mock(name='qpid_message')
-        callback(qpid_message)
-        delivery_tag = self.channel.Message.return_value.delivery_tag
-        self.channel._qos.append.assert_called_once_with(qpid_message,
-                                                         delivery_tag)
-
-    def test_callback_gratuitously_acks(self):
-        self.channel.basic_ack = Mock(name='basic_ack')
+        mock_qpid_message = Mock()
+        callback(mock_qpid_message)
+        mock_delivery_tag = self.channel.Message.return_value.delivery_tag
+        self.channel._qos.append.assert_called_once_with(
+            mock_qpid_message, mock_delivery_tag,
+        )
+
+    def test_channel_basic_consume_callback_gratuitously_acks(self):
+        self.channel.basic_ack = Mock()
         callback = self.get_callback()
-        qpid_message = Mock(name='qpid_message')
-        callback(qpid_message)
-        delivery_tag = self.channel.Message.return_value.delivery_tag
-        self.channel.basic_ack.assert_called_once_with(delivery_tag)
+        mock_qpid_message = Mock()
+        callback(mock_qpid_message)
+        mock_delivery_tag = self.channel.Message.return_value.delivery_tag
+        self.channel.basic_ack.assert_called_once_with(mock_delivery_tag)
 
-    def test_callback_does_not_ack_when_needed(self):
-        self.channel.basic_ack = Mock(name='basic_ack')
+    def test_channel_basic_consume_callback_does_not_ack_when_needed(self):
+        self.channel.basic_ack = Mock()
         callback = self.get_callback(no_ack=False)
-        qpid_message = Mock(name='qpid_message')
-        callback(qpid_message)
-        self.channel.basic_ack.assert_not_called()
-
-    def test_callback_calls_real_callback(self):
-        self.channel.basic_ack = Mock(name='basic_ack')
-        original_callback = Mock(name='original_callback')
-        callback = self.get_callback(original_cb=original_callback)
-        qpid_message = Mock(name='qpid_message')
-        callback(qpid_message)
+        mock_qpid_message = Mock()
+        callback(mock_qpid_message)
+        self.assertFalse(self.channel.basic_ack.called)
+
+    def test_channel_basic_consume_callback_calls_real_callback(self):
+        self.channel.basic_ack = Mock()
+        mock_original_callback = Mock()
+        callback = self.get_callback(original_cb=mock_original_callback)
+        mock_qpid_message = Mock()
+        callback(mock_qpid_message)
         expected_message = self.channel.Message.return_value
-        original_callback.assert_called_once_with(expected_message)
+        mock_original_callback.assert_called_once_with(expected_message)
 
 
-class test_Channel__queue_delete(ChannelCase):
+@case_no_python3
+@case_no_pypy
+class TestChannelQueueDelete(ChannelTestBase):
 
-    def setup(self):
-        has_queue_patcher = patch.object(self.channel, '_has_queue')
-        self.has_queue = has_queue_patcher.start()
-        self.addCleanup(has_queue_patcher.stop)
-
-        size_patcher = patch.object(self.channel, '_size')
-        self.size = size_patcher.start()
-        self.addCleanup(size_patcher.stop)
-
-        delete_patcher = patch.object(self.channel, '_delete')
-        self.delete = delete_patcher.start()
-        self.addCleanup(delete_patcher.stop)
+    def setUp(self):
+        super(TestChannelQueueDelete, self).setUp()
+        self.patch__has_queue = patch.object(self.channel, '_has_queue')
+        self.mock__has_queue = self.patch__has_queue.start()
+        self.patch__size = patch.object(self.channel, '_size')
+        self.mock__size = self.patch__size.start()
+        self.patch__delete = patch.object(self.channel, '_delete')
+        self.mock__delete = self.patch__delete.start()
+        self.mock_queue = Mock()
 
-        self.queue = Mock(name='queue')
+    def tearDown(self):
+        self.patch__has_queue.stop()
+        self.patch__size.stop()
+        self.patch__delete.stop()
+        super(TestChannelQueueDelete, self).tearDown()
 
     def test_checks_if_queue_exists(self):
-        self.channel.queue_delete(self.queue)
-        self.has_queue.assert_called_once_with(self.queue)
+        self.channel.queue_delete(self.mock_queue)
+        self.mock__has_queue.assert_called_once_with(self.mock_queue)
 
     def test_does_nothing_if_queue_does_not_exist(self):
-        self.has_queue.return_value = False
-        self.channel.queue_delete(self.queue)
-        self.delete.assert_not_called()
+        self.mock__has_queue.return_value = False
+        self.channel.queue_delete(self.mock_queue)
+        self.assertFalse(self.mock__delete.called)
 
     def test_not_empty_and_if_empty_True_no_delete(self):
-        self.size.return_value = 1
-        self.channel.queue_delete(self.queue, if_empty=True)
-        broker = self.broker_agent.return_value
-        broker.getQueue.assert_not_called()
+        self.mock__size.return_value = 1
+        self.channel.queue_delete(self.mock_queue, if_empty=True)
+        mock_broker = self.mock_broker_agent.return_value
+        self.assertFalse(mock_broker.getQueue.called)
 
     def test_calls_get_queue(self):
-        self.channel.queue_delete(self.queue)
-        getQueue = self.broker_agent.return_value.getQueue
-        getQueue.assert_called_once_with(self.queue)
+        self.channel.queue_delete(self.mock_queue)
+        getQueue = self.mock_broker_agent.return_value.getQueue
+        getQueue.assert_called_once_with(self.mock_queue)
 
     def test_gets_queue_attribute(self):
-        self.channel.queue_delete(self.queue)
-        queue_obj = self.broker_agent.return_value.getQueue.return_value
+        self.channel.queue_delete(self.mock_queue)
+        queue_obj = self.mock_broker_agent.return_value.getQueue.return_value
         queue_obj.getAttributes.assert_called_once_with()
 
     def test_queue_in_use_and_if_unused_no_delete(self):
-        queue_obj = self.broker_agent.return_value.getQueue.return_value
+        queue_obj = self.mock_broker_agent.return_value.getQueue.return_value
         queue_obj.getAttributes.return_value = {'consumerCount': 1}
-        self.channel.queue_delete(self.queue, if_unused=True)
-        self.delete.assert_not_called()
+        self.channel.queue_delete(self.mock_queue, if_unused=True)
+        self.assertFalse(self.mock__delete.called)
 
     def test_calls__delete_with_queue(self):
-        self.channel.queue_delete(self.queue)
-        self.delete.assert_called_once_with(self.queue)
+        self.channel.queue_delete(self.mock_queue)
+        self.mock__delete.assert_called_once_with(self.mock_queue)
 
 
-class test_Channel(QPidCase):
+@case_no_python3
+@case_no_pypy
+class TestChannel(ExtraAssertionsMixin, Case):
 
     @patch(QPID_MODULE + '.qpidtoollibs')
-    def setup(self, qpidtoollibs):
-        self.connection = Mock(name='connection')
-        self.qpid_connection = Mock(name='qpid_connection')
-        self.qpid_session = Mock(name='qpid_session')
-        self.qpid_connection.session = Mock(
-            name='session',
-            return_value=self.qpid_session,
+    def setUp(self, mock_qpidtoollibs):
+        self.mock_connection = Mock()
+        self.mock_qpid_connection = Mock()
+        self.mock_qpid_session = Mock()
+        self.mock_qpid_connection.session = Mock(
+            return_value=self.mock_qpid_session,
+        )
+        self.mock_connection.get_qpid_connection = Mock(
+            return_value=self.mock_qpid_connection,
         )
-        self.connection.get_qpid_connection = Mock(
-            name='get_qpid_connection',
-            return_value=self.qpid_connection,
+        self.mock_transport = Mock()
+        self.mock_broker = Mock()
+        self.mock_Message = Mock()
+        self.mock_BrokerAgent = mock_qpidtoollibs.BrokerAgent
+        self.mock_BrokerAgent.return_value = self.mock_broker
+        self.my_channel = Channel(
+            self.mock_connection, self.mock_transport,
         )
-        self.transport = Mock(name='transport')
-        self.broker = Mock(name='broker')
-        self.Message = Mock(name='Message')
-        self.BrokerAgent = qpidtoollibs.BrokerAgent
-        self.BrokerAgent.return_value = self.broker
-        self.my_channel = Channel(self.connection,
-                                  self.transport)
-        self.my_channel.Message = self.Message
+        self.my_channel.Message = self.mock_Message
 
     def test_verify_QoS_class_attribute(self):
-        # Verify that the class attribute QoS refers to the QoS object
+        """Verify that the class attribute QoS refers to the QoS object"""
         self.assertIs(QoS, Channel.QoS)
 
     def test_verify_Message_class_attribute(self):
-        # Verify that the class attribute Message refers
-        # to the Message object
+        """Verify that the class attribute Message refers to the Message
+        object."""
         self.assertIs(Message, Channel.Message)
 
     def test_body_encoding_class_attribute(self):
-        # Verify that the class attribute body_encoding is set to base64
+        """Verify that the class attribute body_encoding is set to base64"""
         self.assertEqual('base64', Channel.body_encoding)
 
     def test_codecs_class_attribute(self):
-        # Verify that the codecs class attribute has a correct key and
-        # value
+        """Verify that the codecs class attribute has a correct key and
+        value."""
         self.assertIsInstance(Channel.codecs, dict)
         self.assertIn('base64', Channel.codecs)
         self.assertIsInstance(Channel.codecs['base64'], Base64)
 
-    def test_delivery_tags(self):
-        # Test that _delivery_tags is using itertools
-        self.assertTrue(isinstance(Channel._delivery_tags, count))
-
     def test_size(self):
-        # Test getting the number of messages in a queue specified by
-        # name and returning them.
+        """Test getting the number of messages in a queue specified by
+        name and returning them."""
         message_count = 5
-        queue = Mock(name='queue')
-        queue_to_check = Mock(name='queue_to_check')
-        queue_to_check.values = {'msgDepth': message_count}
-        self.broker.getQueue.return_value = queue_to_check
-        result = self.my_channel._size(queue)
-        self.broker.getQueue.assert_called_with(queue)
+        mock_queue = Mock()
+        mock_queue_to_check = Mock()
+        mock_queue_to_check.values = {'msgDepth': message_count}
+        self.mock_broker.getQueue.return_value = mock_queue_to_check
+        result = self.my_channel._size(mock_queue)
+        self.mock_broker.getQueue.assert_called_with(mock_queue)
         self.assertEqual(message_count, result)
 
     def test_delete(self):
-        # Test deleting a queue calls purge and delQueue with queue name
-        queue = Mock(name='queue')
-        self.my_channel._purge = Mock(name='_purge')
-        result = self.my_channel._delete(queue)
-        self.my_channel._purge.assert_called_with(queue)
-        self.broker.delQueue.assert_called_with(queue)
+        """Test deleting a queue calls purge and delQueue with queue name."""
+        mock_queue = Mock()
+        self.my_channel._purge = Mock()
+        result = self.my_channel._delete(mock_queue)
+        self.my_channel._purge.assert_called_with(mock_queue)
+        self.mock_broker.delQueue.assert_called_with(mock_queue)
         self.assertIsNone(result)
 
     def test_has_queue_true(self):
-        # Test checking if a queue exists, and it does
-        queue = Mock(name='queue')
-        self.broker.getQueue.return_value = True
-        result = self.my_channel._has_queue(queue)
+        """Test checking if a queue exists, and it does."""
+        mock_queue = Mock()
+        self.mock_broker.getQueue.return_value = True
+        result = self.my_channel._has_queue(mock_queue)
         self.assertTrue(result)
 
     def test_has_queue_false(self):
-        # Test checking if a queue exists, and it does not
-        queue = Mock(name='queue')
-        self.broker.getQueue.return_value = False
-        result = self.my_channel._has_queue(queue)
+        """Test checking if a queue exists, and it does not."""
+        mock_queue = Mock()
+        self.mock_broker.getQueue.return_value = False
+        result = self.my_channel._has_queue(mock_queue)
         self.assertFalse(result)
 
     @patch('amqp.protocol.queue_declare_ok_t')
     def test_queue_declare_with_exception_raised(self,
-                                                 queue_declare_ok_t):
-        # Test declare_queue, where an exception is raised and silenced
-        queue = Mock(name='queue')
-        passive = Mock(name='passive')
-        durable = Mock(name='durable')
-        exclusive = Mock(name='exclusive')
-        auto_delete = Mock(name='auto_delete')
-        nowait = Mock(name='nowait')
-        arguments = Mock(name='arguments')
-        msg_count = Mock(name='msg_count')
-        queue.startswith.return_value = False
-        queue.endswith.return_value = False
+                                                 mock_queue_declare_ok_t):
+        """Test declare_queue, where an exception is raised and silenced."""
+        mock_queue = Mock()
+        mock_passive = Mock()
+        mock_durable = Mock()
+        mock_exclusive = Mock()
+        mock_auto_delete = Mock()
+        mock_nowait = Mock()
+        mock_arguments = Mock()
+        mock_msg_count = Mock()
+        mock_queue.startswith.return_value = False
+        mock_queue.endswith.return_value = False
         options = {
-            'passive': passive,
-            'durable': durable,
-            'exclusive': exclusive,
-            'auto-delete': auto_delete,
-            'arguments': arguments,
+            'passive': mock_passive,
+            'durable': mock_durable,
+            'exclusive': mock_exclusive,
+            'auto-delete': mock_auto_delete,
+            'arguments': mock_arguments,
         }
-        consumer_count = Mock(name='consumer_count')
-        expected_return_value = Mock(name='expected_return_value')
+        mock_consumer_count = Mock()
+        mock_return_value = Mock()
         values_dict = {
-            'msgDepth': msg_count,
-            'consumerCount': consumer_count,
+            'msgDepth': mock_msg_count,
+            'consumerCount': mock_consumer_count,
         }
-        queue_data = Mock(name='queue_data')
-        queue_data.values = values_dict
+        mock_queue_data = Mock()
+        mock_queue_data.values = values_dict
         exception_to_raise = Exception('The foo object already exists.')
-        self.broker.addQueue.side_effect = exception_to_raise
-        self.broker.getQueue.return_value = queue_data
-        queue_declare_ok_t.return_value = expected_return_value
+        self.mock_broker.addQueue.side_effect = exception_to_raise
+        self.mock_broker.getQueue.return_value = mock_queue_data
+        mock_queue_declare_ok_t.return_value = mock_return_value
         result = self.my_channel.queue_declare(
-            queue,
-            passive=passive,
-            durable=durable,
-            exclusive=exclusive,
-            auto_delete=auto_delete,
-            nowait=nowait,
-            arguments=arguments,
+            mock_queue,
+            passive=mock_passive,
+            durable=mock_durable,
+            exclusive=mock_exclusive,
+            auto_delete=mock_auto_delete,
+            nowait=mock_nowait,
+            arguments=mock_arguments,
         )
-        self.broker.addQueue.assert_called_with(queue, options=options)
-        queue_declare_ok_t.assert_called_with(queue, msg_count, consumer_count)
-        self.assertIs(result, expected_return_value)
+        self.mock_broker.addQueue.assert_called_with(
+            mock_queue, options=options,
+        )
+        mock_queue_declare_ok_t.assert_called_with(
+            mock_queue, mock_msg_count, mock_consumer_count,
+        )
+        self.assertIs(mock_return_value, result)
 
     def test_queue_declare_set_ring_policy_for_celeryev(self):
-        # Test declare_queue sets ring_policy for celeryev.
-        queue = Mock(name='queue')
-        queue.startswith.return_value = True
-        queue.endswith.return_value = False
+        """Test declare_queue sets ring_policy for celeryev."""
+        mock_queue = Mock()
+        mock_queue.startswith.return_value = True
+        mock_queue.endswith.return_value = False
         expected_default_options = {
             'passive': False,
             'durable': False,
@@ -937,27 +1064,27 @@ class test_Channel(QPidCase):
             'arguments': None,
             'qpid.policy_type': 'ring',
         }
-        msg_count = Mock(name='msg_count')
-        consumer_count = Mock(name='consumer_count')
+        mock_msg_count = Mock()
+        mock_consumer_count = Mock()
         values_dict = {
-            'msgDepth': msg_count,
-            'consumerCount': consumer_count,
+            'msgDepth': mock_msg_count,
+            'consumerCount': mock_consumer_count,
         }
-        queue_data = Mock(name='queue_data')
-        queue_data.values = values_dict
-        self.broker.addQueue.return_value = None
-        self.broker.getQueue.return_value = queue_data
-        self.my_channel.queue_declare(queue)
-        queue.startswith.assert_called_with('celeryev')
-        self.broker.addQueue.assert_called_with(
-            queue, options=expected_default_options,
+        mock_queue_data = Mock()
+        mock_queue_data.values = values_dict
+        self.mock_broker.addQueue.return_value = None
+        self.mock_broker.getQueue.return_value = mock_queue_data
+        self.my_channel.queue_declare(mock_queue)
+        mock_queue.startswith.assert_called_with('celeryev')
+        self.mock_broker.addQueue.assert_called_with(
+            mock_queue, options=expected_default_options,
         )
 
     def test_queue_declare_set_ring_policy_for_pidbox(self):
-        # Test declare_queue sets ring_policy for pidbox.
-        queue = Mock(name='queue')
-        queue.startswith.return_value = False
-        queue.endswith.return_value = True
+        """Test declare_queue sets ring_policy for pidbox."""
+        mock_queue = Mock()
+        mock_queue.startswith.return_value = False
+        mock_queue.endswith.return_value = True
         expected_default_options = {
             'passive': False,
             'durable': False,
@@ -966,26 +1093,27 @@ class test_Channel(QPidCase):
             'arguments': None,
             'qpid.policy_type': 'ring',
         }
-        msg_count = Mock(name='msg_count')
-        consumer_count = Mock(name='consumer_count')
+        mock_msg_count = Mock()
+        mock_consumer_count = Mock()
         values_dict = {
-            'msgDepth': msg_count,
-            'consumerCount': consumer_count,
+            'msgDepth': mock_msg_count,
+            'consumerCount': mock_consumer_count,
         }
-        queue_data = Mock(name='queue_data')
-        queue_data.values = values_dict
-        self.broker.addQueue.return_value = None
-        self.broker.getQueue.return_value = queue_data
-        self.my_channel.queue_declare(queue)
-        queue.endswith.assert_called_with('pidbox')
-        self.broker.addQueue.assert_called_with(
-            queue, options=expected_default_options)
+        mock_queue_data = Mock()
+        mock_queue_data.values = values_dict
+        self.mock_broker.addQueue.return_value = None
+        self.mock_broker.getQueue.return_value = mock_queue_data
+        self.my_channel.queue_declare(mock_queue)
+        mock_queue.endswith.assert_called_with('pidbox')
+        self.mock_broker.addQueue.assert_called_with(
+            mock_queue, options=expected_default_options,
+        )
 
     def test_queue_declare_ring_policy_not_set_as_expected(self):
-        # Test declare_queue does not set ring_policy as expected
-        queue = Mock(name='queue')
-        queue.startswith.return_value = False
-        queue.endswith.return_value = False
+        """Test declare_queue does not set ring_policy as expected."""
+        mock_queue = Mock()
+        mock_queue.startswith.return_value = False
+        mock_queue.endswith.return_value = False
         expected_default_options = {
             'passive': False,
             'durable': False,
@@ -993,27 +1121,28 @@ class test_Channel(QPidCase):
             'auto-delete': True,
             'arguments': None,
         }
-        msg_count = Mock(name='msg_count')
-        consumer_count = Mock(name='consumer_count')
+        mock_msg_count = Mock()
+        mock_consumer_count = Mock()
         values_dict = {
-            'msgDepth': msg_count,
-            'consumerCount': consumer_count,
+            'msgDepth': mock_msg_count,
+            'consumerCount': mock_consumer_count,
         }
-        queue_data = Mock(name='queue_data')
-        queue_data.values = values_dict
-        self.broker.addQueue.return_value = None
-        self.broker.getQueue.return_value = queue_data
-        self.my_channel.queue_declare(queue)
-        queue.startswith.assert_called_with('celeryev')
-        queue.endswith.assert_called_with('pidbox')
-        self.broker.addQueue.assert_called_with(
-            queue, options=expected_default_options)
+        mock_queue_data = Mock()
+        mock_queue_data.values = values_dict
+        self.mock_broker.addQueue.return_value = None
+        self.mock_broker.getQueue.return_value = mock_queue_data
+        self.my_channel.queue_declare(mock_queue)
+        mock_queue.startswith.assert_called_with('celeryev')
+        mock_queue.endswith.assert_called_with('pidbox')
+        self.mock_broker.addQueue.assert_called_with(
+            mock_queue, options=expected_default_options,
+        )
 
     def test_queue_declare_test_defaults(self):
-        # Test declare_queue defaults
-        queue = Mock(name='queue')
-        queue.startswith.return_value = False
-        queue.endswith.return_value = False
+        """Test declare_queue defaults."""
+        mock_queue = Mock()
+        mock_queue.startswith.return_value = False
+        mock_queue.endswith.return_value = False
         expected_default_options = {
             'passive': False,
             'durable': False,
@@ -1021,511 +1150,382 @@ class test_Channel(QPidCase):
             'auto-delete': True,
             'arguments': None,
         }
-        msg_count = Mock(name='msg_count')
-        consumer_count = Mock(name='consumer_count')
+        mock_msg_count = Mock()
+        mock_consumer_count = Mock()
         values_dict = {
-            'msgDepth': msg_count,
-            'consumerCount': consumer_count,
+            'msgDepth': mock_msg_count,
+            'consumerCount': mock_consumer_count,
         }
-        queue_data = Mock(name='queue_data')
-        queue_data.values = values_dict
-        self.broker.addQueue.return_value = None
-        self.broker.getQueue.return_value = queue_data
-        self.my_channel.queue_declare(queue)
-        self.broker.addQueue.assert_called_with(
-            queue, options=expected_default_options)
+        mock_queue_data = Mock()
+        mock_queue_data.values = values_dict
+        self.mock_broker.addQueue.return_value = None
+        self.mock_broker.getQueue.return_value = mock_queue_data
+        self.my_channel.queue_declare(mock_queue)
+        self.mock_broker.addQueue.assert_called_with(
+            mock_queue,
+            options=expected_default_options,
+        )
 
     def test_queue_declare_raises_exception_not_silenced(self):
         unique_exception = Exception('This exception should not be silenced')
-        queue = Mock(name='queue')
-        self.broker.addQueue.side_effect = unique_exception
+        mock_queue = Mock()
+        self.mock_broker.addQueue.side_effect = unique_exception
         with self.assertRaises(unique_exception.__class__):
-            self.my_channel.queue_declare(queue)
-        self.broker.addQueue.assert_called_once_with(queue, options={
-            'exclusive': False,
-            'durable': False,
-            'qpid.policy_type': 'ring',
-            'passive': False,
-            'arguments': None,
-            'auto-delete': True,
-        })
+            self.my_channel.queue_declare(mock_queue)
+        self.mock_broker.addQueue.assert_called_once_with(
+            mock_queue,
+            options={
+                'exclusive': False,
+                'durable': False,
+                'qpid.policy_type': 'ring',
+                'passive': False,
+                'arguments': None,
+                'auto-delete': True
+            })
 
     def test_exchange_declare_raises_exception_and_silenced(self):
-        # Create exchange where an exception is raised and then silenced.
-        self.broker.addExchange.side_effect = Exception(
+        """Create exchange where an exception is raised and then silenced"""
+        self.mock_broker.addExchange.side_effect = Exception(
             'The foo object already exists.',
         )
         self.my_channel.exchange_declare()
 
     def test_exchange_declare_raises_exception_not_silenced(self):
-        # Create Exchange where an exception is raised and not silenced
+        """Create Exchange where an exception is raised and not silenced."""
         unique_exception = Exception('This exception should not be silenced')
-        self.broker.addExchange.side_effect = unique_exception
+        self.mock_broker.addExchange.side_effect = unique_exception
         with self.assertRaises(unique_exception.__class__):
             self.my_channel.exchange_declare()
 
     def test_exchange_declare(self):
-        # Create Exchange where an exception is NOT raised.
-        exchange = Mock(name='exchange')
-        exchange_type = Mock(name='exchange_type')
-        durable = Mock(name='durable')
-        options = {'durable': durable}
+        """Create Exchange where an exception is NOT raised."""
+        mock_exchange = Mock()
+        mock_type = Mock()
+        mock_durable = Mock()
+        options = {'durable': mock_durable}
         result = self.my_channel.exchange_declare(
-            exchange, exchange_type, durable,
+            mock_exchange, mock_type, mock_durable,
         )
-        self.broker.addExchange.assert_called_with(
-            exchange_type, exchange, options,
+        self.mock_broker.addExchange.assert_called_with(
+            mock_type, mock_exchange, options,
         )
         self.assertIsNone(result)
 
     def test_exchange_delete(self):
-        # Test the deletion of an exchange by name.
-        exchange = Mock(name='exchange')
-        result = self.my_channel.exchange_delete(exchange)
-        self.broker.delExchange.assert_called_with(exchange)
+        """Test the deletion of an exchange by name."""
+        mock_exchange = Mock()
+        result = self.my_channel.exchange_delete(mock_exchange)
+        self.mock_broker.delExchange.assert_called_with(mock_exchange)
         self.assertIsNone(result)
 
     def test_queue_bind(self):
-        # Test binding a queue to an exchange using a routing key.
-        queue = Mock(name='queue')
-        exchange = Mock(name='exchange')
-        routing_key = Mock(name='routing_key')
-        self.my_channel.queue_bind(queue, exchange, routing_key)
-        self.broker.bind.assert_called_with(exchange, queue, routing_key)
+        """Test binding a queue to an exchange using a routing key."""
+        mock_queue = Mock()
+        mock_exchange = Mock()
+        mock_routing_key = Mock()
+        self.my_channel.queue_bind(
+            mock_queue, mock_exchange, mock_routing_key,
+        )
+        self.mock_broker.bind.assert_called_with(
+            mock_exchange, mock_queue, mock_routing_key,
+        )
 
     def test_queue_unbind(self):
-        # Test unbinding a queue from an exchange using a routing key.
-        queue = Mock(name='queue')
-        exchange = Mock(name='exchange')
-        routing_key = Mock(name='routing_key')
-        self.my_channel.queue_unbind(queue, exchange, routing_key)
-        self.broker.unbind.assert_called_with(exchange, queue, routing_key)
+        """Test unbinding a queue from an exchange using a routing key."""
+        mock_queue = Mock()
+        mock_exchange = Mock()
+        mock_routing_key = Mock()
+        self.my_channel.queue_unbind(
+            mock_queue, mock_exchange, mock_routing_key,
+        )
+        self.mock_broker.unbind.assert_called_with(
+            mock_exchange, mock_queue, mock_routing_key,
+        )
 
     def test_queue_purge(self):
-        # Test purging a queue by name.
-        queue = Mock(name='queue')
-        purge_result = Mock(name='purge_result')
-        self.my_channel._purge = Mock(name='_purge', return_value=purge_result)
-        result = self.my_channel.queue_purge(queue)
-        self.my_channel._purge.assert_called_with(queue)
+        """Test purging a queue by name."""
+        mock_queue = Mock()
+        purge_result = Mock()
+        self.my_channel._purge = Mock(return_value=purge_result)
+        result = self.my_channel.queue_purge(mock_queue)
+        self.my_channel._purge.assert_called_with(mock_queue)
         self.assertIs(purge_result, result)
 
     @patch(QPID_MODULE + '.Channel.qos')
-    def test_basic_ack(self, qos):
-        # Test that basic_ack calls the QoS object properly.
-        delivery_tag = Mock(name='delivery_tag')
-        self.my_channel.basic_ack(delivery_tag)
-        qos.ack.assert_called_with(delivery_tag)
+    def test_basic_ack(self, mock_qos):
+        """Test that basic_ack calls the QoS object properly."""
+        mock_delivery_tag = Mock()
+        self.my_channel.basic_ack(mock_delivery_tag)
+        mock_qos.ack.assert_called_with(mock_delivery_tag)
 
     @patch(QPID_MODULE + '.Channel.qos')
-    def test_basic_reject(self, qos):
-        # Test that basic_reject calls the QoS object properly.
-        delivery_tag = Mock(name='delivery_tag')
-        requeue_value = Mock(name='requeue_value')
-        self.my_channel.basic_reject(delivery_tag, requeue_value)
-        qos.reject.assert_called_with(delivery_tag, requeue=requeue_value)
+    def test_basic_reject(self, mock_qos):
+        """Test that basic_reject calls the QoS object properly."""
+        mock_delivery_tag = Mock()
+        mock_requeue_value = Mock()
+        self.my_channel.basic_reject(mock_delivery_tag, mock_requeue_value)
+        mock_qos.reject.assert_called_with(
+            mock_delivery_tag, requeue=mock_requeue_value,
+        )
 
     def test_qos_manager_is_none(self):
-        # Test the qos property if the QoS object did not already exist.
+        """Test the qos property if the QoS object did not already exist."""
         self.my_channel._qos = None
         result = self.my_channel.qos
         self.assertIsInstance(result, QoS)
         self.assertEqual(result, self.my_channel._qos)
 
     def test_qos_manager_already_exists(self):
-        # Test the qos property if the QoS object already exists.
-        existing_qos = Mock(name='existing_qos')
-        self.my_channel._qos = existing_qos
+        """Test the qos property if the QoS object already exists."""
+        mock_existing_qos = Mock()
+        self.my_channel._qos = mock_existing_qos
         result = self.my_channel.qos
-        self.assertIs(existing_qos, result)
+        self.assertIs(mock_existing_qos, result)
 
     def test_prepare_message(self):
-        # Test that prepare_message() returns the correct result.
-        body = Mock(name='body')
-        priority = Mock(name='priority')
-        content_encoding = Mock(name='content_encoding')
-        content_type = Mock(name='content_type')
-        header1 = Mock(name='header1')
-        header2 = Mock(name='header2')
-        property1 = Mock(name='property1')
-        property2 = Mock(name='property2')
-        headers = {'header1': header1, 'header2': header2}
-        properties = {
-            'property1': property1,
-            'property2': property2,
-        }
+        """Test that prepare_message() returns the correct result."""
+        mock_body = Mock()
+        mock_priority = Mock()
+        mock_content_encoding = Mock()
+        mock_content_type = Mock()
+        mock_header1 = Mock()
+        mock_header2 = Mock()
+        mock_properties1 = Mock()
+        mock_properties2 = Mock()
+        headers = {'header1': mock_header1, 'header2': mock_header2}
+        properties = {'properties1': mock_properties1,
+                      'properties2': mock_properties2}
         result = self.my_channel.prepare_message(
-            body,
-            priority=priority,
-            content_type=content_type,
-            content_encoding=content_encoding,
+            mock_body,
+            priority=mock_priority,
+            content_type=mock_content_type,
+            content_encoding=mock_content_encoding,
             headers=headers,
             properties=properties)
-        self.assertIs(result['body'], body)
-        self.assertIs(result['content-encoding'], content_encoding)
-        self.assertIs(result['content-type'], content_type)
-        self.assertDictEqual(result['headers'], headers)
-        self.assertDictContainsSubset(result['properties'], properties)
+        self.assertIs(mock_body, result['body'])
+        self.assertIs(mock_content_encoding, result['content-encoding'])
+        self.assertIs(mock_content_type, result['content-type'])
+        self.assertDictEqual(headers, result['headers'])
+        self.assertDictContainsSubset(properties, result['properties'])
         self.assertIs(
-            result['properties']['delivery_info']['priority'],
-            priority,
+            mock_priority, result['properties']['delivery_info']['priority'],
         )
 
     @patch('__builtin__.buffer')
     @patch(QPID_MODULE + '.Channel.body_encoding')
     @patch(QPID_MODULE + '.Channel.encode_body')
     @patch(QPID_MODULE + '.Channel._put')
-    def test_basic_publish(self, put, encode_body, body_encoding, buffer):
-        # Test basic_publish()
-        original_body = Mock(name='original_body')
-        encoded_body = 'this is my encoded body'
-        message = {
-            'body': original_body,
-            'properties': {'delivery_info': {}},
-        }
-        encode_body.return_value = (
-            encoded_body, body_encoding,
+    def test_basic_publish(self, mock_put,
+                           mock_encode_body,
+                           mock_body_encoding,
+                           mock_buffer):
+        """Test basic_publish()."""
+        mock_original_body = Mock()
+        mock_encoded_body = 'this is my encoded body'
+        mock_message = {'body': mock_original_body,
+                        'properties': {'delivery_info': {}}}
+        mock_encode_body.return_value = (
+            mock_encoded_body, mock_body_encoding,
+        )
+        mock_exchange = Mock()
+        mock_routing_key = Mock()
+        mock_encoded_buffered_body = Mock()
+        mock_buffer.return_value = mock_encoded_buffered_body
+        self.my_channel.basic_publish(
+            mock_message, mock_exchange, mock_routing_key,
+        )
+        mock_encode_body.assert_called_once_with(
+            mock_original_body, mock_body_encoding,
+        )
+        mock_buffer.assert_called_once_with(mock_encoded_body)
+        self.assertIs(mock_message['body'], mock_encoded_buffered_body)
+        self.assertIs(
+            mock_message['properties']['body_encoding'], mock_body_encoding,
         )
-        exchange = Mock(name='exchange')
-        routing_key = Mock(name='routing_key')
-        encoded_buffered_body = Mock(name='encoded_buffered_body')
-        buffer.return_value = encoded_buffered_body
-        self.my_channel.basic_publish(message, exchange, routing_key)
-        encode_body.assert_called_once_with(original_body, body_encoding)
-        buffer.assert_called_once_with(encoded_body)
-        self.assertIs(message['body'], encoded_buffered_body)
-        self.assertIs(message['properties']['body_encoding'],
-                      body_encoding)
         self.assertIsInstance(
-            message['properties']['delivery_tag'], int,
+            mock_message['properties']['delivery_tag'], uuid.UUID,
         )
         self.assertIs(
-            message['properties']['delivery_info']['exchange'],
-            exchange,
+            mock_message['properties']['delivery_info']['exchange'],
+            mock_exchange,
         )
         self.assertIs(
-            message['properties']['delivery_info']['routing_key'],
-            routing_key,
+            mock_message['properties']['delivery_info']['routing_key'],
+            mock_routing_key,
         )
-        put.assert_called_with(
-            routing_key, message, exchange,
+        mock_put.assert_called_with(
+            mock_routing_key, mock_message, mock_exchange,
         )
 
     @patch(QPID_MODULE + '.Channel.codecs')
-    def test_encode_body_expected_encoding(self, codecs):
-        # Test if encode_body() works when encoding is set correctly.
-        body = Mock(name='body')
-        encoder = Mock(name='encoder')
-        encoded_result = Mock(name='encoded_result')
-        codecs.get.return_value = encoder
-        encoder.encode.return_value = encoded_result
-        result = self.my_channel.encode_body(body, encoding='base64')
-        expected_result = (encoded_result, 'base64')
+    def test_encode_body_expected_encoding(self, mock_codecs):
+        """Test if encode_body() works when encoding is set correctly"""
+        mock_body = Mock()
+        mock_encoder = Mock()
+        mock_encoded_result = Mock()
+        mock_codecs.get.return_value = mock_encoder
+        mock_encoder.encode.return_value = mock_encoded_result
+        result = self.my_channel.encode_body(mock_body, encoding='base64')
+        expected_result = (mock_encoded_result, 'base64')
         self.assertEqual(expected_result, result)
 
     @patch(QPID_MODULE + '.Channel.codecs')
-    def test_encode_body_not_expected_encoding(self, codecs):
-        # Test if encode_body() works when encoding is not set correctly.
-        body = Mock(name='body')
-        result = self.my_channel.encode_body(body, encoding=None)
-        expected_result = (body, None)
+    def test_encode_body_not_expected_encoding(self, mock_codecs):
+        """Test if encode_body() works when encoding is not set correctly."""
+        mock_body = Mock()
+        result = self.my_channel.encode_body(mock_body, encoding=None)
+        expected_result = mock_body, None
         self.assertEqual(expected_result, result)
 
     @patch(QPID_MODULE + '.Channel.codecs')
-    def test_decode_body_expected_encoding(self, codecs):
-        # Test if decode_body() works when encoding is set correctly.
-        body = Mock(name='body')
-        decoder = Mock(name='decoder')
-        decoded_result = Mock(name='decoded_result')
-        codecs.get.return_value = decoder
-        decoder.decode.return_value = decoded_result
-        result = self.my_channel.decode_body(body, encoding='base64')
-        self.assertEqual(decoded_result, result)
+    def test_decode_body_expected_encoding(self, mock_codecs):
+        """Test if decode_body() works when encoding is set correctly."""
+        mock_body = Mock()
+        mock_decoder = Mock()
+        mock_decoded_result = Mock()
+        mock_codecs.get.return_value = mock_decoder
+        mock_decoder.decode.return_value = mock_decoded_result
+        result = self.my_channel.decode_body(mock_body, encoding='base64')
+        self.assertEqual(mock_decoded_result, result)
 
     @patch(QPID_MODULE + '.Channel.codecs')
-    def test_decode_body_not_expected_encoding(self, codecs):
-        # Test if decode_body() works when encoding is not set correctly.
-        body = Mock(name='body')
-        result = self.my_channel.decode_body(body, encoding=None)
-        self.assertEqual(body, result)
+    def test_decode_body_not_expected_encoding(self, mock_codecs):
+        """Test if decode_body() works when encoding is not set correctly."""
+        mock_body = Mock()
+        result = self.my_channel.decode_body(mock_body, encoding=None)
+        self.assertEqual(mock_body, result)
 
     def test_typeof_exchange_exists(self):
-        # Test that typeof() finds an exchange that already exists.
-        exchange = Mock(name='exchange')
-        qpid_exchange = Mock(name='qpid_exchange')
-        attributes = {}
-        exchange_type = Mock(name='exchange_type')
-        attributes['type'] = exchange_type
-        qpid_exchange.getAttributes.return_value = attributes
-        self.broker.getExchange.return_value = qpid_exchange
-        result = self.my_channel.typeof(exchange)
-        self.assertIs(exchange_type, result)
+        """Test that typeof() finds an exchange that already exists."""
+        mock_exchange = Mock()
+        mock_qpid_exchange = Mock()
+        mock_attributes = {}
+        mock_type = Mock()
+        mock_attributes['type'] = mock_type
+        mock_qpid_exchange.getAttributes.return_value = mock_attributes
+        self.mock_broker.getExchange.return_value = mock_qpid_exchange
+        result = self.my_channel.typeof(mock_exchange)
+        self.assertIs(mock_type, result)
 
     def test_typeof_exchange_does_not_exist(self):
-        # Test that typeof() finds an exchange that does not exists.
-        exchange = Mock(name='exchange')
-        default = Mock(name='default')
-        self.broker.getExchange.return_value = None
-        result = self.my_channel.typeof(exchange, default=default)
-        self.assertIs(default, result)
-
-
-@skip.if_python3()
-@skip.if_pypy()
-class ReceiversMonitorCase(QPidCase):
-
-    def setUp(self):
-        self.session = Mock(name='session')
-        self.w = Mock(name='w')
-        self.monitor = ReceiversMonitor(self.session, self.w)
-        super(ReceiversMonitorCase, self).setUp()
-
-
-class test_ReceiversMonitor__type(ReceiversMonitorCase):
-
-    def test_is_subclass_of_threading(self):
-        self.assertIsInstance(self.monitor, threading.Thread)
+        """Test that typeof() finds an exchange that does not exists."""
+        mock_exchange = Mock()
+        mock_default = Mock()
+        self.mock_broker.getExchange.return_value = None
+        result = self.my_channel.typeof(mock_exchange, default=mock_default)
+        self.assertIs(mock_default, result)
 
 
-class test_ReceiversMonitor__init__(ReceiversMonitorCase):
+@case_no_python3
+@case_no_pypy
+@disable_runtime_dependency_check
+class TestTransportInit(Case):
 
     def setUp(self):
-        self.Thread___init__ = self.patch(
-            QPID_MODULE + '.threading.Thread.__init__',
-        )
-        super(test_ReceiversMonitor__init__, self).setUp()
-
-    def test_saves_session(self):
-        self.assertIs(self.monitor._session, self.session)
-
-    def test_saves_fd(self):
-        self.assertIs(self.monitor._w_fd, self.w)
-
-    def test_calls_parent__init__(self):
-        self.Thread___init__.assert_called_once_with()
-
-
-class test_ReceiversMonitor_run(ReceiversMonitorCase):
-
-    @patch.object(ReceiversMonitor, 'monitor_receivers')
-    @patch(QPID_MODULE + '.time.sleep')
-    def test_calls_monitor_receivers(
-            self, sleep, monitor_receivers):
-        sleep.side_effect = BreakOutException()
-        with self.assertRaises(BreakOutException):
-            self.monitor.run()
-        monitor_receivers.assert_called_once_with()
-
-    @patch.object(Transport, 'connection_errors', new=(BreakOutException,))
-    @patch.object(ReceiversMonitor, 'monitor_receivers')
-    @patch(QPID_MODULE + '.time.sleep')
-    @patch(QPID_MODULE + '.logger')
-    def test_calls_logs_exception_and_sleeps(
-            self, logger, sleep, monitor_receivers):
-        exc_to_raise = IOError()
-        monitor_receivers.side_effect = exc_to_raise
-        sleep.side_effect = BreakOutException()
-        with self.assertRaises(BreakOutException):
-            self.monitor.run()
-        logger.error.assert_called_once_with(exc_to_raise, exc_info=1)
-        sleep.assert_called_once_with(10)
-
-    @patch.object(ReceiversMonitor, 'monitor_receivers')
-    @patch(QPID_MODULE + '.time.sleep')
-    def test_loops_when_exception_is_raised(
-            self, sleep, monitor_receivers):
-
-        def return_once_raise_on_second_call(*args):
-            sleep.side_effect = BreakOutException()
-        sleep.side_effect = return_once_raise_on_second_call
-
-        with self.assertRaises(BreakOutException):
-            self.monitor.run()
-        monitor_receivers.has_calls([call(), call()])
-
-    @patch.object(Transport, 'connection_errors', new=(MockException,))
-    @patch.object(ReceiversMonitor, 'monitor_receivers')
-    @patch(QPID_MODULE + '.time.sleep')
-    @patch(QPID_MODULE + '.logger')
-    @patch(QPID_MODULE + '.os.write')
-    @patch(QPID_MODULE + '.sys.exc_info')
-    def test_exits_when_recoverable_exception_raised(
-            self, sys_exc_info, os_write, logger, sleep, monitor_receivers):
-        monitor_receivers.side_effect = MockException()
-        sleep.side_effect = BreakOutException()
-        try:
-            self.monitor.run()
-        except Exception:
-            self.fail('ReceiversMonitor.run() should exit normally when '
-                      'recoverable error is caught')
-        logger.error.assert_not_called()
-
-    @patch.object(Transport, 'connection_errors', new=(MockException,))
-    @patch.object(ReceiversMonitor, 'monitor_receivers')
-    @patch(QPID_MODULE + '.time.sleep')
-    @patch(QPID_MODULE + '.logger')
-    @patch(QPID_MODULE + '.os.write')
-    def test_saves_exception_when_recoverable_exc_raised(
-            self, os_write, logger, sleep, monitor_receivers):
-        monitor_receivers.side_effect = MockException()
-        sleep.side_effect = BreakOutException()
-        try:
-            self.monitor.run()
-        except Exception:
-            self.fail('ReceiversMonitor.run() should exit normally when '
-                      'recoverable error is caught')
-        self.assertIs(
-            self.session.saved_exception,
-            monitor_receivers.side_effect,
-        )
-
-    @patch.object(Transport, 'connection_errors', new=(MockException,))
-    @patch.object(ReceiversMonitor, 'monitor_receivers')
-    @patch(QPID_MODULE + '.time.sleep')
-    @patch(QPID_MODULE + '.logger')
-    @patch(QPID_MODULE + '.os.write')
-    @patch(QPID_MODULE + '.sys.exc_info')
-    def test_writes_e_to_pipe_when_recoverable_exc_raised(
-            self, sys_exc_info, os_write, logger, sleep, monitor_receivers):
-        monitor_receivers.side_effect = MockException()
-        sleep.side_effect = BreakOutException()
-        try:
-            self.monitor.run()
-        except Exception:
-            self.fail('ReceiversMonitor.run() should exit normally when '
-                      'recoverable error is caught')
-        os_write.assert_called_once_with(self.w, 'e')
-
-
-class test_ReceiversMonitor_monitor_receivers(ReceiversMonitorCase):
-
-    def test_calls_next_receivers(self):
-        self.session.next_receiver.side_effect = BreakOutException()
-        with self.assertRaises(BreakOutException):
-            self.monitor.monitor_receivers()
-        self.session.next_receiver.assert_called_once_with()
-
-    def test_writes_to_fd(self):
-        with patch(QPID_MODULE + '.os.write') as os_write:
-            os_write.side_effect = BreakOutException()
-            with self.assertRaises(BreakOutException):
-                self.monitor.monitor_receivers()
-            os_write.assert_called_once_with(self.w, '0')
-
-
-class test_Transport__init(QPidCase):
-
-    def setup(self):
         self.patch_a = patch.object(Transport, 'verify_runtime_environment')
-        self.verify_runtime_environment = self.patch_a.start()
+        self.mock_verify_runtime_environment = self.patch_a.start()
 
         self.patch_b = patch(QPID_MODULE + '.base.Transport.__init__')
-        self.base_Transport__init__ = self.patch_b.start()
+        self.mock_base_Transport__init__ = self.patch_b.start()
 
-        self.patch_c = patch(QPID_MODULE + '.os')
-        self.os = self.patch_c.start()
-        self.r = Mock(name='r')
-        self.w = Mock(name='w')
-        self.os.pipe.return_value = (self.r, self.w)
-
-        self.patch_d = patch(QPID_MODULE + '.fcntl')
-        self.fcntl = self.patch_d.start()
-
-    def teardown(self):
+    def tearDown(self):
         self.patch_a.stop()
         self.patch_b.stop()
-        self.patch_c.stop()
-        self.patch_d.stop()
 
-    def test_calls_verify_runtime_environment(self):
+    def test_Transport___init___calls_verify_runtime_environment(self):
         Transport(Mock())
-        self.verify_runtime_environment.assert_called_once_with()
+        self.mock_verify_runtime_environment.assert_called_once_with()
 
-    def test_calls_os_pipe(self):
-        Transport(Mock())
-        self.os.pipe.assert_called_once_with()
+    def test_transport___init___calls_parent_class___init__(self):
+        m = Mock()
+        Transport(m)
+        self.mock_base_Transport__init__.assert_called_once_with(m)
 
-    def test_saves_os_pipe_file_descriptors(self):
+    def test_transport___init___sets_use_async_interface_False(self):
         transport = Transport(Mock())
-        self.assertIs(transport.r, self.r)
-        self.assertIs(transport._w, self.w)
-
-    def test_sets_non_blocking_behavior_on_r_fd(self):
-        Transport(Mock())
-        self.fcntl.fcntl.assert_called_once_with(
-            self.r, self.fcntl.F_SETFL, self.os.O_NONBLOCK)
+        self.assertFalse(transport.use_async_interface)
 
 
-class test_Transport__drain_events(QPidCase):
+@case_no_python3
+@case_no_pypy
+@disable_runtime_dependency_check
+class TestTransportDrainEvents(Case):
 
-    def setup(self):
+    def setUp(self):
         self.transport = Transport(Mock())
-        self.transport.session = Mock(name='session')
-        self.queue = Mock(name='queue')
-        self.message = Mock(name='message')
-        self.conn = Mock(name='conn')
-        self.callback = Mock(name='callback')
-        self.conn._callbacks = {self.queue: self.callback}
+        self.transport.session = Mock()
+        self.mock_queue = Mock()
+        self.mock_message = Mock()
+        self.mock_conn = Mock()
+        self.mock_callback = Mock()
+        self.mock_conn._callbacks = {self.mock_queue: self.mock_callback}
 
     def mock_next_receiver(self, timeout):
         time.sleep(0.3)
-        receiver = Mock(name='receiver')
-        receiver.source = self.queue
-        receiver.fetch.return_value = self.message
-        return receiver
+        mock_receiver = Mock()
+        mock_receiver.source = self.mock_queue
+        mock_receiver.fetch.return_value = self.mock_message
+        return mock_receiver
 
     def test_socket_timeout_raised_when_all_receivers_empty(self):
-        with patch(QPID_MODULE + '.QpidEmpty', new=MockException):
-            self.transport.session.next_receiver.side_effect = MockException()
+        with patch(QPID_MODULE + '.QpidEmpty', new=QpidException):
+            self.transport.session.next_receiver.side_effect = QpidException()
             with self.assertRaises(socket.timeout):
                 self.transport.drain_events(Mock())
 
     def test_socket_timeout_raised_when_by_timeout(self):
         self.transport.session.next_receiver = self.mock_next_receiver
         with self.assertRaises(socket.timeout):
-            self.transport.drain_events(self.conn, timeout=1)
+            self.transport.drain_events(self.mock_conn, timeout=1)
 
     def test_timeout_returns_no_earlier_then_asked_for(self):
         self.transport.session.next_receiver = self.mock_next_receiver
-        start_time = datetime.datetime.now()
+        start_time = monotonic()
         try:
-            self.transport.drain_events(self.conn, timeout=1)
+            self.transport.drain_events(self.mock_conn, timeout=1)
         except socket.timeout:
             pass
-        end_time = datetime.datetime.now()
-        elapsed_time = end_time - start_time
-        self.assertGreaterEqual(elapsed_time.total_seconds(), 1)
+        elapsed_time_in_s = monotonic() - start_time
+        self.assertGreaterEqual(elapsed_time_in_s, 1.0)
 
     def test_callback_is_called(self):
         self.transport.session.next_receiver = self.mock_next_receiver
         try:
-            self.transport.drain_events(self.conn, timeout=1)
+            self.transport.drain_events(self.mock_conn, timeout=1)
         except socket.timeout:
             pass
-        self.callback.assert_called_with(self.message)
+        self.mock_callback.assert_called_with(self.mock_message)
 
 
-class test_Transport__create_channel(QPidCase):
+@case_no_python3
+@case_no_pypy
+@disable_runtime_dependency_check
+class TestTransportCreateChannel(Case):
 
-    def setup(self):
+    def setUp(self):
         self.transport = Transport(Mock())
-        self.conn = Mock(name='conn')
-        self.new_channel = Mock(name='new_channel')
-        self.conn.Channel.return_value = self.new_channel
-        self.returned_channel = self.transport.create_channel(self.conn)
+        self.mock_conn = Mock()
+        self.mock_new_channel = Mock()
+        self.mock_conn.Channel.return_value = self.mock_new_channel
+        self.returned_channel = self.transport.create_channel(self.mock_conn)
 
     def test_new_channel_created_from_connection(self):
-        self.assertIs(self.new_channel, self.returned_channel)
-        self.conn.Channel.assert_called_with(self.conn, self.transport)
+        self.assertIs(self.mock_new_channel, self.returned_channel)
+        self.mock_conn.Channel.assert_called_with(
+            self.mock_conn, self.transport,
+        )
 
     def test_new_channel_added_to_connection_channel_list(self):
-        append_method = self.conn.channels.append
-        append_method.assert_called_with(self.new_channel)
+        append_method = self.mock_conn.channels.append
+        append_method.assert_called_with(self.mock_new_channel)
 
 
-class test_Transport__establish_connection(QPidCase):
+@case_no_python3
+@case_no_pypy
+@disable_runtime_dependency_check
+class TestTransportEstablishConnection(Case):
 
-    def setup(self):
+    def setUp(self):
 
         class MockClient(object):
             pass
@@ -1534,110 +1534,119 @@ class test_Transport__establish_connection(QPidCase):
         self.client.connect_timeout = 4
         self.client.ssl = False
         self.client.transport_options = {}
+        self.client.userid = None
+        self.client.password = None
+        self.client.login_method = None
         self.transport = Transport(self.client)
-        self.conn = Mock(name='conn')
-        self.transport.Connection = self.conn
-        path_to_mock = QPID_MODULE + '.ReceiversMonitor'
-        self.patcher = patch(path_to_mock)
-        self.ReceiverMonitor = self.patcher.start()
-
-    def teardown(self):
-        self.patcher.stop()
-
-    def test_new_option_overwrites_default(self):
-        new_userid_string = 'new-userid'
-        self.client.userid = new_userid_string
-        self.transport.establish_connection()
-        self.conn.assert_called_once_with(
-            username=new_userid_string,
-            sasl_mechanisms='PLAIN ANONYMOUS',
-            host='127.0.0.1',
-            timeout=4,
-            password='',
-            port=5672,
-            transport='tcp',
-        )
+        self.mock_conn = Mock()
+        self.transport.Connection = self.mock_conn
 
-    def test_sasl_mech_sorting(self):
-        self.client.sasl_mechanisms = 'MECH1 MECH2'
+    def test_transport_establish_conn_new_option_overwrites_default(self):
+        self.client.userid = 'new-userid'
+        self.client.password = 'new-password'
         self.transport.establish_connection()
-        self.conn.assert_called_once_with(
-            username='guest',
-            sasl_mechanisms='MECH1 MECH2',
-            host='127.0.0.1',
+        self.mock_conn.assert_called_once_with(
+            username=self.client.userid,
+            password=self.client.password,
+            sasl_mechanisms='PLAIN',
+            host='localhost',
             timeout=4,
-            password='',
             port=5672,
             transport='tcp',
         )
 
-    def test_empty_client_is_default(self):
+    def test_transport_establish_conn_empty_client_is_default(self):
         self.transport.establish_connection()
-        self.conn.assert_called_once_with(
-            username='guest',
-            sasl_mechanisms='PLAIN ANONYMOUS',
-            host='127.0.0.1',
+        self.mock_conn.assert_called_once_with(
+            sasl_mechanisms='ANONYMOUS',
+            host='localhost',
             timeout=4,
-            password='',
             port=5672,
             transport='tcp',
         )
 
-    def test_additional_transport_option(self):
+    def test_transport_establish_conn_additional_transport_option(self):
         new_param_value = 'mynewparam'
         self.client.transport_options['new_param'] = new_param_value
         self.transport.establish_connection()
-        self.conn.assert_called_once_with(
-            username='guest',
-            sasl_mechanisms='PLAIN ANONYMOUS',
-            host='127.0.0.1',
+        self.mock_conn.assert_called_once_with(
+            sasl_mechanisms='ANONYMOUS',
+            host='localhost',
             timeout=4,
             new_param=new_param_value,
-            password='',
             port=5672,
             transport='tcp',
         )
 
-    def test_transform_localhost_to_127_0_0_1(self):
+    def test_transport_establish_conn_transform_localhost_to_127_0_0_1(self):
         self.client.hostname = 'localhost'
         self.transport.establish_connection()
-        self.conn.assert_called_once_with(
-            username='guest',
-            sasl_mechanisms='PLAIN ANONYMOUS',
-            host='127.0.0.1',
+        self.mock_conn.assert_called_once_with(
+            sasl_mechanisms='ANONYMOUS',
+            host='localhost',
             timeout=4,
-            password='',
             port=5672,
             transport='tcp',
         )
 
-    def test_set_password(self):
+    def test_transport_password_no_userid_raises_exception(self):
         self.client.password = 'somepass'
+        self.assertRaises(Exception, self.transport.establish_connection)
+
+    def test_transport_userid_no_password_raises_exception(self):
+        self.client.userid = 'someusername'
+        self.assertRaises(Exception, self.transport.establish_connection)
+
+    def test_transport_overrides_sasl_mech_from_login_method(self):
+        self.client.login_method = 'EXTERNAL'
+        self.transport.establish_connection()
+        self.mock_conn.assert_called_once_with(
+            sasl_mechanisms='EXTERNAL',
+            host='localhost',
+            timeout=4,
+            port=5672,
+            transport='tcp',
+        )
+
+    def test_transport_overrides_sasl_mech_has_username(self):
+        self.client.userid = 'new-userid'
+        self.client.login_method = 'EXTERNAL'
         self.transport.establish_connection()
-        self.conn.assert_called_once_with(
-            username='guest',
-            sasl_mechanisms='PLAIN ANONYMOUS',
-            host='127.0.0.1',
+        self.mock_conn.assert_called_once_with(
+            username=self.client.userid,
+            sasl_mechanisms='EXTERNAL',
+            host='localhost',
             timeout=4,
+            port=5672,
+            transport='tcp',
+        )
+
+    def test_transport_establish_conn_set_password(self):
+        self.client.userid = 'someuser'
+        self.client.password = 'somepass'
+        self.transport.establish_connection()
+        self.mock_conn.assert_called_once_with(
+            username='someuser',
             password='somepass',
+            sasl_mechanisms='PLAIN',
+            host='localhost',
+            timeout=4,
             port=5672,
             transport='tcp',
         )
 
-    def test_no_ssl_sets_transport_tcp(self):
+    def test_transport_establish_conn_no_ssl_sets_transport_tcp(self):
         self.client.ssl = False
         self.transport.establish_connection()
-        self.conn.assert_called_once_with(
-            username='guest',
-            sasl_mechanisms='PLAIN ANONYMOUS',
-            host='127.0.0.1',
+        self.mock_conn.assert_called_once_with(
+            sasl_mechanisms='ANONYMOUS',
+            host='localhost',
             timeout=4,
-            password='',
             port=5672,
             transport='tcp',
         )
 
-    def test_with_ssl_with_hostname_check(self):
+    def test_transport_establish_conn_with_ssl_with_hostname_check(self):
         self.client.ssl = {
             'keyfile': 'my_keyfile',
             'certfile': 'my_certfile',
@@ -1645,20 +1654,18 @@ class test_Transport__establish_connection(QPidCase):
             'cert_reqs': ssl.CERT_REQUIRED,
         }
         self.transport.establish_connection()
-        self.conn.assert_called_once_with(
-            username='guest',
+        self.mock_conn.assert_called_once_with(
             ssl_certfile='my_certfile',
             ssl_trustfile='my_cacerts',
             timeout=4,
             ssl_skip_hostname_check=False,
-            sasl_mechanisms='PLAIN ANONYMOUS',
-            host='127.0.0.1',
+            sasl_mechanisms='ANONYMOUS',
+            host='localhost',
             ssl_keyfile='my_keyfile',
-            password='',
             port=5672, transport='ssl',
         )
 
-    def test_with_ssl_skip_hostname_check(self):
+    def test_transport_establish_conn_with_ssl_skip_hostname_check(self):
         self.client.ssl = {
             'keyfile': 'my_keyfile',
             'certfile': 'my_certfile',
@@ -1666,151 +1673,209 @@ class test_Transport__establish_connection(QPidCase):
             'cert_reqs': ssl.CERT_OPTIONAL,
         }
         self.transport.establish_connection()
-        self.conn.assert_called_once_with(
-            username='guest',
+        self.mock_conn.assert_called_once_with(
             ssl_certfile='my_certfile',
             ssl_trustfile='my_cacerts',
             timeout=4,
             ssl_skip_hostname_check=True,
-            sasl_mechanisms='PLAIN ANONYMOUS',
-            host='127.0.0.1',
+            sasl_mechanisms='ANONYMOUS',
+            host='localhost',
             ssl_keyfile='my_keyfile',
-            password='',
             port=5672, transport='ssl',
         )
 
-    def test_sets_client_on_connection_object(self):
+    def test_transport_establish_conn_sets_client_on_connection_object(self):
         self.transport.establish_connection()
-        self.assertIs(self.conn.return_value.client, self.client)
+        self.assertIs(self.mock_conn.return_value.client, self.client)
 
-    def test_creates_session_on_transport(self):
+    def test_transport_establish_conn_creates_session_on_transport(self):
         self.transport.establish_connection()
-        qpid_conn = self.conn.return_value.get_qpid_connection
-        new_session = qpid_conn.return_value.session.return_value
-        self.assertIs(self.transport.session, new_session)
+        qpid_conn = self.mock_conn.return_value.get_qpid_connection
+        new_mock_session = qpid_conn.return_value.session.return_value
+        self.assertIs(self.transport.session, new_mock_session)
 
-    def test_returns_new_connection_object(self):
+    def test_transport_establish_conn_returns_new_connection_object(self):
         new_conn = self.transport.establish_connection()
-        self.assertIs(new_conn, self.conn.return_value)
+        self.assertIs(new_conn, self.mock_conn.return_value)
 
-    def test_creates_ReceiversMonitor(self):
+    def test_transport_establish_conn_uses_hostname_if_not_default(self):
+        self.client.hostname = 'some_other_hostname'
         self.transport.establish_connection()
-        self.ReceiverMonitor.assert_called_once_with(
-            self.transport.session, self.transport._w)
+        self.mock_conn.assert_called_once_with(
+            sasl_mechanisms='ANONYMOUS',
+            host='some_other_hostname',
+            timeout=4,
+            port=5672,
+            transport='tcp',
+        )
 
-    def test_daemonizes_thread(self):
+    def test_transport_sets_qpid_message_ready_handler(self):
         self.transport.establish_connection()
-        self.assertTrue(self.ReceiverMonitor.return_value.daemon)
+        qpid_conn_call = self.mock_conn.return_value.get_qpid_connection
+        mock_session = qpid_conn_call.return_value.session.return_value
+        mock_set_callback = mock_session.set_message_received_notify_handler
+        expected_msg_callback = self.transport._qpid_message_ready_handler
+        mock_set_callback.assert_called_once_with(expected_msg_callback)
 
-    def test_starts_thread(self):
+    def test_transport_sets_session_exception_handler(self):
         self.transport.establish_connection()
-        new_receiver_monitor = self.ReceiverMonitor.return_value
-        new_receiver_monitor.start.assert_called_once_with()
+        qpid_conn_call = self.mock_conn.return_value.get_qpid_connection
+        mock_session = qpid_conn_call.return_value.session.return_value
+        mock_set_callback = mock_session.set_async_exception_notify_handler
+        exc_callback = self.transport._qpid_async_exception_notify_handler
+        mock_set_callback.assert_called_once_with(exc_callback)
 
-    def test_ignores_hostname_if_not_localhost(self):
-        self.client.hostname = 'some_other_hostname'
+    def test_transport_sets_connection_exception_handler(self):
         self.transport.establish_connection()
-        self.conn.assert_called_once_with(
-            username='guest',
-            sasl_mechanisms='PLAIN ANONYMOUS',
-            host='some_other_hostname',
-            timeout=4, password='',
-            port=5672, transport='tcp',
-        )
+        qpid_conn_call = self.mock_conn.return_value.get_qpid_connection
+        qpid_conn = qpid_conn_call.return_value
+        mock_set_callback = qpid_conn.set_async_exception_notify_handler
+        exc_callback = self.transport._qpid_async_exception_notify_handler
+        mock_set_callback.assert_called_once_with(exc_callback)
 
 
-class test_Transport__class_attributes(QPidCase):
+@case_no_python3
+@case_no_pypy
+class TestTransportClassAttributes(Case):
 
-    def test_Connection_attribute(self):
+    def test_verify_Connection_attribute(self):
         self.assertIs(Connection, Transport.Connection)
 
-    def test_default_port(self):
-        self.assertEqual(Transport.default_port, 5672)
-
-    def test_polling_disabled(self):
+    def test_verify_polling_disabled(self):
         self.assertIsNone(Transport.polling_interval)
 
-    def test_supports_asynchronous_events(self):
+    def test_transport_verify_supports_asynchronous_events(self):
         self.assertTrue(Transport.supports_ev)
 
     def test_verify_driver_type_and_name(self):
-        self.assertEqual(Transport.driver_type, 'qpid')
-        self.assertEqual(Transport.driver_name, 'qpid')
+        self.assertEqual('qpid', Transport.driver_type)
+        self.assertEqual('qpid', Transport.driver_name)
+
+    def test_transport_verify_recoverable_connection_errors(self):
+        connection_errors = Transport.recoverable_connection_errors
+        self.assertIn(ConnectionError, connection_errors)
+        self.assertIn(select.error, connection_errors)
 
-    def test_channel_error_contains_qpid_ConnectionError(self):
-        self.assertIn(ConnectionError, Transport.connection_errors)
+    def test_transport_verify_recoverable_channel_errors(self):
+        channel_errors = Transport.recoverable_channel_errors
+        self.assertIn(NotFound, channel_errors)
 
-    def test_channel_error_contains_socket_error(self):
-        self.assertIn(select.error, Transport.connection_errors)
+    def test_transport_verify_pre_kombu_3_0_exception_labels(self):
+        self.assertEqual(Transport.recoverable_channel_errors,
+                         Transport.channel_errors)
+        self.assertEqual(Transport.recoverable_connection_errors,
+                         Transport.connection_errors)
 
 
-class test_Transport__register_with_event_loop(QPidCase):
+@case_no_python3
+@case_no_pypy
+@disable_runtime_dependency_check
+class TestTransportRegisterWithEventLoop(Case):
 
-    def test_calls_add_reader(self):
+    def test_transport_register_with_event_loop_calls_add_reader(self):
         transport = Transport(Mock())
-        connection = Mock(name='connection')
-        loop = Mock(name='loop')
-        transport.register_with_event_loop(connection, loop)
-        loop.add_reader.assert_called_with(
-            transport.r,
-            transport.on_readable,
-            connection, loop,
+        mock_connection = Mock()
+        mock_loop = Mock()
+        transport.register_with_event_loop(mock_connection, mock_loop)
+        mock_loop.add_reader.assert_called_with(
+            transport.r, transport.on_readable, mock_connection, mock_loop,
         )
 
 
-class test_Transport__on_readable(QPidCase):
+@case_no_python3
+@case_no_pypy
+@disable_runtime_dependency_check
+class TestTransportQpidCallbackHandlersAsync(Case):
+
+    def setUp(self):
+        self.patch_a = patch(QPID_MODULE + '.os.write')
+        self.mock_os_write = self.patch_a.start()
+        self.transport = Transport(Mock())
+        self.transport.register_with_event_loop(Mock(), Mock())
+
+    def tearDown(self):
+        self.patch_a.stop()
+
+    def test__qpid_message_ready_handler_writes_symbol_to_fd(self):
+        self.transport._qpid_message_ready_handler(Mock())
+        self.mock_os_write.assert_called_once_with(self.transport._w, '0')
+
+    def test__qpid_async_exception_notify_handler_writes_symbol_to_fd(self):
+        self.transport._qpid_async_exception_notify_handler(Mock(), Mock())
+        self.mock_os_write.assert_called_once_with(self.transport._w, 'e')
+
+
+@case_no_python3
+@case_no_pypy
+@disable_runtime_dependency_check
+class TestTransportQpidCallbackHandlersSync(Case):
+
+    def setUp(self):
+        self.patch_a = patch(QPID_MODULE + '.os.write')
+        self.mock_os_write = self.patch_a.start()
+        self.transport = Transport(Mock())
+
+    def tearDown(self):
+        self.patch_a.stop()
+
+    def test__qpid_message_ready_handler_dows_not_write(self):
+        self.transport._qpid_message_ready_handler(Mock())
+        self.assertTrue(not self.mock_os_write.called)
+
+    def test__qpid_async_exception_notify_handler_does_not_write(self):
+        self.transport._qpid_async_exception_notify_handler(Mock(), Mock())
+        self.assertTrue(not self.mock_os_write.called)
+
 
-    def setup(self):
+@case_no_python3
+@case_no_pypy
+@disable_runtime_dependency_check
+class TestTransportOnReadable(Case):
+
+    def setUp(self):
         self.patch_a = patch(QPID_MODULE + '.os.read')
-        self.os_read = self.patch_a.start()
+        self.mock_os_read = self.patch_a.start()
+
         self.patch_b = patch.object(Transport, 'drain_events')
-        self.drain_events = self.patch_b.start()
+        self.mock_drain_events = self.patch_b.start()
         self.transport = Transport(Mock())
+        self.transport.register_with_event_loop(Mock(), Mock())
 
-    def teardown(self):
+    def tearDown(self):
         self.patch_a.stop()
+        self.patch_b.stop()
 
-    def test_reads_symbol_from_fd(self):
+    def test_transport_on_readable_reads_symbol_from_fd(self):
         self.transport.on_readable(Mock(), Mock())
-        self.os_read.assert_called_once_with(self.transport.r, 1)
+        self.mock_os_read.assert_called_once_with(self.transport.r, 1)
 
-    def test_calls_drain_events(self):
-        connection = Mock(name='connection')
-        self.transport.on_readable(connection, Mock())
-        self.drain_events.assert_called_with(connection)
+    def test_transport_on_readable_calls_drain_events(self):
+        mock_connection = Mock()
+        self.transport.on_readable(mock_connection, Mock())
+        self.mock_drain_events.assert_called_with(mock_connection)
 
-    def test_catches_socket_timeout(self):
-        self.drain_events.side_effect = socket.timeout()
-        try:
-            self.transport.on_readable(Mock(), Mock())
-        except Exception:
-            self.fail('Transport.on_readable did not catch socket.timeout()')
-
-    def test_ignores_non_socket_timeout_exception(self):
-        self.drain_events.side_effect = IOError()
-        with self.assertRaises(IOError):
-            self.transport.on_readable(Mock(), Mock())
+    def test_transport_on_readable_catches_socket_timeout(self):
+        self.mock_drain_events.side_effect = socket.timeout()
+        self.transport.on_readable(Mock(), Mock())
 
-    def test_reads_e_off_of_pipe_raises_exc_info(self):
-        self.transport.session = Mock(name='session')
-        try:
-            raise IOError()
-        except Exception as error:
-            self.transport.session.saved_exception = error
-        self.os_read.return_value = 'e'
+    def test_transport_on_readable_ignores_non_socket_timeout_exception(self):
+        self.mock_drain_events.side_effect = IOError()
         with self.assertRaises(IOError):
             self.transport.on_readable(Mock(), Mock())
 
 
-class test_Transport__verify_runtime_environment(QPidCase):
+@case_no_python3
+@case_no_pypy
+@disable_runtime_dependency_check
+class TestTransportVerifyRuntimeEnvironment(Case):
 
-    def setup(self):
+    def setUp(self):
         self.verify_runtime_environment = Transport.verify_runtime_environment
         self.patch_a = patch.object(Transport, 'verify_runtime_environment')
         self.patch_a.start()
         self.transport = Transport(Mock())
 
-    def teardown(self):
+    def tearDown(self):
         self.patch_a.stop()
 
     @patch(QPID_MODULE + '.PY3', new=True)
@@ -1824,40 +1889,65 @@ class test_Transport__verify_runtime_environment(QPidCase):
         with self.assertRaises(RuntimeError):
             self.verify_runtime_environment(self.transport)
 
-    def test_raises_no_exception(self):
-        try:
+    @patch(QPID_MODULE + '.dependency_is_none')
+    def test_raises_exc_dep_missing(self, mock_dep_is_none):
+        mock_dep_is_none.return_value = True
+        with self.assertRaises(RuntimeError):
             self.verify_runtime_environment(self.transport)
-        except Exception:
-            self.fail(
-                'verify_runtime_environment raised an unexpected Exception')
 
+    @patch(QPID_MODULE + '.dependency_is_none')
+    def test_calls_dependency_is_none(self, mock_dep_is_none):
+        mock_dep_is_none.return_value = False
+        self.verify_runtime_environment(self.transport)
+        self.assertTrue(mock_dep_is_none.called)
 
-class test_Transport(QPidCase):
+    def test_raises_no_exception(self):
+        self.verify_runtime_environment(self.transport)
+
+
+@case_no_python3
+@case_no_pypy
+@disable_runtime_dependency_check
+class TestTransport(ExtraAssertionsMixin, Case):
 
-    def setup(self):
-        # Creates a mock client to be used in testing.
-        self.client = Mock(name='client')
+    def setUp(self):
+        """Creates a mock_client to be used in testing."""
+        self.mock_client = Mock()
 
     def test_close_connection(self):
-        # Test that close_connection calls close on each channel in the
-        # list of channels on the connection object.
-        my_transport = Transport(self.client)
-        connection = Mock(name='connection')
-        channel1 = Mock(name='channel1')
-        channel2 = Mock(name='channel2')
-        connection.channels = [channel1, channel2]
-        my_transport.close_connection(connection)
-        channel1.close.assert_called_with()
-        channel2.close.assert_called_with()
+        """Test that close_connection calls close on the connection."""
+        my_transport = Transport(self.mock_client)
+        mock_connection = Mock()
+        my_transport.close_connection(mock_connection)
+        mock_connection.close.assert_called_once_with()
 
     def test_default_connection_params(self):
-        # Test that the default_connection_params are correct.
+        """Test that the default_connection_params are correct"""
         correct_params = {
-            'userid': 'guest', 'password': '',
-            'port': 5672, 'virtual_host': '',
             'hostname': 'localhost',
-            'sasl_mechanisms': 'PLAIN ANONYMOUS',
+            'port': 5672,
         }
-        my_transport = Transport(self.client)
+        my_transport = Transport(self.mock_client)
         result_params = my_transport.default_connection_params
         self.assertDictEqual(correct_params, result_params)
+
+    @patch(QPID_MODULE + '.os.close')
+    def test_del_sync(self, close):
+        my_transport = Transport(self.mock_client)
+        my_transport.__del__()
+        self.assertFalse(close.called)
+
+    @patch(QPID_MODULE + '.os.close')
+    def test_del_async(self, close):
+        my_transport = Transport(self.mock_client)
+        my_transport.register_with_event_loop(Mock(), Mock())
+        my_transport.__del__()
+        self.assertTrue(close.called)
+
+    @patch(QPID_MODULE + '.os.close')
+    def test_del_async_failed(self, close):
+        close.side_effect = OSError()
+        my_transport = Transport(self.mock_client)
+        my_transport.register_with_event_loop(Mock(), Mock())
+        my_transport.__del__()
+        self.assertTrue(close.called)
diff --git a/kombu/transport/qpid.py b/kombu/transport/qpid.py
index f664799c..9afa8474 100644
--- a/kombu/transport/qpid.py
+++ b/kombu/transport/qpid.py
@@ -25,15 +25,9 @@ or to install the requirements manually:
     to underlying dependencies not being compatible. This version is
     tested and works with with Python 2.7.
 
-.. admonition:: Potential Deadlock
-
-    This transport should be used with caution due to a known
-    potential deadlock. See `Issue 2199`_ for more details.
-
 .. _`Qpid`: http://qpid.apache.org/
 .. _`qpid-python`: http://pypi.python.org/pypi/qpid-python/
 .. _`qpid-tools`: http://pypi.python.org/pypi/qpid-tools/
-.. _`Issue 2199`: https://github.com/celery/celery/issues/2199
 
 Authentication
 ==============
@@ -83,33 +77,27 @@ The :attr:`~kombu.Connection.transport_options` argument to the
 options override and replace any other default or specified values. If using
 Celery, this can be accomplished by setting the
 *BROKER_TRANSPORT_OPTIONS* Celery option.
->>>>>>> ba4fa60... [qpid] Fixes rst syntax errors in docstrings
 
 """
 from __future__ import absolute_import, unicode_literals
 
+from collections import OrderedDict
 import os
 import select
 import socket
 import ssl
 import sys
-import threading
 import time
+import uuid
 
-from collections import OrderedDict
-from itertools import count
+from gettext import gettext as _
 
 import amqp.protocol
 
-from kombu.five import Empty, items
-from kombu.log import get_logger
-from kombu.transport.virtual import Base64, Message
-from kombu.transport import base
-
 try:
     import fcntl
 except ImportError:
-    fcntl = None
+    fcntl = None  # noqa
 
 try:
     import qpidtoollibs
@@ -117,24 +105,29 @@ except ImportError:  # pragma: no cover
     qpidtoollibs = None     # noqa
 
 try:
-    from qpid.messaging.exceptions import ConnectionError
+    from qpid.messaging.exceptions import ConnectionError, NotFound
     from qpid.messaging.exceptions import Empty as QpidEmpty
+    from qpid.messaging.exceptions import SessionClosed
 except ImportError:  # pragma: no cover
     ConnectionError = None
+    NotFound = None
     QpidEmpty = None
-
-# ## The Following Import Applies Monkey Patches at Import Time ##
-import kombu.transport.qpid_patches  # noqa
-################################################################
+    SessionClosed = None
 
 try:
     import qpid
 except ImportError:  # pragma: no cover
     qpid = None
 
+
+from kombu.five import Empty, items
+from kombu.log import get_logger
+from kombu.transport.virtual import Base64, Message
+from kombu.transport import base
+
+
 logger = get_logger(__name__)
 
-DEFAULT_PORT = 5672
 
 OBJECT_ALREADY_EXISTS_STRING = 'object already exists'
 
@@ -144,70 +137,19 @@ __version__ = '.'.join(map(str, VERSION))
 PY3 = sys.version_info[0] == 3
 
 
-E_AUTH = """
-Unable to authenticate to qpid using the following mechanisms: %s
-"""
-
-E_UNREACHABLE = """
-Unable to connect to qpid with SASL mechanism %s
-"""
-
-
-class AuthenticationFailure(Exception):
-    pass
-
+def dependency_is_none(dependency):
+    """Return True if the dependency is None, otherwise False. This is done
+    using a function so that tests can mock this behavior easily.
 
-class QpidMessagingExceptionHandler(object):
-    """An exception handling decorator that silences some exceptions.
-
-    An exception handling class designed to silence specific exceptions
-    that qpid.messaging raises as part of normal operation. qpid.messaging
-    exceptions require string parsing, and are not machine consumable.
-    This is designed to be used as a decorator, and accepts a whitelist
-    string as an argument.
-
-    Usage:
-    @QpidMessagingExceptionHandler('whitelist string goes here')
+    :param dependency: The module to check if it is None
+    :return: True if dependency is None otherwise False.
 
     """
+    return dependency is None
 
-    def __init__(self, allowed_exception_string):
-        """Instantiate a QpidMessagingExceptionHandler object.
-
-        :param allowed_exception_string: a string that, if present in the
-            exception message, will be silenced.
-        :type allowed_exception_string: str
 
-        """
-        self.allowed_exception_string = allowed_exception_string
-
-    def __call__(self, original_func):
-        """The decorator method.
-
-        Method that wraps the actual function with exception silencing
-        functionality. Any exception that contains the string
-        :attr:`allowed_exception_string` in the message will be silenced.
-
-        :param original_func: function that is automatically passed in
-            when this object is used as a decorator.
-        :type original_func: function
-
-        :return: A function that decorates (wraps) the original function.
-        :rtype: function
-
-        """
-
-        def decorator(*args, **kwargs):
-            """A runtime-built function that will be returned which contains
-            a reference to the original function, and wraps a call to it in
-            a try/except block that can silence errors."""
-            try:
-                return original_func(*args, **kwargs)
-            except Exception as exc:
-                if self.allowed_exception_string not in str(exc):
-                    raise
-
-        return decorator
+class AuthenticationFailure(Exception):
+    pass
 
 
 class QoS(object):
@@ -230,7 +172,7 @@ class QoS(object):
     ACKed asynchronously through a call to :meth:`ack`. Messages that are
     received, but not ACKed will not be delivered by the broker to another
     consumer until an ACK is received, or the session is closed. Messages
-    are referred to using delivery_tag integers, which are unique per
+    are referred to using delivery_tag, which are unique per
     :class:`Channel`. Delivery tags are managed outside of this object and
     are passed in with a message to :meth:`append`. Un-ACKed messages can
     be looked up from QoS using :meth:`get` and can be rejected and
@@ -282,13 +224,13 @@ class QoS(object):
     def append(self, message, delivery_tag):
         """Append message to the list of un-ACKed messages.
 
-        Add a message, referenced by the integer delivery_tag, for ACKing,
+        Add a message, referenced by the delivery_tag, for ACKing,
         rejecting, or getting later. Messages are saved into an
         :class:`collections.OrderedDict` by delivery_tag.
 
         :param message: A received message that has not yet been ACKed.
         :type message: qpid.messaging.Message
-        :param delivery_tag: An integer number to refer to this message by
+        :param delivery_tag: A UUID to refer to this message by
             upon receipt.
         :type delivery_tag: uuid.UUID
 
@@ -301,7 +243,7 @@ class QoS(object):
 
         :param delivery_tag: The delivery tag associated with the message
             to be returned.
-        :type delivery_tag: int
+        :type delivery_tag: uuid.UUID
 
         :return: An un-ACKed message that is looked up by delivery_tag.
         :rtype: qpid.messaging.Message
@@ -336,7 +278,7 @@ class QoS(object):
 
         :param delivery_tag: The delivery tag associated with the message
             to be rejected.
-        :type delivery_tag: int
+        :type delivery_tag: uuid.UUID
         :keyword requeue: If True, the broker will be notified to requeue
             the message. If False, the broker will be told to drop the
             message entirely. In both cases, the message will be removed
@@ -389,10 +331,9 @@ class Channel(base.StdChannel):
     Messages sent using this channel are assigned a delivery_tag. The
     delivery_tag is generated for a message as they are prepared for
     sending by :meth:`basic_publish`. The delivery_tag is unique per
-    Channel instance using :meth:`~itertools.count`. The delivery_tag has
-    no meaningful context in other objects, and is only maintained in the
-    memory of this object, and the underlying :class:`QoS` object that
-    provides support.
+    channel instance. The delivery_tag has no meaningful context in other
+    objects, and is only maintained in the memory of this object, and the
+    underlying :class:`QoS` object that provides support.
 
     Each channel object instantiates exactly one :class:`QoS` object for
     prefetch limiting, and asynchronous ACKing. The :class:`QoS` object is
@@ -423,8 +364,8 @@ class Channel(base.StdChannel):
 
     Each call to :meth:`basic_consume` creates a consumer, which is given a
     consumer tag that is identified by the caller of :meth:`basic_consume`.
-    Already started consumers can be canceled using by their consumer_tag
-    using :meth:`basic_cancel`. Cancelation of a consumer causes the
+    Already started consumers can be cancelled using by their consumer_tag
+    using :meth:`basic_cancel`. Cancellation of a consumer causes the
     :class:`~qpid.messaging.endpoints.Receiver` object to be closed.
 
     Asynchronous message ACKing is supported through :meth:`basic_ack`,
@@ -447,9 +388,6 @@ class Channel(base.StdChannel):
     #: Binary <-> ASCII codecs.
     codecs = {'base64': Base64()}
 
-    #: counter used to generate delivery tags for this channel.
-    _delivery_tags = count(1)
-
     def __init__(self, connection, transport):
         self.connection = connection
         self.transport = transport
@@ -529,13 +467,11 @@ class Channel(base.StdChannel):
         """
         if not exchange:
             address = '%s; {assert: always, node: {type: queue}}' % (
-                routing_key,
-            )
+                routing_key,)
             msg_subject = None
         else:
             address = '%s/%s; {assert: always, node: {type: topic}}' % (
-                exchange, routing_key,
-            )
+                exchange, routing_key)
             msg_subject = str(routing_key)
         sender = self.transport.session.sender(address)
         qpid_message = qpid.messaging.Message(content=message,
@@ -549,11 +485,14 @@ class Channel(base.StdChannel):
         """Purge all undelivered messages from a queue specified by name.
 
         An internal method to purge all undelivered messages from a queue
-        specified by name. The queue message depth is first checked,
-        and then the broker is asked to purge that number of messages. The
-        integer number of messages requested to be purged is returned. The
-        actual number of messages purged may be different than the
-        requested number of messages to purge (see below).
+        specified by name. If the queue does not exist a
+        :class:`qpid.messaging.exceptions.NotFound` exception is raised.
+
+        The queue message depth is first checked, and then the broker is
+        asked to purge that number of messages. The integer number of
+        messages requested to be purged is returned. The actual number of
+        messages purged may be different than the requested number of
+        messages to purge (see below).
 
         Sometimes delivered messages are asked to be purged, but are not.
         This case fails silently, which is the correct behavior when a
@@ -577,6 +516,9 @@ class Channel(base.StdChannel):
 
         """
         queue_to_purge = self._broker.getQueue(queue)
+        if queue_to_purge is None:
+            error_text = "NOT_FOUND - no queue '{0}'".format(queue)
+            raise NotFound(code=404, text=error_text)
         message_count = queue_to_purge.values['msgDepth']
         if message_count > 0:
             queue_to_purge.purge(message_count)
@@ -661,7 +603,7 @@ class Channel(base.StdChannel):
         'exclusive' flag always implies 'auto-delete'. Default is False.
 
         If auto_delete is True,  the queue is deleted when all consumers
-        have finished using it. The last consumer can be canceled either
+        have finished using it. The last consumer can be cancelled either
         explicitly or because its channel is closed. If there was no
         consumer ever on the queue, it won't be deleted. Default is True.
 
@@ -720,7 +662,7 @@ class Channel(base.StdChannel):
             self._broker.addQueue(queue, options=options)
         except Exception as exc:
             if OBJECT_ALREADY_EXISTS_STRING not in str(exc):
-                raise
+                raise exc
         queue_to_check = self._broker.getQueue(queue)
         message_count = queue_to_check.values['msgDepth']
         consumer_count = queue_to_check.values['consumerCount']
@@ -755,7 +697,6 @@ class Channel(base.StdChannel):
                 return
             self._delete(queue)
 
-    @QpidMessagingExceptionHandler(OBJECT_ALREADY_EXISTS_STRING)
     def exchange_declare(self, exchange='', type='direct', durable=False,
                          **kwargs):
         """Create a new exchange.
@@ -772,18 +713,23 @@ class Channel(base.StdChannel):
         functionality.
 
         :keyword type: The exchange type. Valid values include 'direct',
-        'topic', and 'fanout'.
+            'topic', and 'fanout'.
         :type type: str
         :keyword exchange: The name of the exchange to be created. If no
-        exchange is specified, then a blank string will be used as the name.
+            exchange is specified, then a blank string will be used as the
+            name.
         :type exchange: str
         :keyword durable: True if the exchange should be durable, or False
-        otherwise.
+            otherwise.
         :type durable: bool
 
         """
         options = {'durable': durable}
-        self._broker.addExchange(type, exchange, options)
+        try:
+            self._broker.addExchange(type, exchange, options)
+        except Exception as exc:
+            if OBJECT_ALREADY_EXISTS_STRING not in str(exc):
+                raise exc
 
     def exchange_delete(self, exchange_name, **kwargs):
         """Delete an exchange specified by name
@@ -839,12 +785,12 @@ class Channel(base.StdChannel):
     def queue_purge(self, queue, **kwargs):
         """Remove all undelivered messages from queue.
 
-        Purge all undelivered messages from a queue specified by name. The
-        queue message depth is first checked, and then the broker is asked
-        to purge that number of messages. The integer number of messages
-        requested to be purged is returned. The actual number of messages
-        purged may be different than the requested number of messages to
-        purge.
+        Purge all undelivered messages from a queue specified by name. If the
+        queue does not exist an exception is raised. The queue message
+        depth is first checked, and then the broker is asked to purge that
+        number of messages. The integer number of messages requested to be
+        purged is returned. The actual number of messages purged may be
+        different than the requested number of messages to purge.
 
         Sometimes delivered messages are asked to be purged, but are not.
         This case fails silently, which is the correct behavior when a
@@ -939,7 +885,7 @@ class Channel(base.StdChannel):
 
         :param delivery_tag: The delivery tag associated with the message
             to be rejected.
-        :type delivery_tag: int
+        :type delivery_tag: uuid.UUID
         :keyword requeue: If False, the rejected message will be dropped by
             the broker and not delivered to any other consumers. If True,
             then the rejected message will be requeued for delivery to
@@ -980,10 +926,9 @@ class Channel(base.StdChannel):
         handled by the caller of :meth:`~Transport.drain_events`. Messages
         can be ACKed after being received through a call to :meth:`basic_ack`.
 
-        If no_ack is True, the messages are immediately ACKed to avoid a
-        memory leak in qpid.messaging when messages go un-ACKed. The no_ack
-        flag indicates that the receiver of the message does not intent to
-        call :meth:`basic_ack`.
+        If no_ack is True, The no_ack flag indicates that the receiver of
+        the message will not call :meth:`basic_ack` later. Since the
+        message will not be ACKed later, it is ACKed immediately.
 
         :meth:`basic_consume` transforms the message object type prior to
         calling the callback. Initially the message comes in as a
@@ -1020,9 +965,7 @@ class Channel(base.StdChannel):
             delivery_tag = message.delivery_tag
             self.qos.append(qpid_message, delivery_tag)
             if no_ack:
-                # Celery will not ack this message later, so we should to
-                # avoid a memory leak in qpid.messaging due to un-ACKed
-                # messages.
+                # Celery will not ack this message later, so we should ack now
                 self.basic_ack(delivery_tag)
             return callback(message)
 
@@ -1041,7 +984,7 @@ class Channel(base.StdChannel):
         This method also cleans up all lingering references of the consumer.
 
         :param consumer_tag: The tag which refers to the consumer to be
-            canceled. Originally specified when the consumer was created
+            cancelled. Originally specified when the consumer was created
             as a parameter to :meth:`basic_consume`.
         :type consumer_tag: an immutable object
 
@@ -1053,7 +996,7 @@ class Channel(base.StdChannel):
             self.connection._callbacks.pop(queue, None)
 
     def close(self):
-        """Close Channel and all associated messages.
+        """Cancel all associated messages and close the Channel.
 
         This cancels all consumers by calling :meth:`basic_cancel` for each
         known consumer_tag. It also closes the self._broker sessions. Closing
@@ -1136,13 +1079,11 @@ class Channel(base.StdChannel):
         info = properties.setdefault('delivery_info', {})
         info['priority'] = priority or 0
 
-        return {
-            'body': body,
-            'content-encoding': content_encoding,
-            'content-type': content_type,
-            'headers': headers or {},
-            'properties': properties or {},
-        }
+        return {'body': body,
+                'content-encoding': content_encoding,
+                'content-type': content_type,
+                'headers': headers or {},
+                'properties': properties or {}}
 
     def basic_publish(self, message, exchange, routing_key, **kwargs):
         """Publish message onto an exchange using a routing key.
@@ -1155,7 +1096,7 @@ class Channel(base.StdChannel):
         - wraps the body as a buffer object, so that
             :class:`qpid.messaging.endpoints.Sender` uses a content type
             that can support arbitrarily large messages.
-        - assigns a delivery_tag generated through self._delivery_tags
+        - sets delivery_tag to a random uuid.UUID
         - sets the exchange and routing_key info as delivery_info
 
         Internally uses :meth:`_put` to send the message synchronously. This
@@ -1182,7 +1123,7 @@ class Channel(base.StdChannel):
         props = message['properties']
         props.update(
             body_encoding=body_encoding,
-            delivery_tag=next(self._delivery_tags),
+            delivery_tag=uuid.uuid4(),
         )
         props['delivery_info'].update(
             exchange=exchange,
@@ -1262,7 +1203,7 @@ class Channel(base.StdChannel):
         qpid_exchange = self._broker.getExchange(exchange)
         if qpid_exchange:
             qpid_exchange_attributes = qpid_exchange.getAttributes()
-            return qpid_exchange_attributes["type"]
+            return qpid_exchange_attributes['type']
         else:
             return default
 
@@ -1334,7 +1275,6 @@ class Connection(object):
 
     # A class reference to the :class:`Channel` object
     Channel = Channel
-    SASL_ANONYMOUS_MECH = 'ANONYMOUS'
 
     def __init__(self, **connection_options):
         self.connection_options = connection_options
@@ -1343,55 +1283,43 @@ class Connection(object):
         self._qpid_conn = None
         establish = qpid.messaging.Connection.establish
 
-        # There is a behavior difference in qpid.messaging's sasl_mechanism
-        # selection method and cyrus-sasl's. The former will put PLAIN before
-        # ANONYMOUS if a username and password is given, but the latter will
-        # simply take whichever mech is listed first. Thus, if we had
-        # "ANONYMOUS PLAIN" as the default, the user would never be able to
-        # select PLAIN if cyrus-sasl was installed.
-
-        # The following code will put ANONYMOUS last in the mech list, and then
-        # try sasl mechs one by one. This should still result in secure
-        # behavior since it will select the first suitable mech. Unsuitable
-        # mechs will be rejected by the server.
-
-        sasl_mechanisms = [
-            x for x in connection_options['sasl_mechanisms'].split()
-            if x != self.SASL_ANONYMOUS_MECH
-        ]
-        if self.SASL_ANONYMOUS_MECH in \
-                connection_options['sasl_mechanisms'].split():
-            sasl_mechanisms.append(self.SASL_ANONYMOUS_MECH)
-
-        for sasl_mech in sasl_mechanisms:
-            try:
-                logger.debug(
-                    'Attempting to connect to qpid with SASL mechanism %s',
-                    sasl_mech,
-                )
-                modified_conn_opts = self.connection_options
-                modified_conn_opts['sasl_mechanisms'] = sasl_mech
-                self._qpid_conn = establish(**modified_conn_opts)
-                # connection was successful if we got this far
-                logger.info(
-                    'Connected to qpid with SASL mechanism %s', sasl_mech)
-                break
-            except ConnectionError as exc:
-                if self._is_unreachable_error(exc):
-                    logger.debug(E_UNREACHABLE, sasl_mech)
-                else:
-                    raise
-
-        if not self.get_qpid_connection():
-            logger.error(E_AUTH, sasl_mechanisms, exc_info=1)
-            raise AuthenticationFailure(sys.exc_info()[1])
-
-    def _is_unreachable_error(self, exc):
-        return (
-            getattr(exc, 'code', None) == 320 or
-            'Authentication failed' in exc.text or
-            'sasl negotiation failed: no mechanism agreed' in exc.text
-        )
+        # There are several inconsistent behaviors in the sasl libraries
+        # used on different systems. Although qpid.messaging allows
+        # multiple space separated sasl mechanisms, this implementation
+        # only advertises one type to the server. These are either
+        # ANONYMOUS, PLAIN, or an overridden value specified by the user.
+
+        sasl_mech = connection_options['sasl_mechanisms']
+
+        try:
+            msg = _('Attempting to connect to qpid with '
+                    'SASL mechanism %s') % sasl_mech
+            logger.debug(msg)
+            self._qpid_conn = establish(**self.connection_options)
+            # connection was successful if we got this far
+            msg = _('Connected to qpid with SASL '
+                    'mechanism %s') % sasl_mech
+            logger.info(msg)
+        except ConnectionError as conn_exc:
+            # if we get one of these errors, do not raise an exception.
+            # Raising will cause the connection to be retried. Instead,
+            # just continue on to the next mech.
+            coded_as_auth_failure = getattr(conn_exc, 'code', None) == 320
+            contains_auth_fail_text = \
+                'Authentication failed' in conn_exc.text
+            contains_mech_fail_text = \
+                'sasl negotiation failed: no mechanism agreed' \
+                in conn_exc.text
+            contains_mech_unavail_text = 'no mechanism available' \
+                in conn_exc.text
+            if coded_as_auth_failure or \
+                    contains_auth_fail_text or contains_mech_fail_text or \
+                    contains_mech_unavail_text:
+                msg = _('Unable to connect to qpid with SASL '
+                        'mechanism %s') % sasl_mech
+                logger.error(msg)
+                raise AuthenticationFailure(sys.exc_info()[1])
+            raise
 
     def get_qpid_connection(self):
         """Return the existing connection (singleton).
@@ -1409,7 +1337,7 @@ class Connection(object):
         receivers used by the Connection.
 
         """
-        return self._qpid_conn
+        self._qpid_conn.close()
 
     def close_channel(self, channel):
         """Close a Channel.
@@ -1429,88 +1357,6 @@ class Connection(object):
             channel.connection = None
 
 
-class ReceiversMonitor(threading.Thread):
-    """A monitoring thread that reads and handles messages from all receivers.
-
-    A single instance of ReceiversMonitor is expected to be created by
-    :class:`Transport`.
-
-    In :meth:`monitor_receivers`, the thread monitors all receivers
-    associated with the session created by the Transport using the blocking
-    call to session.next_receiver(). When any receiver has messages
-    available, a symbol '0' is written to the self._w_fd file descriptor. The
-    :meth:`monitor_receivers` is designed not to exit, and loops over
-    session.next_receiver() forever.
-
-    The entry point of the thread is :meth:`run` which calls
-    :meth:`monitor_receivers` and catches and logs all exceptions raised.
-    After an exception is logged, the method sleeps for 10 seconds, and
-    re-enters :meth:`monitor_receivers`
-
-    The thread is designed to be daemonized, and will be forcefully killed
-    when all non-daemon threads have already exited.
-
-    """
-
-    def __init__(self, session, w):
-        """Instantiate a ReceiversMonitor object
-
-        :param session: The session which needs all of its receivers
-            monitored.
-        :type session: :class:`qpid.messaging.endpoints.Session`
-        :param w: The file descriptor to write the '0' into when
-            next_receiver unblocks.
-        :type w: int
-
-        """
-        super(ReceiversMonitor, self).__init__()
-        self._session = session
-        self._w_fd = w
-
-    def run(self):
-        """Thread entry point for ReceiversMonitor
-
-        Calls :meth:`monitor_receivers` with a log-and-reenter behavior. This
-        guards against unexpected exceptions which could cause this thread to
-        exit unexpectedly.
-
-        If a recoverable error occurs, then the exception needs to be
-        propagated to the Main Thread where an exception handler can properly
-        handle it. An Exception is checked if it is recoverable, and if so,
-        it is stored as saved_exception on the self._session object. The
-        character 'e' is then written to the self.w_fd file descriptor
-        causing Main Thread to raise the saved exception. Once the Exception
-        info is saved and the file descriptor is written, this Thread
-        gracefully exits.
-
-        Typically recoverable errors are connection errors, and can be
-        recovered through a call to Transport.establish_connection which will
-        spawn a new ReceiversMonitor Thread.
-
-        """
-        while 1:
-            try:
-                self.monitor_receivers()
-            except Transport.connection_errors as exc:
-                self._session.saved_exception = exc
-                os.write(self._w_fd, 'e')
-                return
-            except Exception as exc:
-                logger.error(exc, exc_info=1)
-            time.sleep(10)
-
-    def monitor_receivers(self):
-        """Monitor all receivers, and write to _w_fd when a message is ready.
-
-        The call to next_receiver() blocks until a message is ready. Once a
-        message is ready, write a '0' to _w_fd.
-
-        """
-        while 1:
-            self._session.next_receiver()
-            os.write(self._w_fd, '0')
-
-
 class Transport(base.Transport):
     """Kombu native transport for a Qpid broker.
 
@@ -1533,20 +1379,23 @@ class Transport(base.Transport):
     The Transport can create :class:`Channel` objects to communicate with the
     broker with using the :meth:`create_channel` method.
 
-    The Transport identifies recoverable errors, allowing for error recovery
-    when certain exceptions occur. These exception types are stored in the
-    Transport class attribute connection_errors. This adds support for Kombu
-    to retry an operation if a ConnectionError occurs. ConnectionErrors occur
-    when the Transport cannot communicate with the Qpid broker.
+    The Transport identifies recoverable connection errors and recoverable
+    channel errors according to the Kombu 3.0 interface. These exception are
+    listed as tuples and store in the Transport class attribute
+    `recoverable_connection_errors` and `recoverable_channel_errors`
+    respectively. Any exception raised that is not a member of one of these
+    tuples is considered non-recoverable. This allows Kombu support for
+    automatic retry of certain operations to function correctly.
+
+    For backwards compatibility to the pre Kombu 3.0 exception interface, the
+    recoverable errors are also listed as `connection_errors` and
+    `channel_errors`.
 
     """
 
     # Reference to the class that should be used as the Connection object
     Connection = Connection
 
-    # The default port
-    default_port = DEFAULT_PORT
-
     # This Transport does not specify a polling interval.
     polling_interval = None
 
@@ -1557,33 +1406,75 @@ class Transport(base.Transport):
     driver_type = 'qpid'
     driver_name = 'qpid'
 
-    connection_errors = (
+    # Exceptions that can be recovered from, but where the connection must be
+    # closed and re-established first.
+    recoverable_connection_errors = (
         ConnectionError,
-        select.error
+        select.error,
+    )
+
+    # Exceptions that can be automatically recovered from without
+    # re-establishing the connection.
+    recoverable_channel_errors = (
+        NotFound,
     )
 
+    # Support the pre 3.0 Kombu exception labeling interface which treats
+    # connection_errors and channel_errors both as recoverable via a
+    # reconnect.
+    connection_errors = recoverable_connection_errors
+    channel_errors = recoverable_channel_errors
+
     def __init__(self, *args, **kwargs):
         self.verify_runtime_environment()
         super(Transport, self).__init__(*args, **kwargs)
-        self.r, self._w = os.pipe()
-        if fcntl is not None:
-            fcntl.fcntl(self.r, fcntl.F_SETFL, os.O_NONBLOCK)
+        self.use_async_interface = False
 
     def verify_runtime_environment(self):
         """Verify that the runtime environment is acceptable.
 
         This method is called as part of __init__ and raises a RuntimeError
-        in Python 3 or PyPi environments. This module is not compatible with
-        Python 3 or PyPi. The RuntimeError identifies this to the user up
-        front along with suggesting Python 2.7 be used instead.
+        in Python3 or PyPi environments. This module is not compatible with
+        Python3 or PyPi. The RuntimeError identifies this to the user up
+        front along with suggesting Python 2.6+ be used instead.
+
+        This method also checks that the dependencies qpidtoollibs and
+        qpid.messaging are installed. If either one is not installed a
+        RuntimeError is raised.
+
+        :raises: RuntimeError if the runtime environment is not acceptable.
 
         """
         if getattr(sys, 'pypy_version_info', None):
-            raise RuntimeError('The Qpid transport for Kombu does not '
-                               'support PyPy. Try using Python 2.7')
+            raise RuntimeError(
+                'The Qpid transport for Kombu does not '
+                'support PyPy. Try using Python 2.6+',
+            )
         if PY3:
-            raise RuntimeError('The Qpid transport for Kombu does not '
-                               'support Python 3. Try using Python 2.7')
+            raise RuntimeError(
+                'The Qpid transport for Kombu does not '
+                'support Python 3. Try using Python 2.6+',
+            )
+
+        if dependency_is_none(qpidtoollibs):
+            raise RuntimeError(
+                'The Python package "qpidtoollibs" is missing. Install it '
+                'with your package manager. You can also try `pip install '
+                'qpid-tools`.')
+
+        if dependency_is_none(qpid):
+            raise RuntimeError(
+                'The Python package "qpid.messaging" is missing. Install it '
+                'with your package manager. You can also try `pip install '
+                'qpid-python`.')
+
+    def _qpid_message_ready_handler(self, session):
+        if self.use_async_interface:
+            os.write(self._w, '0')
+
+    def _qpid_async_exception_notify_handler(self, obj_with_exception, exc):
+        if self.use_async_interface:
+            os.write(self._w, 'e')
 
     def on_readable(self, connection, loop):
         """Handle any messages associated with this Transport.
@@ -1591,17 +1482,13 @@ class Transport(base.Transport):
         This method clears a single message from the externally monitored
         file descriptor by issuing a read call to the self.r file descriptor
         which removes a single '0' character that was placed into the pipe
-        by :class:`ReceiversMonitor`. Once a '0' is read, all available
-        events are drained through a call to :meth:`drain_events`.
-
-        The behavior of self.r is adjusted in __init__ to be non-blocking,
-        ensuring that an accidental call to this method when no more messages
-        will arrive will not cause indefinite blocking.
+        by the Qpid session message callback handler. Once a '0' is read,
+        all available events are drained through a call to
+        :meth:`drain_events`.
 
-        If the self.r file descriptor returns the character 'e', a
-        recoverable error occurred in the background thread, and this thread
-        should raise the saved exception. The exception is stored as
-        saved_exception on the session object.
+        The file descriptor self.r is modified to be non-blocking, ensuring
+        that an accidental call to this method when no more messages will
+        not cause indefinite blocking.
 
         Nothing is expected to be returned from :meth:`drain_events` because
         :meth:`drain_events` handles messages by calling callbacks that are
@@ -1638,9 +1525,7 @@ class Transport(base.Transport):
         :type loop: kombu.async.Hub
 
         """
-        symbol = os.read(self.r, 1)
-        if symbol == 'e':
-            raise self.session.saved_exception
+        os.read(self.r, 1)
         try:
             self.drain_events(connection)
         except socket.timeout:
@@ -1652,7 +1537,7 @@ class Transport(base.Transport):
         Register the callback self.on_readable to be called when an
         external epoll loop sees that the file descriptor registered is
         ready for reading. The file descriptor is created by this Transport,
-        and is updated by the ReceiversMonitor thread.
+        and is written to when a message is available.
 
         Because supports_ev == True, Celery expects to call this method to
         give the Transport an opportunity to register a read file descriptor
@@ -1672,24 +1557,29 @@ class Transport(base.Transport):
         :type loop: kombu.async.hub.Hub
 
         """
+        self.r, self._w = os.pipe()
+        if fcntl is not None:
+            fcntl.fcntl(self.r, fcntl.F_SETFL, os.O_NONBLOCK)
+        self.use_async_interface = True
         loop.add_reader(self.r, self.on_readable, connection, loop)
 
     def establish_connection(self):
         """Establish a Connection object.
 
-        Determines the correct options to use when creating any connections
-        needed by this Transport, and create a :class:`Connection` object
-        which saves those values for connections generated as they are
-        needed. The options are a mixture of what is passed in through the
-        creator of the Transport, and the defaults provided by
+        Determines the correct options to use when creating any
+        connections needed by this Transport, and create a
+        :class:`Connection` object which saves those values for
+        connections generated as they are needed. The options are a
+        mixture of what is passed in through the creator of the
+        Transport, and the defaults provided by
         :meth:`default_connection_params`. Options cover broker network
         settings, timeout behaviors, authentication, and identity
         verification settings.
 
         This method also creates and stores a
         :class:`~qpid.messaging.endpoints.Session` using the
-        :class:`~qpid.messaging.endpoints.Connection` created by this method.
-        The Session is stored on self.
+        :class:`~qpid.messaging.endpoints.Connection` created by this
+        method. The Session is stored on self.
 
         :return: The created :class:`Connection` object is returned.
         :rtype: :class:`Connection`
@@ -1699,8 +1589,6 @@ class Transport(base.Transport):
         for name, default_value in items(self.default_connection_params):
             if not getattr(conninfo, name, None):
                 setattr(conninfo, name, default_value)
-        if conninfo.hostname == 'localhost':
-            conninfo.hostname = '127.0.0.1'
         if conninfo.ssl:
             conninfo.qpid_transport = 'ssl'
             conninfo.transport_options['ssl_keyfile'] = conninfo.ssl[
@@ -1715,19 +1603,51 @@ class Transport(base.Transport):
                 conninfo.transport_options['ssl_skip_hostname_check'] = True
         else:
             conninfo.qpid_transport = 'tcp'
-        opts = dict({'host': conninfo.hostname, 'port': conninfo.port,
-                     'username': conninfo.userid,
-                     'password': conninfo.password,
-                     'transport': conninfo.qpid_transport,
-                     'timeout': conninfo.connect_timeout,
-                     'sasl_mechanisms': conninfo.sasl_mechanisms},
-                    **conninfo.transport_options or {})
+
+        credentials = {}
+        if conninfo.login_method is None:
+            if conninfo.userid is not None and conninfo.password is not None:
+                sasl_mech = 'PLAIN'
+                credentials['username'] = conninfo.userid
+                credentials['password'] = conninfo.password
+            elif conninfo.userid is None and conninfo.password is not None:
+                raise Exception(
+                    'Password configured but no username. SASL PLAIN '
+                    'requires a username when using a password.')
+            elif conninfo.userid is not None and conninfo.password is None:
+                raise Exception(
+                    'Username configured but no password. SASL PLAIN '
+                    'requires a password when using a username.')
+            else:
+                sasl_mech = 'ANONYMOUS'
+        else:
+            sasl_mech = conninfo.login_method
+            if conninfo.userid is not None:
+                credentials['username'] = conninfo.userid
+
+        opts = {
+            'host': conninfo.hostname,
+            'port': conninfo.port,
+            'sasl_mechanisms': sasl_mech,
+            'timeout': conninfo.connect_timeout,
+            'transport': conninfo.qpid_transport
+        }
+
+        opts.update(credentials)
+        opts.update(conninfo.transport_options)
+
         conn = self.Connection(**opts)
         conn.client = self.client
         self.session = conn.get_qpid_connection().session()
-        monitor_thread = ReceiversMonitor(self.session, self._w)
-        monitor_thread.daemon = True
-        monitor_thread.start()
+        self.session.set_message_received_notify_handler(
+            self._qpid_message_ready_handler
+        )
+        conn.get_qpid_connection().set_async_exception_notify_handler(
+            self._qpid_async_exception_notify_handler
+        )
+        self.session.set_async_exception_notify_handler(
+            self._qpid_async_exception_notify_handler
+        )
         return conn
 
     def close_connection(self, connection):
@@ -1737,8 +1657,7 @@ class Transport(base.Transport):
         :type connection: :class:`kombu.transport.qpid.Connection`
 
         """
-        for channel in connection.channels:
-                channel.close()
+        connection.close()
 
     def drain_events(self, connection, timeout=0, **kwargs):
         """Handle and call callbacks for all ready Transport messages.
@@ -1801,18 +1720,21 @@ class Transport(base.Transport):
         These connection parameters will be used whenever the creator of
         Transport does not specify a required parameter.
 
-        NOTE: password is set to '' by default instead of None so the a
-        connection is attempted[1]. An empty password is considered valid for
-        qpidd if "auth=no" is set on the server.
-
-        [1] https://issues.apache.org/jira/browse/QPID-6109
-
         :return: A dict containing the default parameters.
         :rtype: dict
 
         """
         return {
-            'userid': 'guest', 'password': '',
-            'port': self.default_port, 'virtual_host': '',
-            'hostname': 'localhost', 'sasl_mechanisms': 'PLAIN ANONYMOUS',
+            'hostname': 'localhost',
+            'port': 5672,
         }
+
+    def __del__(self):
+        """Ensure file descriptors opened in __init__() are closed."""
+        if self.use_async_interface:
+            for fd in (self.r, self._w):
+                try:
+                    os.close(fd)
+                except OSError:
+                    # ignored
+                    pass
diff --git a/kombu/transport/qpid_patches.py b/kombu/transport/qpid_patches.py
deleted file mode 100644
index f4866372..00000000
--- a/kombu/transport/qpid_patches.py
+++ /dev/null
@@ -1,167 +0,0 @@
-# This module applies two patches to qpid.messaging that are required for
-# correct operation. Each patch fixes a bug. See links to the bugs below:
-# https://issues.apache.org/jira/browse/QPID-5637
-# https://issues.apache.org/jira/browse/QPID-5557
-
-# ## Begin Monkey Patch 1 ###
-# https://issues.apache.org/jira/browse/QPID-5637
-
-#############################################################################
-#  _   _  ___ _____ _____
-# | \ | |/ _ \_   _| ____|
-# |  \| | | | || | |  _|
-# | |\  | |_| || | | |___
-# |_| \_|\___/ |_| |_____|
-#
-# If you have code that also uses qpid.messaging and imports kombu,
-# or causes this file to be imported, then you need to make sure that this
-# import occurs first.
-#
-# Failure to do this will cause the following exception:
-# AttributeError: 'Selector' object has no attribute '_current_pid'
-#
-# Fix this by importing this module prior to using qpid.messaging in other
-# code that also uses this module.
-#############################################################################
-
-
-# this import is needed for Python 2.6. Without it, qpid.py will "mask" the
-# system's qpid lib
-from __future__ import absolute_import, unicode_literals
-
-import os
-
-
-# Imports for Monkey Patch 1
-try:
-    from qpid.selector import Selector
-except ImportError:  # pragma: no cover
-    Selector = None     # noqa
-import atexit
-
-
-# Prepare for Monkey Patch 1
-def default_monkey():  # pragma: no cover
-    Selector.lock.acquire()
-    try:
-        if Selector.DEFAULT is None:
-            sel = Selector()
-            atexit.register(sel.stop)
-            sel.start()
-            Selector.DEFAULT = sel
-            Selector._current_pid = os.getpid()
-        elif Selector._current_pid != os.getpid():
-            sel = Selector()
-            atexit.register(sel.stop)
-            sel.start()
-            Selector.DEFAULT = sel
-            Selector._current_pid = os.getpid()
-        return Selector.DEFAULT
-    finally:
-        Selector.lock.release()
-
-# Apply Monkey Patch 1
-
-try:
-    import qpid.selector
-    qpid.selector.Selector.default = staticmethod(default_monkey)
-except ImportError:  # pragma: no cover
-    pass
-
-# ## End Monkey Patch 1 ###
-
-# ## Begin Monkey Patch 2 ###
-# https://issues.apache.org/jira/browse/QPID-5557
-
-# Imports for Monkey Patch 2
-try:
-    from qpid.ops import ExchangeQuery, QueueQuery
-except ImportError:  # pragma: no cover
-    ExchangeQuery = None
-    QueueQuery = None
-
-try:
-    from qpid.messaging.exceptions import (
-        NotFound, AssertionFailed, ConnectionError,
-    )
-except ImportError:  # pragma: no cover
-    NotFound = None
-    AssertionFailed = None
-    ConnectionError = None
-
-
-# Prepare for Monkey Patch 2
-def resolve_declare_monkey(self, sst, lnk, dir, action):  # pragma: no cover
-    declare = lnk.options.get('create') in ('always', dir)
-    assrt = lnk.options.get('assert') in ('always', dir)
-    requested_type = lnk.options.get('node', {}).get('type')
-
-    def do_resolved(type, subtype):
-        err = None
-        if type is None:
-            if declare:
-                err = self.declare(sst, lnk, action)
-            else:
-                err = NotFound(text='no such queue: %s' % lnk.name)
-        else:
-            if assrt:
-                expected = lnk.options.get('node', {}).get('type')
-                if expected and type != expected:
-                    err = AssertionFailed(
-                        text='expected %s, got %s' % (expected, type))
-            if err is None:
-                action(type, subtype)
-        if err:
-            tgt = lnk.target
-            tgt.error = err
-            del self._attachments[tgt]
-            tgt.closed = True
-            return
-
-    self.resolve(sst, lnk.name, do_resolved, node_type=requested_type,
-                 force=declare)
-
-
-def resolve_monkey(self, sst, name, action, force=False,
-                   node_type=None):  # pragma: no cover
-    if not force and not node_type:
-        try:
-            type, subtype = self.address_cache[name]
-            action(type, subtype)
-            return
-        except KeyError:
-            pass
-    args = []
-
-    def do_result(r):
-        args.append(r)
-
-    def do_action(r):
-        do_result(r)
-        er, qr = args
-        if node_type == 'topic' and not er.not_found:
-            type, subtype = 'topic', er.type
-        elif node_type == 'queue' and qr.queue:
-            type, subtype = 'queue', None
-        elif er.not_found and not qr.queue:
-            type, subtype = None, None
-        elif qr.queue:
-            type, subtype = 'queue', None
-        else:
-            type, subtype = 'topic', er.type
-        if type is not None:
-            self.address_cache[name] = (type, subtype)
-        action(type, subtype)
-
-    sst.write_query(ExchangeQuery(name), do_result)
-    sst.write_query(QueueQuery(name), do_action)
-
-
-# Apply monkey patch 2
-try:
-    import qpid.messaging.driver
-    qpid.messaging.driver.Engine.resolve_declare = resolve_declare_monkey
-    qpid.messaging.driver.Engine.resolve = resolve_monkey
-except ImportError:  # pragma: no cover
-    pass
-# ## End Monkey Patch 2 ###