vitaly-krugl: Ready for review
diff --git a/pika/adapters/blocking_connection.py b/pika/adapters/blocking_connection.py
index b1fb7dc..be1db63 100644
--- a/pika/adapters/blocking_connection.py
+++ b/pika/adapters/blocking_connection.py
@@ -542,22 +542,24 @@ class BlockingConnection(object):
         self.add_timeout(0, user_callback)
 
 
-    def _on_connection_blocked(self, user_callback, method_frame):
+    def _on_connection_blocked(self, user_callback, _impl, method_frame):
         """Handle Connection.Blocked notification from RabbitMQ broker
 
         :param callable user_callback: callback passed to
            `add_on_connection_blocked_callback`
+        :param SelectConnection _impl:
         :param pika.frame.Method method_frame: method frame having `method`
             member of type `pika.spec.Connection.Blocked`
         """
         self._ready_events.append(
             _ConnectionBlockedEvt(user_callback, method_frame))
 
-    def _on_connection_unblocked(self, user_callback, method_frame):
+    def _on_connection_unblocked(self, user_callback, _impl, method_frame):
         """Handle Connection.Unblocked notification from RabbitMQ broker
 
         :param callable user_callback: callback passed to
            `add_on_connection_unblocked_callback`
+        :param SelectConnection _impl:
         :param pika.frame.Method method_frame: method frame having `method`
             member of type `pika.spec.Connection.Blocked`
         """
@@ -595,13 +597,14 @@ class BlockingConnection(object):
         See also `ConnectionParameters.blocked_connection_timeout`.
 
         :param method callback: Callback to call on `Connection.Blocked`,
-            having the signature `callback(pika.frame.Method)`, where the
-            method frame's `method` member is of type
-            `pika.spec.Connection.Blocked`
+            having the signature `callback(connection, pika.frame.Method)`,
+            where connection is the `BlockingConnection` instance and the method
+            frame's `method` member is of type `pika.spec.Connection.Blocked`
 
         """
         self._impl.add_on_connection_blocked_callback(
-            functools.partial(self._on_connection_blocked, callback))
+            functools.partial(self._on_connection_blocked,
+                              functools.partial(callback, self)))
 
     def add_on_connection_unblocked_callback(self, callback):
         """Add a callback to be notified when RabbitMQ has sent a
@@ -609,14 +612,15 @@ class BlockingConnection(object):
         to start publishing again. The callback will be passed the
         `Connection.Unblocked` method frame.
 
