Repository: pika/pika

Issue:
Pika drops Connection.Close from broker before Connection.Open-Ok
When running in Fedora with RabbitMQ 3.6.0, I noticed that the test async_adapter_tests.TestZ_AccessDenied times out. This test attempts to open a connection using a non-existent vhost. In this scenario, RabbitMQ eventually closes the socket connection (about 30 seconds after Connection.Open with the non-existent vhost from the client).

It turns out that RabbitMQ 3.6.0 sends a Connection.Close immediately after receiving Connection.Open with bad vhost from the client. However, there is a bug in pika: pika doesn't register to handle Connection.Close from the broker until it receives Connection.Open-Ok, so pika just drops the Connection.Close frame in this scenario. Pika eventually detects that the connection is closed, subject to whenever RabbitMQ decides to close the connection.

pika needs to register its handler for Connection.Close before initiating the connection. `Connection._init_connection_state` might be a good place for it, right next to the call `self._add_connection_start_callback()`.

Discussion:
vitaly-krugl: Can anyone help me figure out why `assert_any_call` is failing in the python 2.6 build https://travis-ci.org/pika/pika/jobs/100035773 ?  Many thanks

CC @gst, @gmr 
gst: having a look into that..
gst: can't reproduce directly on my side..
gst: the only error I can trigger (in python2.6 but I think it's the same with others versions), is this one : 

```
Test that poll() is properly restarted after receiving EINTR error. ... FAIL

======================================================================
FAIL: Test that poll() is properly restarted after receiving EINTR error.
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/usr/local/lib/python2.7/dist-packages/mock.py", line 1201, in patched
    return func(*args, **keywargs)
  File "/home/gstarck/work/public/python/pika/tests/unit/select_connection_ioloop_tests.py", line 396, in test_eintr
    self.assertEqual(is_resumable_mock.call_count, 1)
AssertionError: 0 != 1
-------------------- >> begin captured logging << --------------------
pika.adapters.select_connection: DEBUG: Using SelectPoller
pika.adapters.select_connection: DEBUG: Using SelectPoller
pika.adapters.select_connection: DEBUG: Starting IOLoop
pika.adapters.select_connection: DEBUG: Stopping IOLoop
--------------------- >> end captured logging << ---------------------
```

I can trigger it, "simply" by executing many instances of the test at the same time, with something like this : 

`$ for i in $(seq 100) ; do ( nosetests -x tests/acceptance/blocking_adapter_test.py:TestUnroutableMessagesReturnedInNonPubackMode  &>/tmp/res$i || echo res$i failed) & done`

this also triggers quite a lot of Timed out errors which is expected given the load generated by the execution of so many tests in // ..
gst: strange, when you give a look at the debug output : 
```
root: DEBUG: ZZZ self.connection.callbacks.process.call_args_list: [call(0, '_on_connection_error', <pika.connection.Connection object at 0x2f92dd0>, <pika.connection.Connection object at 0x2f92dd0>, Client was disconnected at a connection stage indicating a probable denial of access to the specified virtual host: (1, 'error text')),
```
you see the mock has well been called with the, as far as i see, expected arguments.. 

strange..
gst: For refs/archive the actual error:
```
======================================================================
FAIL: on_disconnect invokes `ON_CONNECTION_ERROR` with `ProbableAccessDeniedError` and `ON_CONNECTION_CLOSED` callbacks
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/home/travis/build/pika/pika/tests/unit/connection_tests.py", line 228, in test_on_disconnect_invokes_access_on_connection_error_and_closed
    mock.ANY)
  File "/home/travis/virtualenv/python2.6.9/lib/python2.6/site-packages/mock/mock.py", line 999, in assert_any_call
    ), cause)
  File "/home/travis/virtualenv/python2.6.9/lib/python2.6/site-packages/six.py", line 718, in raise_from
    raise value
AssertionError: mock(0, '_on_connection_error', <pika.connection.Connection object at 0x2ec4dd0>, <pika.connection.Connection object at 0x2ec4dd0>, <ANY>) call not found
-------------------- >> begin captured logging << --------------------
pika.callback: DEBUG: Added: {'callback': <bound method Connection._on_connection_error of <pika.connection.Connection object at 0x2ec4dd0>>, 'only': None, 'one_shot': False, 'arguments': None}
pika.callback: DEBUG: Added: {'callback': <bound method Connection._on_connection_start of <pika.connection.Connection object at 0x2ec4dd0>>, 'only': None, 'one_shot': True, 'arguments': None, 'calls': 1}
pika.callback: DEBUG: Added: {'callback': <bound method Connection._on_connection_closed of <pika.connection.Connection object at 0x2ec4dd0>>, 'only': None, 'one_shot': True, 'arguments': None, 'calls': 1}
pika.connection: WARNING: Disconnected from RabbitMQ at localhost:5672 from_adapter=True (1): error text
pika.connection: ERROR: Socket closed while tuning the connection indicating a probable permission error when accessing a virtual host
pika.connection: ERROR: Connection setup failed due to Client was disconnected at a connection stage indicating a probable denial of access to the specified virtual host: (1, 'error text')
pika.callback: DEBUG: Incremented callback reference counter: {'callback': <bound method Connection._on_connection_start of <pika.connection.Connection object at 0x2ec4dd0>>, 'only': None, 'one_shot': True, 'arguments': None, 'calls': 2}
pika.callback: DEBUG: Incremented callback reference counter: {'callback': <bound method Connection._on_connection_closed of <pika.connection.Connection object at 0x2ec4dd0>>, 'only': None, 'one_shot': True, 'arguments': None, 'calls': 2}
root: DEBUG: ZZZ self.connection.callbacks.process.call_args_list: [call(0, '_on_connection_error', <pika.connection.Connection object at 0x2ec4dd0>, <pika.connection.Connection object at 0x2ec4dd0>, Client was disconnected at a connection stage indicating a probable denial of access to the specified virtual host: (1, 'error text')),
 call(0, '_on_connection_closed', <pika.connection.Connection object at 0x2ec4dd0>, <pika.connection.Connection object at 0x2ec4dd0>, 1, 'error text')]
--------------------- >> end captured logging << ---------------------
```

does it reproduce on travis ?
gst: could be due to mock version.. 
gst: @vitaly-krugl 
I'd like to know the mock version used by travis, how can I know ?

in fact: it would be good if before the execution of the tests there was a "pip freeze" executed, so that we can know all versions which are in use by travis..
vitaly-krugl: @gst, I was off the grid for a few days. Just got back today. Thanks for looking at this problem.

I don't know which version of mock it uses, but https://github.com/pika/pika/blob/master/test-requirements.txt specifies `mock` without a version. This would imply that it would load the most recent version. I like your idea about adding `pip freeze`. Perhaps this is something that could be added via https://github.com/pika/pika/blob/master/.travis.yml?
vitaly-krugl: @gst, I am able to reproduce these problems with mock 1.3.0 on python 2.7.10.

I've had nothing but trouble with the latest mock.
vitaly-krugl: @gst, getting rid of the spec arg in the mock allows the tests to pass now. I also switched to a context manager, but for patching, but I don't think that the context manager is important in this case. However, the spec/autospec/spec_set are really valuable, and it's a shame that I can't get the tests to pass using them.
gst: @vitaly-krugl : good to know about the reproduce :)