-        :param method callback: Callback to call on
-            `Connection.Unblocked`, having the signature
-            `callback(pika.frame.Method)`, where the method frame's
-            `method` member is of type `pika.spec.Connection.Unblocked`
+        :param method callback: Callback to call on Connection.Unblocked`,
+            having the signature `callback(connection, pika.frame.Method)`,
+            where connection is the `BlockingConnection` instance and the method
+             frame's `method` member is of type `pika.spec.Connection.Unblocked`
 
         """
         self._impl.add_on_connection_unblocked_callback(
-            functools.partial(self._on_connection_unblocked, callback))
+            functools.partial(self._on_connection_unblocked,
+                              functools.partial(callback, self)))
 
     def add_timeout(self, deadline, callback):
         """Create a single-shot timer to fire after deadline seconds. Do not
diff --git a/pika/connection.py b/pika/connection.py
index c784cc2..25ec506 100644
--- a/pika/connection.py
+++ b/pika/connection.py
@@ -6,6 +6,7 @@ import ast
 import sys
 import collections
 import copy
+import functools
 import logging
 import math
 import numbers
@@ -1018,11 +1019,9 @@ class Connection(object):
     # pylint: disable=R0201
 
     ON_CONNECTION_BACKPRESSURE = '_on_connection_backpressure'
-    ON_CONNECTION_BLOCKED = '_on_connection_blocked'
     ON_CONNECTION_CLOSED = '_on_connection_closed'
     ON_CONNECTION_ERROR = '_on_connection_error'
     ON_CONNECTION_OPEN = '_on_connection_open'
-    ON_CONNECTION_UNBLOCKED = '_on_connection_unblocked'
     CONNECTION_CLOSED = 0
     CONNECTION_INIT = 1
     CONNECTION_PROTOCOL = 2
@@ -1146,19 +1145,21 @@ class Connection(object):
         instead of relying on back pressure throttling. The callback
         will be passed the ``Connection.Blocked`` method frame.
 
-        TODO Also pass the connection as the callback's first arg
-
         See also `ConnectionParameters.blocked_connection_timeout`.
 
         :param method callback: Callback to call on `Connection.Blocked`,
-            having the signature `callback(pika.frame.Method)`, where the
-            method frame's `method` member is of type
+            having the signature `callback(connection, pika.frame.Method)`,
+            where the method frame's `method` member is of type
             `pika.spec.Connection.Blocked`
 
         """
         if not callable(callback):
             raise TypeError('callback should be a function or method.')
-        self.callbacks.add(0, spec.Connection.Blocked, callback, False)
+
+        self.callbacks.add(0,
+                           spec.Connection.Blocked,
+                           functools.partial(callback, self),
+                           one_shot=False)
 
     def add_on_connection_unblocked_callback(self, callback):
         """Add a callback to be notified when RabbitMQ has sent a
@@ -1166,17 +1167,19 @@ class Connection(object):
         to start publishing again. The callback will be passed the
         ``Connection.Unblocked`` method frame.
 
-        TODO Also pass the connection as the callback's first arg
-
         :param method callback: Callback to call on
             `Connection.Unblocked`, having the signature
-            `callback(pika.frame.Method)`, where the method frame's
+            `callback(connection, pika.frame.Method)`, where the method frame's
             `method` member is of type `pika.spec.Connection.Unblocked`
 
         """
         if not callable(callback):
             raise TypeError('callback should be a function or method.')
-        self.callbacks.add(0, spec.Connection.Unblocked, callback, False)
+
+        self.callbacks.add(0,
+                           spec.Connection.Unblocked,
+                           functools.partial(callback, self),
+                           one_shot=False)
 
     def add_on_open_callback(self, callback):
         """Add a callback notification when the connection has opened. The
@@ -1789,7 +1792,7 @@ class Connection(object):
         self._on_terminate(InternalCloseReasons.BLOCKED_CONNECTION_TIMEOUT,
                            'Blocked connection timeout expired')
 
-    def _on_connection_blocked(self, method_frame):
+    def _on_connection_blocked(self, _connection, method_frame):
         """Handle Connection.Blocked notification from RabbitMQ broker
 
         :param pika.frame.Method method_frame: method frame having `method`
@@ -1808,7 +1811,7 @@ class Connection(object):
                 self.params.blocked_connection_timeout,
                 self._on_blocked_connection_timeout)
 
-    def _on_connection_unblocked(self, method_frame):
+    def _on_connection_unblocked(self, _connection, method_frame):
         """Handle Connection.Unblocked notification from RabbitMQ broker
 
         :param pika.frame.Method method_frame: method frame having `method`

Pass connection instance as first arg of callbacks for add_on_connection_blocked_callback and add_on_connection_unblocked_callback
Targeting 1.0.0
diff --git a/tests/acceptance/async_adapter_tests.py b/tests/acceptance/async_adapter_tests.py
index d27a625..0539596 100644
--- a/tests/acceptance/async_adapter_tests.py
+++ b/tests/acceptance/async_adapter_tests.py
@@ -445,9 +445,11 @@ class TestBlockedConnectionTimesOut(AsyncTestCase, AsyncAdapters):  # pylint: di
     def begin(self, channel):
 
         # Simulate Connection.Blocked
-        channel.connection._on_connection_blocked(pika.frame.Method(
-            0,
-            pika.spec.Connection.Blocked('Testing blocked connection timeout')))
+        channel.connection._on_connection_blocked(
+            channel.connection,
+            pika.frame.Method(0,
+                              pika.spec.Connection.Blocked(
+                                  'Testing blocked connection timeout')))
 
     def on_closed(self, connection, reply_code, reply_text):
         """called when the connection has finished closing"""
@@ -471,15 +473,16 @@ class TestBlockedConnectionUnblocks(AsyncTestCase, AsyncAdapters):  # pylint: di
     def begin(self, channel):
 
         # Simulate Connection.Blocked
-        channel.connection._on_connection_blocked(pika.frame.Method(
-            0,
-            pika.spec.Connection.Blocked(
-                'Testing blocked connection unblocks')))
+        channel.connection._on_connection_blocked(
+            channel.connection,
+            pika.frame.Method(0,
+                              pika.spec.Connection.Blocked(
+                                  'Testing blocked connection unblocks')))
 
         # Simulate Connection.Unblocked
-        channel.connection._on_connection_unblocked(pika.frame.Method(
-            0,
-            pika.spec.Connection.Unblocked()))
+        channel.connection._on_connection_unblocked(
+            channel.connection,
+            pika.frame.Method(0, pika.spec.Connection.Unblocked()))
 
         # Schedule shutdown after blocked connection timeout would expire
         channel.connection.add_timeout(0.005, self.on_cleanup_timer)
diff --git a/tests/acceptance/blocking_adapter_test.py b/tests/acceptance/blocking_adapter_test.py
index 5623922..03b4681 100644
--- a/tests/acceptance/blocking_adapter_test.py
+++ b/tests/acceptance/blocking_adapter_test.py
@@ -401,27 +401,34 @@ class TestConnectionRegisterForBlockAndUnblock(BlockingTestCaseBase):
 
         # NOTE: I haven't figured out yet how to coerce RabbitMQ to emit
         # Connection.Block and Connection.Unblock from the test, so we'll
-        # just call the registration functions for now, to make sure that
-        # registration doesn't crash
-
-        connection.add_on_connection_blocked_callback(lambda frame: None)
+        # just call the registration functions for now and simulate incoming
+        # blocked/unblocked frames
 
         blocked_buffer = []
-        evt = blocking_connection._ConnectionBlockedEvt(
-            lambda f: blocked_buffer.append("blocked"),
-            pika.frame.Method(1, pika.spec.Connection.Blocked('reason')))
-        repr(evt)
-        evt.dispatch()
-        self.assertEqual(blocked_buffer, ["blocked"])
+        connection.add_on_connection_blocked_callback(
+            lambda conn, frame: blocked_buffer.append((conn, frame)))
+        # Simulate dispatch of blocked connection
+        blocked_frame = pika.frame.Method(
+            0,
+            pika.spec.Connection.Blocked('reason'))
+        connection._impl._process_frame(blocked_frame)
+        connection.sleep(0) # facilitate dispatch of pending events
+        self.assertEqual(len(blocked_buffer), 1)
+        conn, frame = blocked_buffer[0]
+        self.assertIs(conn, connection)
+        self.assertIs(frame, blocked_frame)
 
         unblocked_buffer = []
-        connection.add_on_connection_unblocked_callback(lambda frame: None)
-        evt = blocking_connection._ConnectionUnblockedEvt(
-            lambda f: unblocked_buffer.append("unblocked"),
-            pika.frame.Method(1, pika.spec.Connection.Unblocked()))
-        repr(evt)
-        evt.dispatch()
-        self.assertEqual(unblocked_buffer, ["unblocked"])
+        connection.add_on_connection_unblocked_callback(
+            lambda conn, frame: unblocked_buffer.append((conn, frame)))
+        # Simulate dispatch of unblocked connection
+        unblocked_frame = pika.frame.Method(0, pika.spec.Connection.Unblocked())
+        connection._impl._process_frame(unblocked_frame)
+        connection.sleep(0) # facilitate dispatch of pending events
+        self.assertEqual(len(unblocked_buffer), 1)
+        conn, frame = unblocked_buffer[0]
+        self.assertIs(conn, connection)
+        self.assertIs(frame, unblocked_frame)
 
 
 class TestBlockedConnectionTimeout(BlockingTestCaseBase):
@@ -436,9 +443,11 @@ class TestBlockedConnectionTimeout(BlockingTestCaseBase):
         # simulate it for now
 
         # Simulate Connection.Blocked
-        conn._impl._on_connection_blocked(pika.frame.Method(
-            0,
-            pika.spec.Connection.Blocked('TestBlockedConnectionTimeout')))
+        conn._impl._on_connection_blocked(
+            conn._impl,
+            pika.frame.Method(
+                0,
+                pika.spec.Connection.Blocked('TestBlockedConnectionTimeout')))
 
         # Wait for connection teardown
         with self.assertRaises(pika.exceptions.ConnectionClosed) as excCtx:
diff --git a/tests/unit/blocking_connection_tests.py b/tests/unit/blocking_connection_tests.py
index f303023..5fda321 100644
--- a/tests/unit/blocking_connection_tests.py
+++ b/tests/unit/blocking_connection_tests.py
@@ -15,6 +15,19 @@ import pika.channel
 from pika.exceptions import AMQPConnectionError, ChannelClosed
 
 
+# Disable protected-access
+# pylint: disable=W0212
+
+# Disable missing-docstring
+# pylint: disable=C0111
+
+# Disable invalid-name
+# pylint: disable=C0103
+
+# Disable no-self-use
+# pylint: disable=R0201
+
+
 class BlockingConnectionMockTemplate(blocking_connection.BlockingConnection):
     pass
 
@@ -325,3 +338,25 @@ class BlockingConnectionTests(unittest.TestCase):
             # and the raised error must then looks like:
             self.assertEqual('Connection to 127.0.0.1:5672 failed: timeout',
                              str(ctx.exception))
+
+    def test_connection_blocked_evt(self):
+        blocked_buffer = []
+        frame = pika.frame.Method(0, pika.spec.Connection.Blocked('reason'))
+        evt = blocking_connection._ConnectionBlockedEvt(
+            blocked_buffer.append,
+            frame)
+        repr(evt)
+        evt.dispatch()
+        self.assertEqual(len(blocked_buffer), 1)
+        self.assertIs(blocked_buffer[0], frame)
+
+    def test_connection_unblocked_evt(self):
+        unblocked_buffer = []
+        frame = pika.frame.Method(0, pika.spec.Connection.Unblocked())
+        evt = blocking_connection._ConnectionUnblockedEvt(
+            unblocked_buffer.append,
+            frame)
+        repr(evt)
+        evt.dispatch()
+        self.assertEqual(len(unblocked_buffer), 1)
+        self.assertIs(unblocked_buffer[0], frame)
diff --git a/tests/unit/connection_tests.py b/tests/unit/connection_tests.py
index 38db1d7..9d99314 100644
--- a/tests/unit/connection_tests.py
+++ b/tests/unit/connection_tests.py
@@ -678,6 +678,36 @@ class ConnectionTests(unittest.TestCase):  # pylint: disable=R0904
             if frame_type == frame.Heartbeat:
                 self.assertTrue(self.connection.heartbeat.received.called)
 
+    def test_add_on_connection_blocked_callback(self):
+        blocked_buffer = []
+        self.connection.add_on_connection_blocked_callback(
+            lambda conn, frame: blocked_buffer.append((conn, frame)))
+
+        # Simulate dispatch of blocked connection
+        blocked_frame = pika.frame.Method(
+            0,
+            pika.spec.Connection.Blocked('reason'))
+        self.connection._process_frame(blocked_frame)
+
+        self.assertEqual(len(blocked_buffer), 1)
+        conn, frame = blocked_buffer[0]
+        self.assertIs(conn, self.connection)
+        self.assertIs(frame, blocked_frame)
+
+    def test_add_on_connection_unblocked_callback(self):
+        unblocked_buffer = []
+        self.connection.add_on_connection_unblocked_callback(
+            lambda conn, frame: unblocked_buffer.append((conn, frame)))
+
+        # Simulate dispatch of unblocked connection
+        unblocked_frame = pika.frame.Method(0, pika.spec.Connection.Unblocked())
+        self.connection._process_frame(unblocked_frame)
+
+        self.assertEqual(len(unblocked_buffer), 1)
+        conn, frame = unblocked_buffer[0]
+        self.assertIs(conn, self.connection)
+        self.assertIs(frame, unblocked_frame)
+
     @mock.patch.object(
         connection.Connection,
         'connect',
@@ -714,6 +744,7 @@ class ConnectionTests(unittest.TestCase):  # pylint: disable=R0904
                 blocked_connection_timeout=60))
 
         conn._on_connection_blocked(
+            conn,
             mock.Mock(name='frame.Method(Connection.Blocked)'))
 
         # Check
@@ -736,6 +767,7 @@ class ConnectionTests(unittest.TestCase):  # pylint: disable=R0904
 
         # Simulate Connection.Blocked trigger
         conn._on_connection_blocked(
+            conn,
             mock.Mock(name='frame.Method(Connection.Blocked)'))
 
         # Check
@@ -748,6 +780,7 @@ class ConnectionTests(unittest.TestCase):  # pylint: disable=R0904
 
         # Simulate Connection.Blocked trigger again
         conn._on_connection_blocked(
+            conn,
             mock.Mock(name='frame.Method(Connection.Blocked)'))
 
         self.assertEqual(conn.add_timeout.call_count, 1)
@@ -770,6 +803,7 @@ class ConnectionTests(unittest.TestCase):  # pylint: disable=R0904
                 blocked_connection_timeout=60))
 
         conn._on_connection_blocked(
+            conn,
             mock.Mock(name='frame.Method(Connection.Blocked)'))
 
         conn._on_blocked_connection_timeout()
@@ -798,6 +832,7 @@ class ConnectionTests(unittest.TestCase):  # pylint: disable=R0904
                 blocked_connection_timeout=60))
 
         conn._on_connection_blocked(
+            conn,
             mock.Mock(name='frame.Method(Connection.Blocked)'))
 
         self.assertIsNotNone(conn._blocked_conn_timer)
@@ -805,6 +840,7 @@ class ConnectionTests(unittest.TestCase):  # pylint: disable=R0904
         timer = conn._blocked_conn_timer
 
         conn._on_connection_unblocked(
+            conn,
             mock.Mock(name='frame.Method(Connection.Unblocked)'))
 
         # Check
@@ -829,6 +865,7 @@ class ConnectionTests(unittest.TestCase):  # pylint: disable=R0904
 
         # Simulate Connection.Blocked
         conn._on_connection_blocked(
+            conn,
             mock.Mock(name='frame.Method(Connection.Blocked)'))
 
         self.assertIsNotNone(conn._blocked_conn_timer)
@@ -837,6 +874,7 @@ class ConnectionTests(unittest.TestCase):  # pylint: disable=R0904
 
         # Simulate Connection.Unblocked
         conn._on_connection_unblocked(
+            conn,
             mock.Mock(name='frame.Method(Connection.Unblocked)'))
 
         # Check
@@ -845,6 +883,7 @@ class ConnectionTests(unittest.TestCase):  # pylint: disable=R0904
 
         # Simulate Connection.Unblocked again
         conn._on_connection_unblocked(
+            conn,
             mock.Mock(name='frame.Method(Connection.Unblocked)'))
 
         self.assertEqual(conn.remove_timeout.call_count, 1)
@@ -872,6 +911,7 @@ class ConnectionTests(unittest.TestCase):  # pylint: disable=R0904
                 blocked_connection_timeout=60))
 
         conn._on_connection_blocked(
+            conn,
             mock.Mock(name='frame.Method(Connection.Blocked)'))
 
         self.assertIsNotNone(conn._blocked_conn_timer)

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

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

tests/unit/connection_tests.py::ConnectionTests::test_add_on_connection_blocked_callback tests/unit/connection_tests.py::ConnectionTests::test_add_on_connection_unblocked_callback tests/unit/connection_tests.py::ConnectionTests::test_blocked_connection_multiple_blocked_in_a_row_sets_timer_once tests/unit/connection_tests.py::ConnectionTests::test_blocked_connection_multiple_unblocked_in_a_row_removes_timer_once tests/unit/connection_tests.py::ConnectionTests::test_blocked_connection_on_terminate_removes_timer tests/unit/connection_tests.py::ConnectionTests::test_blocked_connection_timeout_teminates_connection tests/unit/connection_tests.py::ConnectionTests::test_blocked_connection_unblocked_removes_timer tests/unit/connection_tests.py::ConnectionTests::test_connection_blocked_sets_timer
tests/acceptance/blocking_adapter_test.py::TestCreateAndCloseConnection::test tests/acceptance/blocking_adapter_test.py::TestMultiCloseConnection::test tests/acceptance/blocking_adapter_test.py::TestConnectionContextManagerClosesConnection::test tests/acceptance/blocking_adapter_test.py::TestConnectionContextManagerClosesConnectionAndPassesOriginalException::test tests/acceptance/blocking_adapter_test.py::TestConnectionContextManagerClosesConnectionAndPassesSystemException::test tests/acceptance/blocking_adapter_test.py::TestLostConnectionResultsInIsClosedConnectionAndChannel::test tests/acceptance/blocking_adapter_test.py::TestInvalidExchangeTypeRaisesConnectionClosed::test tests/acceptance/blocking_adapter_test.py::TestCreateAndCloseConnectionWithChannelAndConsumer::test tests/acceptance/blocking_adapter_test.py::TestSuddenBrokerDisconnectBeforeChannel::test tests/acceptance/blocking_adapter_test.py::TestNoAccessToFileDescriptorAfterConnectionClosed::test tests/acceptance/blocking_adapter_test.py::TestDisconnectDuringConnectionStart::test tests/acceptance/blocking_adapter_test.py::TestDisconnectDuringConnectionTune::test tests/acceptance/blocking_adapter_test.py::TestProcessDataEvents::test tests/acceptance/blocking_adapter_test.py::TestConnectionRegisterForBlockAndUnblock::test tests/acceptance/blocking_adapter_test.py::TestBlockedConnectionTimeout::test tests/acceptance/blocking_adapter_test.py::TestAddCallbackThreadsafeFromSameThread::test tests/acceptance/blocking_adapter_test.py::TestAddCallbackThreadsafeFromAnotherThread::test tests/acceptance/blocking_adapter_test.py::TestAddTimeoutRemoveTimeout::test tests/acceptance/blocking_adapter_test.py::TestViabilityOfMultipleTimeoutsWithSameDeadlineAndCallback::test tests/acceptance/blocking_adapter_test.py::TestRemoveTimeoutFromTimeoutCallback::test tests/acceptance/blocking_adapter_test.py::TestSleep::test tests/acceptance/blocking_adapter_test.py::TestConnectionProperties::test tests/acceptance/blocking_adapter_test.py::TestCreateAndCloseChannel::test tests/acceptance/blocking_adapter_test.py::TestExchangeDeclareAndDelete::test tests/acceptance/blocking_adapter_test.py::TestExchangeBindAndUnbind::test tests/acceptance/blocking_adapter_test.py::TestQueueDeclareAndDelete::test tests/acceptance/blocking_adapter_test.py::TestPassiveQueueDeclareOfUnknownQueueRaisesChannelClosed::test tests/acceptance/blocking_adapter_test.py::TestQueueBindAndUnbindAndPurge::test tests/acceptance/blocking_adapter_test.py::TestBasicGet::test tests/acceptance/blocking_adapter_test.py::TestBasicReject::test tests/acceptance/blocking_adapter_test.py::TestBasicRejectNoRequeue::test tests/acceptance/blocking_adapter_test.py::TestBasicNack::test tests/acceptance/blocking_adapter_test.py::TestBasicNackNoRequeue::test tests/acceptance/blocking_adapter_test.py::TestBasicNackMultiple::test tests/acceptance/blocking_adapter_test.py::TestBasicRecoverWithRequeue::test tests/acceptance/blocking_adapter_test.py::TestTxCommit::test tests/acceptance/blocking_adapter_test.py::TestTxRollback::test tests/acceptance/blocking_adapter_test.py::TestBasicConsumeFromUnknownQueueRaisesChannelClosed::test tests/acceptance/blocking_adapter_test.py::TestPublishAndBasicPublishWithPubacksUnroutable::test tests/acceptance/blocking_adapter_test.py::TestConfirmDeliveryAfterUnroutableMessage::test tests/acceptance/blocking_adapter_test.py::TestUnroutableMessagesReturnedInNonPubackMode::test tests/acceptance/blocking_adapter_test.py::TestUnroutableMessageReturnedInPubackMode::test tests/acceptance/blocking_adapter_test.py::TestBasicPublishDeliveredWhenPendingUnroutable::test tests/acceptance/blocking_adapter_test.py::TestPublishAndConsumeWithPubacksAndQosOfOne::test tests/acceptance/blocking_adapter_test.py::TestBasicConsumeWithAckFromAnotherThread::test tests/acceptance/blocking_adapter_test.py::TestConsumeGeneratorWithAckFromAnotherThread::test tests/acceptance/blocking_adapter_test.py::TestTwoBasicConsumersOnSameChannel::test tests/acceptance/blocking_adapter_test.py::TestBasicCancelPurgesPendingConsumerCancellationEvt::test tests/acceptance/blocking_adapter_test.py::TestBasicPublishWithoutPubacks::test tests/acceptance/blocking_adapter_test.py::TestPublishFromBasicConsumeCallback::test tests/acceptance/blocking_adapter_test.py::TestStopConsumingFromBasicConsumeCallback::test tests/acceptance/blocking_adapter_test.py::TestCloseChannelFromBasicConsumeCallback::test tests/acceptance/blocking_adapter_test.py::TestCloseConnectionFromBasicConsumeCallback::test tests/acceptance/blocking_adapter_test.py::TestNonPubAckPublishAndConsumeHugeMessage::test tests/acceptance/blocking_adapter_test.py::TestNonPubackPublishAndConsumeManyMessages::test tests/acceptance/blocking_adapter_test.py::TestBasicCancelWithNonAckableConsumer::test tests/acceptance/blocking_adapter_test.py::TestBasicCancelWithAckableConsumer::test tests/acceptance/blocking_adapter_test.py::TestUnackedMessageAutoRestoredToQueueOnChannelClose::test tests/acceptance/blocking_adapter_test.py::TestNoAckMessageNotRestoredToQueueOnChannelClose::test tests/acceptance/blocking_adapter_test.py::TestConsumeInactivityTimeout::test tests/acceptance/blocking_adapter_test.py::TestConsumeGeneratorInterruptedByCancelFromBroker::test tests/acceptance/blocking_adapter_test.py::TestConsumeGeneratorCancelEncountersCancelFromBroker::test tests/acceptance/blocking_adapter_test.py::TestChannelFlow::test
tests/acceptance/blocking_adapter_test.py::TestConnectWithDownedBroker::test tests/acceptance/blocking_adapter_test.py::TestDisconnectDuringConnectionProtocol::test tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_channel tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_close tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_close_with_channel_closed_exception tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_connection_attempts_with_timeout tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_connection_blocked_evt tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_connection_unblocked_evt tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_constructor tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_flush_output tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_flush_output_server_initiated_error_close tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_flush_output_server_initiated_no_error_close tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_flush_output_user_initiated_close tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_process_io_for_connection_setup tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_process_io_for_connection_setup_fails_with_open_error tests/unit/blocking_connection_tests.py::BlockingConnectionTests::test_sleep tests/unit/connection_tests.py::ConnectionTests::test_add_callbacks tests/unit/connection_tests.py::ConnectionTests::test_add_on_close_callback tests/unit/connection_tests.py::ConnectionTests::test_add_on_open_error_callback tests/unit/connection_tests.py::ConnectionTests::test_channel tests/unit/connection_tests.py::ConnectionTests::test_channel_on_closed_connection_raises_connection_closed tests/unit/connection_tests.py::ConnectionTests::test_channel_on_closing_connection_raises_connection_closed tests/unit/connection_tests.py::ConnectionTests::test_channel_on_init_connection_raises_connection_closed tests/unit/connection_tests.py::ConnectionTests::test_channel_on_protocol_connection_raises_connection_closed tests/unit/connection_tests.py::ConnectionTests::test_channel_on_start_connection_raises_connection_closed tests/unit/connection_tests.py::ConnectionTests::test_channel_on_tune_connection_raises_connection_closed tests/unit/connection_tests.py::ConnectionTests::test_client_properties tests/unit/connection_tests.py::ConnectionTests::test_client_properties_default tests/unit/connection_tests.py::ConnectionTests::test_client_properties_override tests/unit/connection_tests.py::ConnectionTests::test_close_bails_out_if_already_closed_or_closing tests/unit/connection_tests.py::ConnectionTests::test_close_calls_on_close_ready_when_no_channels tests/unit/connection_tests.py::ConnectionTests::test_close_channels tests/unit/connection_tests.py::ConnectionTests::test_close_closes_open_channels tests/unit/connection_tests.py::ConnectionTests::test_close_closes_opening_channels tests/unit/connection_tests.py::ConnectionTests::test_close_does_not_close_closing_channels tests/unit/connection_tests.py::ConnectionTests::test_connect_no_adapter_connect_from_constructor tests/unit/connection_tests.py::ConnectionTests::test_create_with_blocked_connection_timeout_config tests/unit/connection_tests.py::ConnectionTests::test_deliver_frame_to_channel_with_frame_for_unknown_channel tests/unit/connection_tests.py::ConnectionTests::test_new_conn_should_use_first_channel tests/unit/connection_tests.py::ConnectionTests::test_next_channel_number_returns_lowest_unused tests/unit/connection_tests.py::ConnectionTests::test_on_channel_cleanup_closing_state_last_channel_calls_on_close_ready tests/unit/connection_tests.py::ConnectionTests::test_on_channel_cleanup_closing_state_more_channels_no_on_close_ready tests/unit/connection_tests.py::ConnectionTests::test_on_channel_cleanup_non_closing_state tests/unit/connection_tests.py::ConnectionTests::test_on_channel_cleanup_with_closing_channels tests/unit/connection_tests.py::ConnectionTests::test_on_connect_timer tests/unit/connection_tests.py::ConnectionTests::test_on_connect_timer_reconnect tests/unit/connection_tests.py::ConnectionTests::test_on_connection_close_ok tests/unit/connection_tests.py::ConnectionTests::test_on_connection_closed tests/unit/connection_tests.py::ConnectionTests::test_on_connection_start tests/unit/connection_tests.py::ConnectionTests::test_on_connection_tune tests/unit/connection_tests.py::ConnectionTests::test_on_data_available tests/unit/connection_tests.py::ConnectionTests::test_on_terminate_cleans_up tests/unit/connection_tests.py::ConnectionTests::test_on_terminate_invokes_access_denied_on_connection_error_and_closed tests/unit/connection_tests.py::ConnectionTests::test_on_terminate_invokes_auth_on_connection_error_and_closed tests/unit/connection_tests.py::ConnectionTests::test_on_terminate_invokes_connection_closed_callback tests/unit/connection_tests.py::ConnectionTests::test_on_terminate_invokes_protocol_on_connection_error_and_closed tests/unit/connection_tests.py::ConnectionTests::test_send_message_updates_frames_sent_and_bytes_sent tests/unit/connection_tests.py::ConnectionTests::test_set_backpressure_multiplier
pika/adapters/blocking_connection.py pika/connection.py
pika/adapters/blocking_connection.py pika/connection.py