and yes for having pip freeze output it's effectively within `.travis.yml` ; could be put after the `  - pip install -r test-requirements.txt` in install section..
vitaly-krugl: @gst, would you like to submit a PR with the `pip freeze` change? thx
gst: here it is.. https://github.com/pika/pika/pull/689

Patch:
diff --git a/docs/version_history.rst b/docs/version_history.rst
index 349fbe7..06530b7 100644
--- a/docs/version_history.rst
+++ b/docs/version_history.rst
@@ -10,6 +10,8 @@ Next Release
  - In BaseConnection.close, call _handle_ioloop_stop only if the connection is
    already closed to allow the asynchronous close operation to complete
    gracefully.
+ - Pass error information from failed socket connection to user callbacks
+   on_open_error_callback and on_close_callback with result_code=-1.
 
 0.10.0 2015-09-02
 -----------------
diff --git a/pika/adapters/base_connection.py b/pika/adapters/base_connection.py
index 4bb436c..87afc99 100644
--- a/pika/adapters/base_connection.py
+++ b/pika/adapters/base_connection.py
@@ -10,7 +10,6 @@ import ssl
 
 import pika.compat
 from pika import connection
-from pika import exceptions
 
 try:
     SOL_TCP = socket.SOL_TCP
@@ -52,10 +51,10 @@ class BaseConnection(connection.Connection):
 
         :param pika.connection.Parameters parameters: Connection parameters
         :param method on_open_callback: Method to call on connection open
-        :param on_open_error_callback: Method to call if the connection cant
-                                       be opened
-        :type on_open_error_callback: method
-        :param method on_close_callback: Method to call on connection close
+        :param method on_open_error_callback: Called if the connection can't
+            be established: on_open_error_callback(connection, str|exception)
+        :param method on_close_callback: Called when the connection is closed:
+            on_close_callback(connection, reason_code, reason_text)
         :param object ioloop: IOLoop object to use
         :param bool stop_ioloop_on_close: Call ioloop.stop() if disconnected
         :raises: RuntimeError
@@ -152,38 +151,9 @@ class BaseConnection(connection.Connection):
     def _adapter_disconnect(self):
         """Invoked if the connection is being told to disconnect"""
         try:
-            self._remove_heartbeat()
             self._cleanup_socket()
-            self._check_state_on_disconnect()
         finally:
-            # Ensure proper cleanup since _check_state_on_disconnect may raise
-            # an exception
             self._handle_ioloop_stop()
-            self._init_connection_state()
-
-    def _check_state_on_disconnect(self):
-        """Checks to see if we were in opening a connection with RabbitMQ when
-        we were disconnected and raises exceptions for the anticipated
-        exception types.
-
-        """
-        if self.connection_state == self.CONNECTION_PROTOCOL:
-            LOGGER.error('Incompatible Protocol Versions')
-            raise exceptions.IncompatibleProtocolError
-        elif self.connection_state == self.CONNECTION_START:
-            LOGGER.error("Socket closed while authenticating indicating a "
-                         "probable authentication error")
-            raise exceptions.ProbableAuthenticationError
-        elif self.connection_state == self.CONNECTION_TUNE:
-            LOGGER.error("Socket closed while tuning the connection indicating "
-                         "a probable permission error when accessing a virtual "
-                         "host")
-            raise exceptions.ProbableAccessDeniedError
-        elif self.is_open:
-            LOGGER.warning("Socket closed when connection was open")
-        elif not self.is_closed and not self.is_closing:
-            LOGGER.warning('Unknown state on disconnect: %i',
-                           self.connection_state)
 
     def _cleanup_socket(self):
         """Close the socket cleanly"""
@@ -272,11 +242,14 @@ class BaseConnection(connection.Connection):
         """
         if not error_value:
             return None
+
         if hasattr(error_value, 'errno'):  # Python >= 2.6
             return error_value.errno
-        elif error_value is not None:
+        else:
+            # TODO: this doesn't look right; error_value.args[0] ??? Could
+            # probably remove this code path since pika doesn't test against
+            # Python 2.5
             return error_value[0]  # Python <= 2.5
-        return None
 
     def _flush_outbound(self):
         """Have the state manager schedule the necessary I/O.
@@ -291,21 +264,6 @@ class BaseConnection(connection.Connection):
         # called), etc., etc., etc.
         self._manage_event_state()
 
-    def _handle_disconnect(self):
-        """Called internally when the socket is disconnected already
-        """
-        try:
-            self._adapter_disconnect()
-        except (exceptions.ProbableAccessDeniedError,
-                exceptions.ProbableAuthenticationError) as error:
-            LOGGER.error('disconnected due to %r', error)
-            self.callbacks.process(0,
-                                   self.ON_CONNECTION_ERROR,
-                                   self,
-                                   self, error)
-
-        self._on_connection_closed(None, True)
-
     def _handle_ioloop_stop(self):
         """Invoked when the connection is closed to determine if the IOLoop
         should be stopped or not.
@@ -323,9 +281,10 @@ class BaseConnection(connection.Connection):
         :param int|object error_value: The inbound error
 
         """
-        if 'timed out' in str(error_value):
-            raise socket.timeout
+        # TODO: doesn't seem right: docstring defines error_value as int|object,
+        # but _get_error_code expects a falsie or an exception-like object
         error_code = self._get_error_code(error_value)
+
         if not error_code:
             LOGGER.critical("Tried to handle an error where no error existed")
             return
@@ -342,6 +301,8 @@ class BaseConnection(connection.Connection):
         elif self.params.ssl and isinstance(error_value, ssl.SSLError):
 
             if error_value.args[0] == ssl.SSL_ERROR_WANT_READ:
+                # TODO: doesn't seem right: this logic updates event state, but
+                # the logic at the bottom unconditionaly disconnects anyway.
                 self.event_state = self.READ
             elif error_value.args[0] == ssl.SSL_ERROR_WANT_WRITE:
                 self.event_state = self.WRITE
@@ -353,20 +314,21 @@ class BaseConnection(connection.Connection):
             LOGGER.error("Socket Error: %s", error_code)
 
         # Disconnect from our IOLoop and let Connection know what's up
-        self._handle_disconnect()
+        self._on_terminate(-1, repr(error_value))
 
     def _handle_timeout(self):
         """Handle a socket timeout in read or write.
         We don't do anything in the non-blocking handlers because we
         only have the socket in a blocking state during connect."""
-        pass
+        LOGGER.warning("Unexpected socket timeout")
 
     def _handle_events(self, fd, events, error=None, write_only=False):
         """Handle IO/Event loop events, processing them.
 
         :param int fd: The file descriptor for the events
         :param int events: Events from the IO/Event loop
-        :param int error: Was an error specified
+        :param int error: Was an error specified; TODO none of the current
+          adapters appear to be able to pass the `error` arg - is it needed?
         :param bool write_only: Only handle write events
 
         """
@@ -382,10 +344,11 @@ class BaseConnection(connection.Connection):
             self._handle_read()
 
         if (self.socket and write_only and (events & self.READ) and
-            (events & self.ERROR)):
-            LOGGER.error('BAD libc:  Write-Only but Read+Error. '
+                (events & self.ERROR)):
+            error_msg = ('BAD libc:  Write-Only but Read+Error. '
                          'Assume socket disconnected.')
-            self._handle_disconnect()
+            LOGGER.error(error_msg)
+            self._on_terminate(-1, error_msg)
 
         if self.socket and (events & self.ERROR):
             LOGGER.error('Error event %r, %r', events, error)
@@ -427,7 +390,7 @@ class BaseConnection(connection.Connection):
         # Empty data, should disconnect
         if not data or data == 0:
             LOGGER.error('Read empty data, calling disconnect')
-            return self._handle_disconnect()
+            return self._on_terminate(-1, "EOF")
 
         # Pass the data into our top level frame dispatching method
         self._on_data_available(data)
diff --git a/pika/adapters/blocking_connection.py b/pika/adapters/blocking_connection.py
index 1877dd6..f6881cd 100644
--- a/pika/adapters/blocking_connection.py
+++ b/pika/adapters/blocking_connection.py
@@ -394,7 +394,7 @@ class BlockingConnection(object):  # pylint: disable=R0902
                         returning true when it's time to stop processing.
                         Their results are OR'ed together.
         """
-        if self._impl.is_closed:
+        if self.is_closed:
             raise exceptions.ConnectionClosed()
 
         # Conditions for terminating the processing loop:
@@ -404,38 +404,35 @@ class BlockingConnection(object):  # pylint: disable=R0902
         #         OR
         #   empty outbound buffer and any waiter is ready
         is_done = (lambda:
-            self._closed_result.ready or
-            (not self._impl.outbound_buffer and
-             (not waiters or any(ready() for ready in  waiters))))
+                   self._closed_result.ready or
+                   (not self._impl.outbound_buffer and
+                    (not waiters or any(ready() for ready in  waiters))))
 
         # Process I/O until our completion condition is satisified
         while not is_done():
             self._impl.ioloop.poll()
             self._impl.ioloop.process_timeouts()
 
-        if self._closed_result.ready:
+        if self._open_error_result.ready or self._closed_result.ready:
             try:
-                result = self._closed_result.value
-                if result.reason_code not in [0, 200]:
-                    LOGGER.critical('Connection close detected; result=%r',
-                                    result)
-                    raise exceptions.ConnectionClosed(result.reason_code,
-                                                      result.reason_text)
-                elif not self._user_initiated_close:
-                    # NOTE: unfortunately, upon socket error, on_close_callback
-                    # presently passes reason_code=0, so we don't detect that as
-                    # an error
+                if not self._user_initiated_close:
                     if self._open_error_result.ready:
                         maybe_exception = self._open_error_result.value.error
-                        LOGGER.critical('Connection open failed - %r',
-                                        maybe_exception)
+                        LOGGER.error('Connection open failed - %r',
+                                     maybe_exception)
                         if isinstance(maybe_exception, Exception):
                             raise maybe_exception
-
-                    LOGGER.critical('Connection close detected')
-                    raise exceptions.ConnectionClosed()
+                        else:
+                            raise exceptions.ConnectionClosed(maybe_exception)
+                    else:
+                        result = self._closed_result.value
+                        LOGGER.error('Connection close detected; result=%r',
+                                     result)
+                        raise exceptions.ConnectionClosed(result.reason_code,
+                                                          result.reason_text)
                 else:
-                    LOGGER.info('Connection closed; result=%r', result)
+                    LOGGER.info('Connection closed; result=%r',
+                                self._closed_result.value)
             finally:
                 self._cleanup()
 
@@ -732,7 +729,8 @@ class BlockingConnection(object):  # pylint: disable=R0902
     @property
     def is_closing(self):
         """
-        Returns a boolean reporting the current connection state.
+        Returns True if connection is in the process of closing due to
+        client-initiated `close` request, but closing is not yet complete.
         """
         return self._impl.is_closing
 
@@ -1143,7 +1141,8 @@ class BlockingChannel(object):  # pylint: disable=R0904,R0902
 
     @property
     def is_closing(self):
-        """Returns True if the channel is closing.
+        """Returns True if client-initiated closing of the channel is in
+        progress.
 
         :rtype: bool
 
@@ -1173,7 +1172,7 @@ class BlockingChannel(object):  # pylint: disable=R0904,R0902
                         returning true when it's time to stop processing.
                         Their results are OR'ed together.
         """
-        if self._impl.is_closed:
+        if self.is_closed:
             raise exceptions.ChannelClosed()
 
         if not waiters:
diff --git a/pika/adapters/libev_connection.py b/pika/adapters/libev_connection.py
index ed3ec81..ce491c9 100644
--- a/pika/adapters/libev_connection.py
+++ b/pika/adapters/libev_connection.py
@@ -84,9 +84,10 @@ class LibevConnection(BaseConnection):
         :param pika.connection.Parameters parameters: Connection parameters
         :param on_open_callback: The method to call when the connection is open
         :type on_open_callback: method
-        :param on_open_error_callback: Method to call if the connection cannot
-                                       be opened
-        :type on_open_error_callback: method
+        :param method on_open_error_callback: Called if the connection can't
+            be established: on_open_error_callback(connection, str|exception)
+        :param method on_close_callback: Called when the connection is closed:
+            on_close_callback(connection, reason_code, reason_text)
         :param bool stop_ioloop_on_close: Call ioloop.stop() if disconnected
         :param custom_ioloop: Override using the default IOLoop in libev
         :param on_signal_callback: Method to call if SIGINT or SIGTERM occur
diff --git a/pika/adapters/select_connection.py b/pika/adapters/select_connection.py
index 64e2bbe..645cb38 100644
--- a/pika/adapters/select_connection.py
+++ b/pika/adapters/select_connection.py
@@ -75,10 +75,10 @@ class SelectConnection(BaseConnection):
 
         :param pika.connection.Parameters parameters: Connection parameters
         :param method on_open_callback: Method to call on connection open
-        :param on_open_error_callback: Method to call if the connection cant
-                                       be opened
-        :type on_open_error_callback: method
-        :param method on_close_callback: Method to call on connection close
+        :param method on_open_error_callback: Called if the connection can't
+            be established: on_open_error_callback(connection, str|exception)
+        :param method on_close_callback: Called when the connection is closed:
+            on_close_callback(connection, reason_code, reason_text)
         :param bool stop_ioloop_on_close: Call ioloop.stop() if disconnected
         :param custom_ioloop: Override using the global IOLoop in Tornado
         :raises: RuntimeError
diff --git a/pika/adapters/tornado_connection.py b/pika/adapters/tornado_connection.py
index 1c5c607..ce407d1 100644
--- a/pika/adapters/tornado_connection.py
+++ b/pika/adapters/tornado_connection.py
@@ -39,9 +39,10 @@ class TornadoConnection(base_connection.BaseConnection):
         :param pika.connection.Parameters parameters: Connection parameters
         :param on_open_callback: The method to call when the connection is open
         :type on_open_callback: method
-        :param on_open_error_callback: Method to call if the connection cant
-                                       be opened
-        :type on_open_error_callback: method
+        :param method on_open_error_callback: Called if the connection can't
+            be established: on_open_error_callback(connection, str|exception)
+        :param method on_close_callback: Called when the connection is closed:
+            on_close_callback(connection, reason_code, reason_text)
         :param bool stop_ioloop_on_close: Call ioloop.stop() if disconnected
         :param custom_ioloop: Override using the global IOLoop in Tornado
 
@@ -55,7 +56,7 @@ class TornadoConnection(base_connection.BaseConnection):
 
     def _adapter_connect(self):
         """Connect to the remote socket, adding the socket to the IOLoop if
-        connected. 
+        connected.
 
         :rtype: bool
 
diff --git a/pika/adapters/twisted_connection.py b/pika/adapters/twisted_connection.py
index 2ee65b2..62e595c 100644
--- a/pika/adapters/twisted_connection.py
+++ b/pika/adapters/twisted_connection.py
@@ -105,6 +105,9 @@ class TwistedChannel(object):
 
         try:
             consumer_tag = self.__channel.basic_consume(*args, **kwargs)
+        # TODO this except without types would suppress system-exiting
+        # exceptions, such as SystemExit and KeyboardInterrupt. It should be at
+        # least `except Exception` and preferably more specific.
         except:
             return defer.fail()
 
@@ -163,6 +166,9 @@ class TwistedChannel(object):
 
             try:
                 method(*args, **kwargs)
+            # TODO this except without types would suppress system-exiting
+            # exceptions, such as SystemExit and KeyboardInterrupt. It should be
+            # at least `except Exception` and preferably more specific.
             except:
                 return defer.fail()
             return d
@@ -300,13 +306,6 @@ class TwistedConnection(base_connection.BaseConnection):
         self.ioloop.remove_handler(None)
         self._cleanup_socket()
 
-    def _handle_disconnect(self):
-        """Do not stop the reactor, this would cause the entire process to exit,
-        just fire the disconnect callbacks
-
-        """
-        self._on_connection_closed(None, True)
-
     def _on_connected(self):
         """Call superclass and then update the event state to flush the outgoing
         frame out. Commit 50d842526d9f12d32ad9f3c4910ef60b8c301f59 removed a
@@ -339,7 +338,7 @@ class TwistedConnection(base_connection.BaseConnection):
         if not reason.check(error.ConnectionDone):
             log.err(reason)
 
-        self._handle_disconnect()
+        self._on_terminate(-1, str(reason))
 
     def doRead(self):
         self._handle_read()
diff --git a/pika/channel.py b/pika/channel.py
index 5c67c49..4af9a6e 100644
--- a/pika/channel.py
+++ b/pika/channel.py
@@ -29,7 +29,7 @@ class Channel(object):
     CLOSED = 0
     OPENING = 1
     OPEN = 2
-    CLOSING = 3
+    CLOSING = 3  # client-initiated close in progress
 
     _ON_CHANNEL_CLEANUP_CB_KEY = '_on_channel_cleanup'
 
@@ -615,7 +615,8 @@ class Channel(object):
 
     @property
     def is_closing(self):
-        """Returns True if the channel is closing.
+        """Returns True if client-initiated closing of the channel is in
+        progress.
 
         :rtype: bool
 
diff --git a/pika/connection.py b/pika/connection.py
index 6f59cd0..6288628 100644
--- a/pika/connection.py
+++ b/pika/connection.py
@@ -4,6 +4,7 @@ import sys
 import collections
 import logging
 import math
+import numbers
 import platform
 import threading
 import warnings
@@ -586,7 +587,7 @@ class Connection(object):
     CONNECTION_START = 3
     CONNECTION_TUNE = 4
     CONNECTION_OPEN = 5
-    CONNECTION_CLOSING = 6
+    CONNECTION_CLOSING = 6  # client-initiated close in progress
 
     def __init__(self,
                  parameters=None,
@@ -602,9 +603,10 @@ class Connection(object):
 
         :param pika.connection.Parameters parameters: Connection parameters
         :param method on_open_callback: Called when the connection is opened
-        :param method on_open_error_callback: Called if the connection cant
-                                       be opened
-        :param method on_close_callback: Called when the connection is closed
+        :param method on_open_error_callback: Called if the connection can't
+            be established: on_open_error_callback(connection, str|exception)
+        :param method on_close_callback: Called when the connection is closed:
+            on_close_callback(connection, reason_code, reason_text)
 
         """
         self._write_lock = threading.Lock()
@@ -770,8 +772,9 @@ class Connection(object):
         self.remaining_connection_attempts -= 1
         LOGGER.warning('Could not connect, %i attempts left',
                        self.remaining_connection_attempts)
-        if self.remaining_connection_attempts:
+        if self.remaining_connection_attempts > 0:
             LOGGER.info('Retrying in %i seconds', self.params.retry_delay)
+            # TODO: remove timeout if connection is closed before timer fires
             self.add_timeout(self.params.retry_delay, self.connect)
         else:
             self.callbacks.process(0, self.ON_CONNECTION_ERROR, self, self,
@@ -813,7 +816,8 @@ class Connection(object):
     @property
     def is_closing(self):
         """
-        Returns a boolean reporting the current connection state.
+        Returns True if connection is in the process of closing due to
+        client-initiated `close` request, but closing is not yet complete.
         """
         return self.connection_state == self.CONNECTION_CLOSING
 
@@ -1160,6 +1164,13 @@ class Connection(object):
         # Our starting point once connected, first frame received
         self._add_connection_start_callback()
 
+        # Add a callback handler for the Broker telling us to disconnect.
+        # NOTE: As of RabbitMQ 3.6.0, RabbitMQ broker may send Connection.Close
+        # to signal error during connection setup (and wait a longish time
+        # before closing the TCP/IP stream). Earlier RabbitMQ versions
+        # simply closed the TCP/IP stream.
+        self.callbacks.add(0, spec.Connection.Close, self._on_connection_close)
+
     def _is_basic_deliver_frame(self, frame_value):
         """Returns true if the frame is a Basic.Deliver
 
@@ -1169,17 +1180,6 @@ class Connection(object):
         """
         return isinstance(frame_value, spec.Basic.Deliver)
 
-    def _is_connection_close_frame(self, value):
-        """Returns true if the frame is a Connection.Close frame.
-
-        :param pika.frame.Method value: The frame to check
-        :rtype: bool
-
-        """
-        if not value:
-            return False
-        return isinstance(value.method, spec.Connection.Close)
-
     def _is_method_frame(self, value):
         """Returns true if the frame is a method frame.
 
@@ -1250,32 +1250,29 @@ class Connection(object):
         # Start the communication with the RabbitMQ Broker
         self._send_frame(frame.ProtocolHeader())
 
-    def _on_connection_closed(self, method_frame, from_adapter=False):
-        """Called when the connection is closed remotely. The from_adapter value
-        will be true if the connection adapter has been disconnected from
-        the broker and the method was invoked directly instead of by receiving
-        a Connection.Close frame.
+    def _on_connection_close(self, method_frame):
+        """Called when the connection is closed remotely via Connection.Close
+        frame from broker.
 
-        :param pika.frame.Method: The Connection.Close frame
-        :param bool from_adapter: Called by the connection adapter
+        :param pika.frame.Method method_frame: The Connection.Close frame
 
         """
-        if method_frame and self._is_connection_close_frame(method_frame):
-            self.closing = (method_frame.method.reply_code,
-                            method_frame.method.reply_text)
+        LOGGER.debug('_on_connection_close: frame=%s', method_frame)
 
-        # Save the codes because self.closing gets reset by _adapter_disconnect
-        reply_code, reply_text = self.closing
+        self.closing = (method_frame.method.reply_code,
+                        method_frame.method.reply_text)
 
-        # Stop the heartbeat checker if it exists
-        self._remove_heartbeat()
+        self._on_terminate(self.closing[0], self.closing[1])
 
-        # If this did not come from the connection adapter, close the socket
-        if not from_adapter:
-            self._adapter_disconnect()
+    def _on_connection_close_ok(self, method_frame):
+        """Called when Connection.CloseOk is received from remote.
 
-        # Invoke a method frame neutral close
-        self._on_disconnect(reply_code, reply_text)
+        :param pika.frame.Method method_frame: The Connection.CloseOk frame
+
+        """
+        LOGGER.debug('_on_connection_close_ok: frame=%s', method_frame)
+
+        self._on_terminate(self.closing[0], self.closing[1])
 
     def _on_connection_error(self, connection_unused, error_message=None):
         """Default behavior when the connecting connection can not connect.
@@ -1294,9 +1291,6 @@ class Connection(object):
         """
         self.known_hosts = method_frame.method.known_hosts
 
-        # Add a callback handler for the Broker telling us to disconnect
-        self.callbacks.add(0, spec.Connection.Close, self._on_connection_closed)
-
         # We're now connected at the AMQP level
         self._set_connection_state(self.CONNECTION_OPEN)
 
@@ -1368,27 +1362,89 @@ class Connection(object):
             self._trim_frame_buffer(consumed_count)
             self._process_frame(frame_value)
 
-    def _on_disconnect(self, reply_code, reply_text):
-        """Invoke passing in the reply_code and reply_text from internal
-        methods to the adapter. Called from on_connection_closed and Heartbeat
-        timeouts.
-
-        :param str reply_code: The numeric close code
-        :param str reply_text: The text close reason
+    def _on_terminate(self, reason_code, reason_text):
+        """Terminate the connection and notify registered ON_CONNECTION_ERROR
+        and/or ON_CONNECTION_CLOSED callbacks
 
+        :param integer reason_code: HTTP error code for AMQP-reported closures
+          or -1 for other errors (such as socket errors)
+        :param str reason_text: human-readable text message describing the error
         """
-        LOGGER.warning('Disconnected from RabbitMQ at %s:%i (%s): %s',
-                       self.params.host, self.params.port, reply_code,
-                       reply_text)
+        LOGGER.warning(
+            'Disconnected from RabbitMQ at %s:%i (%s): %s',
+            self.params.host, self.params.port, reason_code,
+            reason_text)
+
+        if not isinstance(reason_code, numbers.Integral):
+            raise TypeError('reason_code must be an integer, but got %r'
+                            % (reason_code,))
+
+        # Stop the heartbeat checker if it exists
+        self._remove_heartbeat()
+
+        # Remove connection management callbacks
+        # TODO: This call was moved here verbatim from legacy code and the
+        # following doesn't seem to be right: `Connection.Open` here is
+        # unexpected, we don't appear to ever register it, and the broker
+        # shouldn't be sending `Connection.Open` to us, anyway.
+        self._remove_callbacks(0, [spec.Connection.Close, spec.Connection.Start,
+                                   spec.Connection.Open])
+
+        # Close the socket
+        self._adapter_disconnect()
+
+        # Determine whether this was an error during connection setup
+        connection_error = None
+
+        if self.connection_state == self.CONNECTION_PROTOCOL:
+            LOGGER.error('Incompatible Protocol Versions')
+            connection_error = exceptions.IncompatibleProtocolError(
+                reason_code,
+                reason_text)
+        elif self.connection_state == self.CONNECTION_START:
+            LOGGER.error('Connection closed while authenticating indicating a '
+                         'probable authentication error')
+            connection_error = exceptions.ProbableAuthenticationError(
+                reason_code,
+                reason_text)
+        elif self.connection_state == self.CONNECTION_TUNE:
+            LOGGER.error('Connection closed while tuning the connection '
+                         'indicating a probable permission error when '
+                         'accessing a virtual host')
+            connection_error = exceptions.ProbableAccessDeniedError(
+                reason_code,
+                reason_text)
+        elif self.connection_state not in [self.CONNECTION_OPEN,
+                                           self.CONNECTION_CLOSED,
+                                           self.CONNECTION_CLOSING]:
+            LOGGER.warning('Unexpected connection state on disconnect: %i',
+                           self.connection_state)
+
+        # Transition to closed state
         self._set_connection_state(self.CONNECTION_CLOSED)
+
+        # Inform our channel proxies
         for channel in dictkeys(self._channels):
             if channel not in self._channels:
                 continue
-            method_frame = frame.Method(channel, spec.Channel.Close(reply_code,
-                                                                    reply_text))
+            method_frame = frame.Method(channel, spec.Channel.Close(
+                reason_code,
+                reason_text))
             self._channels[channel]._on_close(method_frame)
-        self._process_connection_closed_callbacks(reply_code, reply_text)
-        self._remove_connection_callbacks()
+
+        # Inform interested parties
+        if connection_error is not None:
+            LOGGER.error('Connection setup failed due to %r', connection_error)
+            self.callbacks.process(0,
+                                   self.ON_CONNECTION_ERROR,
+                                   self, self,
+                                   connection_error)
+
+        self.callbacks.process(0, self.ON_CONNECTION_CLOSED, self, self,
+                               reason_code, reason_text)
+
+        # Reset connection properties
+        self._init_connection_state()
 
     def _process_callbacks(self, frame_value):
         """Process the callbacks for the frame if the frame is a method frame
@@ -1407,17 +1463,6 @@ class Connection(object):
             return True
         return False
 
-    def _process_connection_closed_callbacks(self, reason_code, reason_text):
-        """Process any callbacks that should be called when the connection is
-        closed.
-
-        :param str reason_code: The numeric code from RabbitMQ for the close
-        :param str reason_text: The text reason fro closing
-
-        """
-        self.callbacks.process(0, self.ON_CONNECTION_CLOSED, self, self,
-                               reason_code, reason_text)
-
     def _process_frame(self, frame_value):
         """Process an inbound frame from the socket.
 
@@ -1489,11 +1534,6 @@ class Connection(object):
         for method_frame in method_frames:
             self._remove_callback(channel_number, method_frame)
 
-    def _remove_connection_callbacks(self):
-        """Remove all callbacks for the connection"""
-        self._remove_callbacks(0, [spec.Connection.Close, spec.Connection.Start,
-                                   spec.Connection.Open])
-
     def _rpc(self, channel_number, method_frame,
              callback_method=None,
              acceptable_replies=None):
@@ -1530,7 +1570,7 @@ class Connection(object):
 
         """
         self._rpc(0, spec.Connection.Close(reply_code, reply_text, 0, 0),
-                  self._on_connection_closed, [spec.Connection.CloseOk])
+                  self._on_connection_close_ok, [spec.Connection.CloseOk])
 
     def _send_connection_open(self):
         """Send a Connection.Open frame"""
diff --git a/pika/exceptions.py b/pika/exceptions.py
index c56f6a0..f219fbb 100644
--- a/pika/exceptions.py
+++ b/pika/exceptions.py
@@ -26,7 +26,8 @@ class AMQPConnectionError(AMQPError):
 class IncompatibleProtocolError(AMQPConnectionError):
 
     def __repr__(self):
-        return 'The protocol returned by the server is not supported'
+        return ('The protocol returned by the server is not supported: %s' %
+                (self.args,))
 
 
 class AuthenticationError(AMQPConnectionError):
@@ -40,14 +41,15 @@ class ProbableAuthenticationError(AMQPConnectionError):
 
     def __repr__(self):
         return ('Client was disconnected at a connection stage indicating a '
-                'probable authentication error')
+                'probable authentication error: %s' % (self.args,))
 
 
 class ProbableAccessDeniedError(AMQPConnectionError):
 
     def __repr__(self):
         return ('Client was disconnected at a connection stage indicating a '
-                'probable denial of access to the specified virtual host')
+                'probable denial of access to the specified virtual host: %s' %
+                (self.args,))
 
 
 class NoFreeChannels(AMQPConnectionError):
diff --git a/pika/heartbeat.py b/pika/heartbeat.py
index bbd5c1f..64026cc 100644
--- a/pika/heartbeat.py
+++ b/pika/heartbeat.py
@@ -115,10 +115,14 @@ class HeartbeatChecker(object):
                     self._idle_byte_intervals)
         duration = self._max_idle_count * self._interval
         text = HeartbeatChecker._STALE_CONNECTION % duration
+
+        # NOTE: this won't achieve the perceived effect of sending
+        # Connection.Close to broker, because the frame will only get buffered
+        # in memory before the next statement terminates the connection.
         self._connection.close(HeartbeatChecker._CONNECTION_FORCED, text)
-        self._connection._adapter_disconnect()
-        self._connection._on_disconnect(HeartbeatChecker._CONNECTION_FORCED,
-                                        text)
+
+        self._connection._on_terminate(HeartbeatChecker._CONNECTION_FORCED,
+                                       text)
 
     @property
     def _has_received_data(self):


Test patch:
diff --git a/tests/unit/blocking_connection_tests.py b/tests/unit/blocking_connection_tests.py
index 70dc065..2888525 100644
--- a/tests/unit/blocking_connection_tests.py
+++ b/tests/unit/blocking_connection_tests.py
@@ -3,6 +3,10 @@
 Tests for pika.adapters.blocking_connection.BlockingConnection
 
 """
+
+# Suppress pylint warnings concering access to protected member
+# pylint: disable=W0212
+
 import socket
 
 from pika.exceptions import AMQPConnectionError
@@ -26,11 +30,11 @@ class BlockingConnectionMockTemplate(blocking_connection.BlockingConnection):
     pass
 
 class SelectConnectionTemplate(blocking_connection.SelectConnection):
-    is_closed = False
-    is_closing = False
-    is_open = True
-    outbound_buffer = []
-    _channels = dict()
+    is_closed = None
+    is_closing = None
+    is_open = None
+    outbound_buffer = None
+    _channels = None
 
 
 class BlockingConnectionTests(unittest.TestCase):
@@ -41,7 +45,7 @@ class BlockingConnectionTests(unittest.TestCase):
     def test_constructor(self, select_connection_class_mock):
         with mock.patch.object(blocking_connection.BlockingConnection,
                                '_process_io_for_connection_setup'):
-            connection = blocking_connection.BlockingConnection('params')
+            blocking_connection.BlockingConnection('params')
 
         select_connection_class_mock.assert_called_once_with(
             parameters='params',
@@ -153,9 +157,7 @@ class BlockingConnectionTests(unittest.TestCase):
         with self.assertRaises(pika.exceptions.ConnectionClosed) as cm:
             connection._flush_output(lambda: False, lambda: True)
 
-        self.assertSequenceEqual(
-            cm.exception.args,
-            ())
+        self.assertSequenceEqual(cm.exception.args, (200, 'ok'))
 
     @patch.object(blocking_connection, 'SelectConnection',
                   spec_set=SelectConnectionTemplate)
@@ -190,7 +192,7 @@ class BlockingConnectionTests(unittest.TestCase):
                 blocking_connection.BlockingConnection,
                 '_flush_output',
                 spec_set=blocking_connection.BlockingConnection._flush_output):
-            channel = connection.channel()
+            connection.channel()
 
     @patch.object(blocking_connection, 'SelectConnection',
                   spec_set=SelectConnectionTemplate)
diff --git a/tests/unit/connection_tests.py b/tests/unit/connection_tests.py
index 7ef7afe..ed8b038 100644
--- a/tests/unit/connection_tests.py
+++ b/tests/unit/connection_tests.py
@@ -2,6 +2,18 @@
 Tests for pika.connection.Connection
 
 """
+
+# Suppress pylint warnings concerning access to protected member
+# pylint: disable=W0212
+
+# Suppress pylint messages concerning missing docstrings
+# pylint: disable=C0111
+
+# Suppress pylint messages concerning invalid method name
+# pylint: disable=C0103
+
+
+
 try:
     import mock
 except ImportError:
@@ -17,6 +29,7 @@ except ImportError:
 from pika import connection
 from pika import channel
 from pika import credentials
+from pika import exceptions
 from pika import frame
 from pika import spec
 from pika.compat import xrange, urlencode
@@ -97,15 +110,128 @@ class ConnectionTests(unittest.TestCase):
         self.assertFalse(on_close_ready.called,
                          '_on_close_ready should not have been called')
 
-    def test_on_disconnect(self):
-        """if connection isn't closing _on_close_ready should not be called"""
-        self.connection._on_disconnect(0, 'Undefined')
+    def test_on_terminate_cleans_up(self):
+        """_on_terminate cleans up heartbeat, adapter, and channels"""
+        heartbeat = mock.Mock()
+        self.connection.heartbeat = heartbeat
+        self.connection._adapter_disconnect = mock.Mock()
+
+        self.connection._on_terminate(0, 'Undefined')
+
+        heartbeat.stop.assert_called_once_with()
+        self.connection._adapter_disconnect.assert_called_once_with()
+
         self.assertTrue(self.channel._on_close.called,
                         'channel._on_close should have been called')
         method_frame = self.channel._on_close.call_args[0][0]
         self.assertEqual(method_frame.method.reply_code, 0)
         self.assertEqual(method_frame.method.reply_text, 'Undefined')
 
+        self.assertTrue(self.connection.is_closed)
+
+    def test_on_terminate_invokes_connection_closed_callback(self):
+        """_on_terminate invokes `Connection.ON_CONNECTION_CLOSED` callbacks"""
+        self.connection.callbacks.process = mock.Mock(
+            wraps=self.connection.callbacks.process)
+
+        self.connection._adapter_disconnect = mock.Mock()
+
+        self.connection._on_terminate(1, 'error text')
+
+        self.connection.callbacks.process.assert_called_once_with(
+            0, self.connection.ON_CONNECTION_CLOSED,
+            self.connection, self.connection,
+            1, 'error text')
+
+        with self.assertRaises(AssertionError):
+            self.connection.callbacks.process.assert_any_call(
+                0, self.connection.ON_CONNECTION_ERROR,
+                self.connection, self.connection,
+                mock.ANY)
+
+    def test_on_terminate_invokes_protocol_on_connection_error_and_closed(self):
+        """_on_terminate invokes `ON_CONNECTION_ERROR` with `IncompatibleProtocolError` and `ON_CONNECTION_CLOSED` callbacks"""
+        with mock.patch.object(self.connection.callbacks, 'process'):
+
+            self.connection._adapter_disconnect = mock.Mock()
+
+            self.connection._set_connection_state(
+                self.connection.CONNECTION_PROTOCOL)
+
+            self.connection._on_terminate(1, 'error text')
+
+            self.assertEqual(self.connection.callbacks.process.call_count, 2)
+
+            self.connection.callbacks.process.assert_any_call(
+                0, self.connection.ON_CONNECTION_ERROR,
+                self.connection, self.connection,
+                mock.ANY)
+
+            conn_exc = self.connection.callbacks.process.call_args_list[0][0][4]
+            self.assertIs(type(conn_exc), exceptions.IncompatibleProtocolError)
+            self.assertSequenceEqual(conn_exc.args, [1, 'error text'])
+
+            self.connection.callbacks.process.assert_any_call(
+                0, self.connection.ON_CONNECTION_CLOSED,
+                self.connection, self.connection,
+                1, 'error text')
+
+    def test_on_terminate_invokes_auth_on_connection_error_and_closed(self):
+        """_on_terminate invokes `ON_CONNECTION_ERROR` with `ProbableAuthenticationError` and `ON_CONNECTION_CLOSED` callbacks"""
+        with mock.patch.object(self.connection.callbacks, 'process'):
+
+            self.connection._adapter_disconnect = mock.Mock()
+
+            self.connection._set_connection_state(
+                self.connection.CONNECTION_START)
+
+            self.connection._on_terminate(1, 'error text')
+
+            self.assertEqual(self.connection.callbacks.process.call_count, 2)
+
+            self.connection.callbacks.process.assert_any_call(
+                0, self.connection.ON_CONNECTION_ERROR,
+                self.connection, self.connection,
+                mock.ANY)
+
+            conn_exc = self.connection.callbacks.process.call_args_list[0][0][4]
+            self.assertIs(type(conn_exc),
+                          exceptions.ProbableAuthenticationError)
+            self.assertSequenceEqual(conn_exc.args, [1, 'error text'])
+
+            self.connection.callbacks.process.assert_any_call(
+                0, self.connection.ON_CONNECTION_CLOSED,
+                self.connection, self.connection,
+                1, 'error text')
+
+    def test_on_terminate_invokes_access_denied_on_connection_error_and_closed(
+            self):
+        """_on_terminate invokes `ON_CONNECTION_ERROR` with `ProbableAccessDeniedError` and `ON_CONNECTION_CLOSED` callbacks"""
+        with mock.patch.object(self.connection.callbacks, 'process'):
+
+            self.connection._adapter_disconnect = mock.Mock()
+
+            self.connection._set_connection_state(
+                self.connection.CONNECTION_TUNE)
+
+            self.connection._on_terminate(1, 'error text')
+
+            self.assertEqual(self.connection.callbacks.process.call_count, 2)
+
+            self.connection.callbacks.process.assert_any_call(
+                0, self.connection.ON_CONNECTION_ERROR,
+                self.connection, self.connection,
+                mock.ANY)
+
+            conn_exc = self.connection.callbacks.process.call_args_list[0][0][4]
+            self.assertIs(type(conn_exc), exceptions.ProbableAccessDeniedError)
+            self.assertSequenceEqual(conn_exc.args, [1, 'error text'])
+
+            self.connection.callbacks.process.assert_any_call(
+                0, self.connection.ON_CONNECTION_CLOSED,
+                self.connection, self.connection,
+                1, 'error text')
+
     @mock.patch('pika.connection.Connection.connect')
     def test_new_conn_should_use_first_channel(self, connect):
         """_next_channel_number in new conn should always be 1"""
@@ -124,12 +250,12 @@ class ConnectionTests(unittest.TestCase):
         """make sure the callback adding works"""
         self.connection.callbacks = mock.Mock(spec=self.connection.callbacks)
         for test_method, expected_key in (
-            (self.connection.add_backpressure_callback,
-             self.connection.ON_CONNECTION_BACKPRESSURE),
-            (self.connection.add_on_open_callback,
-             self.connection.ON_CONNECTION_OPEN),
-            (self.connection.add_on_close_callback,
-             self.connection.ON_CONNECTION_CLOSED)):
+                (self.connection.add_backpressure_callback,
+                 self.connection.ON_CONNECTION_BACKPRESSURE),
+                (self.connection.add_on_open_callback,
+                 self.connection.ON_CONNECTION_OPEN),
+                (self.connection.add_on_close_callback,
+                 self.connection.ON_CONNECTION_CLOSED)):
             self.connection.callbacks.reset_mock()
             test_method(callback_method)
             self.connection.callbacks.add.assert_called_once_with(
@@ -234,12 +360,13 @@ class ConnectionTests(unittest.TestCase):
         }
         #Test Type Errors
         for bad_field, bad_value in (
-            ('host', 15672), ('port', '5672'), ('virtual_host', True),
-            ('channel_max', '4'), ('frame_max', '5'), ('credentials', 'bad'),
-            ('locale', 1), ('heartbeat_interval', '6'),
-            ('socket_timeout', '42'), ('retry_delay', 'two'),
-            ('backpressure_detection', 'true'), ('ssl', {'ssl': 'dict'}),
-            ('ssl_options', True), ('connection_attempts', 'hello')):
+                ('host', 15672), ('port', '5672'), ('virtual_host', True),
+                ('channel_max', '4'), ('frame_max', '5'),
+                ('credentials', 'bad'), ('locale', 1),
+                ('heartbeat_interval', '6'), ('socket_timeout', '42'),
+                ('retry_delay', 'two'), ('backpressure_detection', 'true'),
+                ('ssl', {'ssl': 'dict'}), ('ssl_options', True),
+                ('connection_attempts', 'hello')):
             bkwargs = copy.deepcopy(kwargs)
             bkwargs[bad_field] = bad_value
             self.assertRaises(TypeError, connection.ConnectionParameters,
@@ -371,20 +498,28 @@ class ConnectionTests(unittest.TestCase):
         self.assertEqual(['ab'], list(self.connection.outbound_buffer))
         self.assertEqual('hearbeat obj', self.connection.heartbeat)
 
-    def test_on_connection_closed(self):
-        """make sure connection close sends correct frames"""
+    def test_on_connection_close(self):
+        """make sure _on_connection_close terminates connection"""
         method_frame = mock.Mock()
         method_frame.method = mock.Mock(spec=spec.Connection.Close)
         method_frame.method.reply_code = 1
         method_frame.method.reply_text = 'hello'
-        heartbeat = mock.Mock()
-        self.connection.heartbeat = heartbeat
-        self.connection._adapter_disconnect = mock.Mock()
-        self.connection._on_connection_closed(method_frame, from_adapter=False)
+        self.connection._on_terminate = mock.Mock()
+        self.connection._on_connection_close(method_frame)
         #Check
-        self.assertTupleEqual((1, 'hello'), self.connection.closing)
-        heartbeat.stop.assert_called_once_with()
-        self.connection._adapter_disconnect.assert_called_once_with()
+        self.connection._on_terminate.assert_called_once_with(1, 'hello')
+
+    def test_on_connection_close_ok(self):
+        """make sure _on_connection_close_ok terminates connection"""
+        method_frame = mock.Mock()
+        method_frame.method = mock.Mock(spec=spec.Connection.CloseOk)
+        self.connection.closing = (1, 'bye')
+        self.connection._on_terminate = mock.Mock()
+
+        self.connection._on_connection_close_ok(method_frame)
+
+        #Check
+        self.connection._on_terminate.assert_called_once_with(1, 'bye')
 
     @mock.patch('pika.frame.decode_frame')
     def test_on_data_available(self, decode_frame):
diff --git a/tests/unit/heartbeat_tests.py b/tests/unit/heartbeat_tests.py
index 62aa777..4149eef 100644
--- a/tests/unit/heartbeat_tests.py
+++ b/tests/unit/heartbeat_tests.py
@@ -2,6 +2,16 @@
 Tests for pika.heartbeat
 
 """
+
+# Suppress pylint warnings concering access to protected member
+# pylint: disable=W0212
+
+# Suppress pylint messages concering missing docstring
+# pylint: disable=C0111
+
+# Suppress pylint messages concering invalid method name
+# pylint: disable=C0103
+
 try:
     import mock
 except ImportError:
@@ -58,7 +68,7 @@ class HeartbeatTests(unittest.TestCase):
 
     @mock.patch('pika.heartbeat.HeartbeatChecker._setup_timer')
     def test_constructor_called_setup_timer(self, timer):
-        obj = heartbeat.HeartbeatChecker(self.mock_conn, self.INTERVAL)
+        heartbeat.HeartbeatChecker(self.mock_conn, self.INTERVAL)
         timer.assert_called_once_with()
 
     def test_active_true(self):
@@ -135,9 +145,8 @@ class HeartbeatTests(unittest.TestCase):
                                                self.obj._interval)
         self.mock_conn.close.assert_called_once_with(
             self.obj._CONNECTION_FORCED, reason)
-        self.mock_conn._on_disconnect.assert_called_once_with(
+        self.mock_conn._on_terminate.assert_called_once_with(
             self.obj._CONNECTION_FORCED, reason)
-        self.mock_conn._adapter_disconnect.assert_called_once_with()
 
     def test_has_received_data_false(self):
         self.obj._bytes_received = 100