"""blocking adapter test"""
from datetime import datetime
import functools
import logging
import socket
import threading
import unittest
import uuid

import pika
from pika.adapters import blocking_connection
from pika.compat import as_bytes, time_now
import pika.connection
import pika.exceptions
from pika.exchange_type import ExchangeType

from tests.misc.forward_server import ForwardServer
from tests.misc.test_utils import retry_assertion

# too-many-lines
# pylint: disable=C0302

# Disable warning about access to protected member
# pylint: disable=W0212

# Disable warning Attribute defined outside __init__
# pylint: disable=W0201

# Disable warning Missing docstring
# pylint: disable=C0111

# Disable warning Too many public methods
# pylint: disable=R0904

# Disable warning Invalid variable name
# pylint: disable=C0103


LOGGER = logging.getLogger(__name__)

PARAMS_URL_TEMPLATE = (
    'amqp://guest:guest@127.0.0.1:%(port)s/%%2f?socket_timeout=1')
DEFAULT_URL = PARAMS_URL_TEMPLATE % {'port': 5672}
DEFAULT_PARAMS = pika.URLParameters(DEFAULT_URL)
DEFAULT_TIMEOUT = 15



def setUpModule():
    logging.basicConfig(level=logging.DEBUG)


class BlockingTestCaseBase(unittest.TestCase):

    TIMEOUT = DEFAULT_TIMEOUT

    def _connect(self,
                 url=DEFAULT_URL,
                 connection_class=pika.BlockingConnection,
                 impl_class=None):
        parameters = pika.URLParameters(url)

        return self._connect_params(parameters,
                                    connection_class,
                                    impl_class)

    def _connect_params(self,
                        parameters,
                        connection_class=pika.BlockingConnection,
                        impl_class=None):

        connection = connection_class(parameters, _impl_class=impl_class)
        self.addCleanup(lambda: connection.close()
                        if connection.is_open else None)

        # We use impl's timer directly in order to get a callback regardless
        # of BlockingConnection's event dispatch modality
        connection._impl._adapter_call_later(self.TIMEOUT, # pylint: disable=E1101
                                             self._on_test_timeout)

        # Patch calls into I/O loop to fail test if exceptions are
        # leaked back through SelectConnection or the I/O loop.
        self._instrument_io_loop_exception_leak_detection(connection)

        return connection

    def _instrument_io_loop_exception_leak_detection(self, connection):
        """Instrument the given connection to detect and fail test when
        an exception is leaked through the I/O loop

        NOTE: BlockingConnection's underlying asynchronous connection adapter
        (SelectConnection) uses callbacks to communicate with its user (
        BlockingConnection in this case). If BlockingConnection leaks
        exceptions back into the I/O loop or the asynchronous connection
        adapter, we interrupt their normal workflow and introduce a high
        likelihood of state inconsistency.
        """
        # Patch calls into I/O loop to fail test if exceptions are
        # leaked back through SelectConnection or the I/O loop.
        real_poll = connection._impl.ioloop.poll
        def my_poll(*args, **kwargs):
            try:
                return real_poll(*args, **kwargs)
            except BaseException as exc:
                self.fail('Unwanted exception leaked into asynchronous layer '
                          'via ioloop.poll(): {!r}'.format(exc))

        connection._impl.ioloop.poll = my_poll
        self.addCleanup(setattr, connection._impl.ioloop, 'poll', real_poll)

        real_process_timeouts = connection._impl.ioloop.process_timeouts
        def my_process_timeouts(*args, **kwargs):
            try:
                return real_process_timeouts(*args, **kwargs)
            except AssertionError:
                # Our test timeout logic and unit test assert* routines rely
                # on being able to pass AssertionError
                raise
            except BaseException as exc:
                self.fail('Unwanted exception leaked into asynchronous layer '
                          'via ioloop.process_timeouts(): {!r}'.format(exc))

        connection._impl.ioloop.process_timeouts = my_process_timeouts
        self.addCleanup(setattr, connection._impl.ioloop, 'process_timeouts',
                        real_process_timeouts)

    def _on_test_timeout(self):
        """Called when test times out"""
        LOGGER.info('%s TIMED OUT (%s)', datetime.utcnow(), self)
        self.fail('Test timed out')

    @retry_assertion(TIMEOUT/2)
    def _assert_exact_message_count_with_retries(self,
                                                 channel,
                                                 queue,
                                                 expected_count):
        frame = channel.queue_declare(queue, passive=True)
        self.assertEqual(frame.method.message_count, expected_count)


class TestCreateAndCloseConnection(BlockingTestCaseBase):

    def test(self):
        """BlockingConnection: Create and close connection"""
        connection = self._connect()
        self.assertIsInstance(connection, pika.BlockingConnection)
        self.assertTrue(connection.is_open)
        self.assertFalse(connection.is_closed)
        self.assertFalse(connection._impl.is_closing)

        connection.close()
        self.assertTrue(connection.is_closed)
        self.assertFalse(connection.is_open)
        self.assertFalse(connection._impl.is_closing)


class TestCreateConnectionWithNoneSocketAndStackTimeouts(BlockingTestCaseBase):

    def test(self):
        """ BlockingConnection: create a connection with socket and stack timeouts both None

        """
        params = pika.URLParameters(DEFAULT_URL)
        params.socket_timeout = None
        params.stack_timeout = None

        with self._connect_params(params) as connection:
            self.assertTrue(connection.is_open)


class TestCreateConnectionFromTwoConfigsFirstUnreachable(BlockingTestCaseBase):

    def test(self):
        """ BlockingConnection: create a connection from two configs, first unreachable

        """
        # Reserve a port for use in connect
        sock = socket.socket()
        self.addCleanup(sock.close)

        sock.bind(('127.0.0.1', 0))

        port = sock.getsockname()[1]

        sock.close()

        bad_params = pika.URLParameters(PARAMS_URL_TEMPLATE % {"port": port})
        good_params = pika.URLParameters(DEFAULT_URL)

        with self._connect_params([bad_params, good_params]) as connection:
            self.assertNotEqual(connection._impl.params.port, bad_params.port)
            self.assertEqual(connection._impl.params.port, good_params.port)


class TestCreateConnectionFromTwoUnreachableConfigs(BlockingTestCaseBase):

    def test(self):
        """ BlockingConnection: creating a connection from two unreachable \
        configs raises AMQPConnectionError

        """
        # Reserve a port for use in connect
        sock = socket.socket()
        self.addCleanup(sock.close)

        sock.bind(('127.0.0.1', 0))

        port = sock.getsockname()[1]

        sock.close()

        bad_params = pika.URLParameters(PARAMS_URL_TEMPLATE % {"port": port})

        with self.assertRaises(pika.exceptions.AMQPConnectionError):
            self._connect_params([bad_params, bad_params])


class TestMultiCloseConnectionRaisesWrongState(BlockingTestCaseBase):

    def test(self):
        """BlockingConnection: Close connection twice raises ConnectionWrongStateError"""
        connection = self._connect()
        self.assertIsInstance(connection, pika.BlockingConnection)
        self.assertTrue(connection.is_open)
        self.assertFalse(connection.is_closed)
        self.assertFalse(connection._impl.is_closing)

        connection.close()
        self.assertTrue(connection.is_closed)
        self.assertFalse(connection.is_open)
        self.assertFalse(connection._impl.is_closing)

        with self.assertRaises(pika.exceptions.ConnectionWrongStateError):
            connection.close()


class TestConnectionContextManagerClosesConnection(BlockingTestCaseBase):
    def test(self):
        """BlockingConnection: connection context manager closes connection"""
        with self._connect() as connection:
            self.assertIsInstance(connection, pika.BlockingConnection)
            self.assertTrue(connection.is_open)

        self.assertTrue(connection.is_closed)


class TestConnectionContextManagerExitSurvivesClosedConnection(BlockingTestCaseBase):
    def test(self):
        """BlockingConnection: connection context manager exit survives closed connection"""
        with self._connect() as connection:
            self.assertTrue(connection.is_open)
            connection.close()
            self.assertTrue(connection.is_closed)

        self.assertTrue(connection.is_closed)


class TestConnectionContextManagerClosesConnectionAndPassesOriginalException(BlockingTestCaseBase):
    def test(self):
        """BlockingConnection: connection context manager closes connection and passes original exception"""  # pylint: disable=C0301
        class MyException(Exception):
            pass

        with self.assertRaises(MyException):
            with self._connect() as connection:
                self.assertTrue(connection.is_open)

                raise MyException()

        self.assertTrue(connection.is_closed)


class TestConnectionContextManagerClosesConnectionAndPassesSystemException(BlockingTestCaseBase):
    def test(self):
        """BlockingConnection: connection context manager closes connection and passes system exception"""  # pylint: disable=C0301
        with self.assertRaises(SystemExit):
            with self._connect() as connection:
                self.assertTrue(connection.is_open)
                raise SystemExit()

        self.assertTrue(connection.is_closed)


class TestLostConnectionResultsInIsClosedConnectionAndChannel(BlockingTestCaseBase):
    def test(self):
        connection = self._connect()
        channel = connection.channel()

        # Simulate the server dropping the socket connection
        connection._impl._transport._sock.shutdown(socket.SHUT_RDWR)

        with self.assertRaises(pika.exceptions.StreamLostError):
            # Changing QoS should result in ConnectionClosed
            channel.basic_qos()

        # Now check is_open/is_closed on channel and connection
        self.assertFalse(channel.is_open)
        self.assertTrue(channel.is_closed)
        self.assertFalse(connection.is_open)
        self.assertTrue(connection.is_closed)


class TestUpdateSecret(BlockingTestCaseBase):
    def test(self):
        connection = self._connect()
        channel = connection.channel()

        # Update secret
        connection.update_secret("new_secret", "reason")

        # Now check is_open/is_closed on channel and connection
        self.assertTrue(channel.is_open)
        self.assertFalse(channel.is_closed)
        self.assertTrue(connection.is_open)
        self.assertFalse(connection.is_closed)


class TestUpdateSecretOnClosedRaisesWrongState(BlockingTestCaseBase):
    def test(self):
        connection = self._connect()

        # Close Connection
        connection.close()

        # Attempt to update secret
        with self.assertRaises(pika.exceptions.ConnectionWrongStateError):
            connection.update_secret("new_secret", "reason")

        # Now check is_open/is_closed on channel and connection
        self.assertFalse(connection.is_open)
        self.assertTrue(connection.is_closed)


class TestUpdateSecretExpectsStrings(BlockingTestCaseBase):
    def test(self):
        connection = self._connect()

        # Attempt to update secret with integer as new_secret
        with self.assertRaises(AssertionError):
            connection.update_secret(1, "reason")

        # Attempt to update secret with integer as reason
        with self.assertRaises(AssertionError):
            connection.update_secret("new_secret", 1)


class TestInvalidExchangeTypeRaisesConnectionClosed(BlockingTestCaseBase):
    def test(self):
        """BlockingConnection: ConnectionClosed raised when creating exchange with invalid type"""  # pylint: disable=C0301
        # This test exploits behavior specific to RabbitMQ whereby the broker
        # closes the connection if an attempt is made to declare an exchange
        # with an invalid exchange type
        connection = self._connect()
        ch = connection.channel()

        exg_name = ("TestInvalidExchangeTypeRaisesConnectionClosed_" +
                    uuid.uuid1().hex)

        with self.assertRaises(pika.exceptions.ConnectionClosed) as ex_cm:
            # Attempt to create an exchange with invalid exchange type
            ch.exchange_declare(exg_name, exchange_type='ZZwwInvalid')

        self.assertEqual(ex_cm.exception.args[0], 503)


class TestCreateAndCloseConnectionWithChannelAndConsumer(BlockingTestCaseBase):

    def test(self):
        """BlockingConnection: Create and close connection with channel and consumer"""  # pylint: disable=C0301
        connection = self._connect()

        ch = connection.channel()

        q_name = (
            'TestCreateAndCloseConnectionWithChannelAndConsumer_q' +
            uuid.uuid1().hex)

        body1 = 'a' * 1024

        # Declare a new queue
        ch.queue_declare(q_name, auto_delete=True)
        self.addCleanup(lambda: self._connect().channel().queue_delete(q_name))

        # Publish the message to the queue by way of default exchange
        ch.basic_publish(exchange='', routing_key=q_name, body=body1)

        # Create a consumer that uses automatic ack mode
        ch.basic_consume(q_name, lambda *x: None, auto_ack=True,
                         exclusive=False, arguments=None)

        connection.close()
        self.assertTrue(connection.is_closed)
        self.assertFalse(connection.is_open)
        self.assertFalse(connection._impl.is_closing)

        self.assertFalse(connection._impl._channels)

        self.assertFalse(ch._consumer_infos)
        self.assertFalse(ch._impl._consumers)


class TestUsingInvalidQueueArgument(BlockingTestCaseBase):
    def test(self):
        """BlockingConnection raises expected exception when invalid queue parameter is used
        """
        connection = self._connect()
        ch = connection.channel()
        with self.assertRaises(TypeError):
            ch.queue_declare(queue=[1, 2, 3])


class TestSuddenBrokerDisconnectBeforeChannel(BlockingTestCaseBase):

    def test(self):
        """BlockingConnection resets properly on TCP/IP drop during channel()
        """
        with ForwardServer(remote_addr=(DEFAULT_PARAMS.host, DEFAULT_PARAMS.port),
                           local_linger_args=(1, 0)) as fwd:

            self.connection = self._connect(
                PARAMS_URL_TEMPLATE % {"port": fwd.server_address[1]})

        # Once outside the context, the connection is broken

        # BlockingConnection should raise ConnectionClosed
        with self.assertRaises(pika.exceptions.StreamLostError):
            self.connection.channel()

        self.assertTrue(self.connection.is_closed)
        self.assertFalse(self.connection.is_open)
        self.assertIsNone(self.connection._impl._transport)


class TestNoAccessToConnectionAfterConnectionLost(BlockingTestCaseBase):

    def test(self):
        """BlockingConnection no access file descriptor after StreamLostError
        """
        with ForwardServer(remote_addr=(DEFAULT_PARAMS.host, DEFAULT_PARAMS.port),
                           local_linger_args=(1, 0)) as fwd:

            self.connection = self._connect(
                PARAMS_URL_TEMPLATE % {"port": fwd.server_address[1]})

        # Once outside the context, the connection is broken

        # BlockingConnection should raise ConnectionClosed
        with self.assertRaises(pika.exceptions.StreamLostError):
            self.connection.channel()

        self.assertTrue(self.connection.is_closed)
        self.assertFalse(self.connection.is_open)
        self.assertIsNone(self.connection._impl._transport)

        # Attempt to operate on the connection once again after ConnectionClosed
        with self.assertRaises(pika.exceptions.ConnectionWrongStateError):
            self.connection.channel()


class TestConnectWithDownedBroker(BlockingTestCaseBase):

    def test(self):
        """ BlockingConnection to downed broker results in AMQPConnectionError

        """
        # Reserve a port for use in connect
        sock = socket.socket()
        self.addCleanup(sock.close)

        sock.bind(('127.0.0.1', 0))

        port = sock.getsockname()[1]

        sock.close()

        with self.assertRaises(pika.exceptions.AMQPConnectionError):
            self.connection = self._connect(
                PARAMS_URL_TEMPLATE % {"port": port})


class TestDisconnectDuringConnectionStart(BlockingTestCaseBase):

    def test(self):
        """ BlockingConnection TCP/IP connection loss in CONNECTION_START
        """
        fwd = ForwardServer(
            remote_addr=(DEFAULT_PARAMS.host, DEFAULT_PARAMS.port),
            local_linger_args=(1, 0))

        fwd.start()
        self.addCleanup(lambda: fwd.stop() if fwd.running else None)

        class MySelectConnection(pika.SelectConnection):
            assert hasattr(pika.SelectConnection, '_on_connection_start')

            def _on_connection_start(self, *args, **kwargs):  # pylint: disable=W0221
                fwd.stop()
                return super(MySelectConnection, self)._on_connection_start(
                    *args, **kwargs)

        with self.assertRaises(pika.exceptions.ProbableAuthenticationError):
            self._connect(
                PARAMS_URL_TEMPLATE % {"port": fwd.server_address[1]},
                impl_class=MySelectConnection)


class TestDisconnectDuringConnectionTune(BlockingTestCaseBase):

    def test(self):
        """ BlockingConnection TCP/IP connection loss in CONNECTION_TUNE
        """
        fwd = ForwardServer(
            remote_addr=(DEFAULT_PARAMS.host, DEFAULT_PARAMS.port),
            local_linger_args=(1, 0))
        fwd.start()
        self.addCleanup(lambda: fwd.stop() if fwd.running else None)

        class MySelectConnection(pika.SelectConnection):
            assert hasattr(pika.SelectConnection, '_on_connection_tune')

            def _on_connection_tune(self, *args, **kwargs):  # pylint: disable=W0221
                fwd.stop()
                return super(MySelectConnection, self)._on_connection_tune(
                    *args, **kwargs)

        with self.assertRaises(pika.exceptions.ProbableAccessDeniedError):
            self._connect(
                PARAMS_URL_TEMPLATE % {"port": fwd.server_address[1]},
                impl_class=MySelectConnection)


class TestDisconnectDuringConnectionProtocol(BlockingTestCaseBase):

    def test(self):
        """ BlockingConnection TCP/IP connection loss in CONNECTION_PROTOCOL
        """
        fwd = ForwardServer(
            remote_addr=(DEFAULT_PARAMS.host, DEFAULT_PARAMS.port),
            local_linger_args=(1, 0))

        fwd.start()
        self.addCleanup(lambda: fwd.stop() if fwd.running else None)

        class MySelectConnection(pika.SelectConnection):
            assert hasattr(pika.SelectConnection, '_on_stream_connected')

            def _on_stream_connected(self, *args, **kwargs):  # pylint: disable=W0221
                fwd.stop()
                return super(MySelectConnection, self)._on_stream_connected(
                    *args, **kwargs)

        with self.assertRaises(pika.exceptions.IncompatibleProtocolError):
            self._connect(PARAMS_URL_TEMPLATE % {"port": fwd.server_address[1]},
                          impl_class=MySelectConnection)


class TestProcessDataEvents(BlockingTestCaseBase):

    def test(self):
        """BlockingConnection.process_data_events"""
        connection = self._connect()

        # Try with time_limit=0
        start_time = time_now()
        connection.process_data_events(time_limit=0)
        elapsed = time_now() - start_time
        self.assertLess(elapsed, 0.25)

        # Try with time_limit=0.005
        start_time = time_now()
        connection.process_data_events(time_limit=0.005)
        elapsed = time_now() - start_time
        self.assertGreaterEqual(elapsed, 0.005)
        self.assertLess(elapsed, 0.25)


class TestConnectionRegisterForBlockAndUnblock(BlockingTestCaseBase):

    def test(self):
        """BlockingConnection register for Connection.Blocked/Unblocked"""
        connection = self._connect()

        # 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 and simulate incoming
        # blocked/unblocked frames

        blocked_buffer = []
        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 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):

    def test(self):
        """BlockingConnection Connection.Blocked timeout """
        url = DEFAULT_URL + '&blocked_connection_timeout=0.001'
        conn = self._connect(url=url)

        # NOTE: I haven't figured out yet how to coerce RabbitMQ to emit
        # Connection.Block and Connection.Unblock from the test, so we'll
        # simulate it for now

        # Simulate Connection.Blocked
        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.ConnectionBlockedTimeout):
            while True:
                conn.process_data_events(time_limit=1)


class TestAddCallbackThreadsafeFromSameThread(BlockingTestCaseBase):

    def test(self):
        """BlockingConnection.add_callback_threadsafe from same thread"""
        connection = self._connect()

        # Test timer completion
        start_time = time_now()
        rx_callback = []
        connection.add_callback_threadsafe(
            lambda: rx_callback.append(time_now()))
        while not rx_callback:
            connection.process_data_events(time_limit=None)

        self.assertEqual(len(rx_callback), 1)
        elapsed = time_now() - start_time
        self.assertLess(elapsed, 0.25)


class TestAddCallbackThreadsafeFromAnotherThread(BlockingTestCaseBase):

    def test(self):
        """BlockingConnection.add_callback_threadsafe from another thread"""
        connection = self._connect()

        # Test timer completion
        start_time = time_now()
        rx_callback = []
        timer = threading.Timer(
            0,
            functools.partial(connection.add_callback_threadsafe,
                              lambda: rx_callback.append(time_now())))
        self.addCleanup(timer.cancel)
        timer.start()
        while not rx_callback:
            connection.process_data_events(time_limit=None)

        self.assertEqual(len(rx_callback), 1)
        elapsed = time_now() - start_time
        self.assertLess(elapsed, 0.25)


class TestAddCallbackThreadsafeOnClosedConnectionRaisesWrongState(
        BlockingTestCaseBase):

    def test(self):
        """BlockingConnection.add_callback_threadsafe on closed connection raises ConnectionWrongStateError"""
        connection = self._connect()
        connection.close()

        with self.assertRaises(pika.exceptions.ConnectionWrongStateError):
            connection.add_callback_threadsafe(lambda: None)


class TestAddTimeoutRemoveTimeout(BlockingTestCaseBase):

    def test(self):
        """BlockingConnection.call_later and remove_timeout"""
        connection = self._connect()

        # Test timer completion
        start_time = time_now()
        rx_callback = []
        timer_id = connection.call_later(
            0.005,
            lambda: rx_callback.append(time_now()))
        while not rx_callback:
            connection.process_data_events(time_limit=None)

        self.assertEqual(len(rx_callback), 1)
        elapsed = time_now() - start_time
        self.assertLess(elapsed, 0.25)

        # Test removing triggered timeout
        connection.remove_timeout(timer_id)


        # Test aborted timer
        rx_callback = []
        timer_id = connection.call_later(
            0.001,
            lambda: rx_callback.append(time_now()))
        connection.remove_timeout(timer_id)
        connection.process_data_events(time_limit=0.1)
        self.assertFalse(rx_callback)

        # Make sure _TimerEvt repr doesn't crash
        evt = blocking_connection._TimerEvt(lambda: None)
        repr(evt)


class TestViabilityOfMultipleTimeoutsWithSameDeadlineAndCallback(BlockingTestCaseBase):

    def test(self):
        """BlockingConnection viability of multiple timeouts with same deadline and callback"""
        connection = self._connect()

        rx_callback = []

        def callback():
            rx_callback.append(1)

        timer1 = connection.call_later(0, callback)
        timer2 = connection.call_later(0, callback)

        self.assertIsNot(timer1, timer2)

        connection.remove_timeout(timer1)

        # Wait for second timer to fire
        start_wait_time = time_now()
        while not rx_callback and time_now() - start_wait_time < 0.25:
            connection.process_data_events(time_limit=0.001)

        self.assertListEqual(rx_callback, [1])


class TestRemoveTimeoutFromTimeoutCallback(BlockingTestCaseBase):

    def test(self):
        """BlockingConnection.remove_timeout from timeout callback"""
        connection = self._connect()

        # Test timer completion
        timer_id1 = connection.call_later(5, lambda: 0/0)

        rx_timer2 = []
        def on_timer2():
            connection.remove_timeout(timer_id1)
            connection.remove_timeout(timer_id2)
            rx_timer2.append(1)

        timer_id2 = connection.call_later(0, on_timer2)

        while not rx_timer2:
            connection.process_data_events(time_limit=None)

        self.assertFalse(connection._ready_events)


class TestSleep(BlockingTestCaseBase):

    def test(self):
        """BlockingConnection.sleep"""
        connection = self._connect()

        # Try with duration=0
        start_time = time_now()
        connection.sleep(duration=0)
        elapsed = time_now() - start_time
        self.assertLess(elapsed, 0.25)

        # Try with duration=0.005
        start_time = time_now()
        connection.sleep(duration=0.005)
        elapsed = time_now() - start_time
        self.assertGreaterEqual(elapsed, 0.005)
        self.assertLess(elapsed, 0.25)


class TestConnectionProperties(BlockingTestCaseBase):

    def test(self):
        """Test BlockingConnection properties"""
        connection = self._connect()

        self.assertTrue(connection.is_open)
        self.assertFalse(connection._impl.is_closing)
        self.assertFalse(connection.is_closed)

        self.assertTrue(connection.basic_nack_supported)
        self.assertTrue(connection.consumer_cancel_notify_supported)
        self.assertTrue(connection.exchange_exchange_bindings_supported)
        self.assertTrue(connection.publisher_confirms_supported)

        connection.close()
        self.assertFalse(connection.is_open)
        self.assertFalse(connection._impl.is_closing)
        self.assertTrue(connection.is_closed)



class TestCreateAndCloseChannel(BlockingTestCaseBase):

    def test(self):
        """BlockingChannel: Create and close channel"""
        connection = self._connect()

        ch = connection.channel()
        self.assertIsInstance(ch, blocking_connection.BlockingChannel)
        self.assertTrue(ch.is_open)
        self.assertFalse(ch.is_closed)
        self.assertFalse(ch._impl.is_closing)
        self.assertIs(ch.connection, connection)

        ch.close()
        self.assertTrue(ch.is_closed)
        self.assertFalse(ch.is_open)
        self.assertFalse(ch._impl.is_closing)


class TestExchangeDeclareAndDelete(BlockingTestCaseBase):

    def test(self):
        """BlockingChannel: Test exchange_declare and exchange_delete"""
        connection = self._connect()

        ch = connection.channel()

        name = "TestExchangeDeclareAndDelete_" + uuid.uuid1().hex

        # Declare a new exchange
        frame = ch.exchange_declare(name, exchange_type=ExchangeType.direct)
        self.addCleanup(connection.channel().exchange_delete, name)

        self.assertIsInstance(frame.method, pika.spec.Exchange.DeclareOk)

        # Check if it exists by declaring it passively
        frame = ch.exchange_declare(name, passive=True)
        self.assertIsInstance(frame.method, pika.spec.Exchange.DeclareOk)

        # Delete the exchange
        frame = ch.exchange_delete(name)
        self.assertIsInstance(frame.method, pika.spec.Exchange.DeleteOk)

        # Verify that it's been deleted
        with self.assertRaises(pika.exceptions.ChannelClosedByBroker) as cm:
            ch.exchange_declare(name, passive=True)

        self.assertEqual(cm.exception.args[0], 404)


class TestExchangeBindAndUnbind(BlockingTestCaseBase):

    def test(self):
        """BlockingChannel: Test exchange_bind and exchange_unbind"""
        connection = self._connect()

        ch = connection.channel()

        q_name = 'TestExchangeBindAndUnbind_q' + uuid.uuid1().hex
        src_exg_name = 'TestExchangeBindAndUnbind_src_exg_' + uuid.uuid1().hex
        dest_exg_name = 'TestExchangeBindAndUnbind_dest_exg_' + uuid.uuid1().hex
        routing_key = 'TestExchangeBindAndUnbind'

        # Place channel in publisher-acknowledgments mode so that we may test
        # whether the queue is reachable by publishing with mandatory=True
        res = ch.confirm_delivery()
        self.assertIsNone(res)

        # Declare both exchanges
        ch.exchange_declare(src_exg_name, exchange_type=ExchangeType.direct)
        self.addCleanup(connection.channel().exchange_delete, src_exg_name)
        ch.exchange_declare(dest_exg_name, exchange_type=ExchangeType.direct)
        self.addCleanup(connection.channel().exchange_delete, dest_exg_name)

        # Declare a new queue
        ch.queue_declare(q_name, auto_delete=True)
        self.addCleanup(lambda: self._connect().channel().queue_delete(q_name))

        # Bind the queue to the destination exchange
        ch.queue_bind(q_name, exchange=dest_exg_name, routing_key=routing_key)

        # Verify that the queue is unreachable without exchange-exchange binding
        with self.assertRaises(pika.exceptions.UnroutableError):
            ch.basic_publish(src_exg_name, routing_key, body='', mandatory=True)

        # Bind the exchanges
        frame = ch.exchange_bind(destination=dest_exg_name, source=src_exg_name,
                                 routing_key=routing_key)
        self.assertIsInstance(frame.method, pika.spec.Exchange.BindOk)

        # Publish a message via the source exchange
        ch.basic_publish(src_exg_name, routing_key, body='TestExchangeBindAndUnbind',
                         mandatory=True)

        # Check that the queue now has one message
        self._assert_exact_message_count_with_retries(channel=ch,
                                                      queue=q_name,
                                                      expected_count=1)

        # Unbind the exchanges
        frame = ch.exchange_unbind(destination=dest_exg_name,
                                   source=src_exg_name,
                                   routing_key=routing_key)
        self.assertIsInstance(frame.method, pika.spec.Exchange.UnbindOk)

        # Verify that the queue is now unreachable via the source exchange
        with self.assertRaises(pika.exceptions.UnroutableError):
            ch.basic_publish(src_exg_name, routing_key, body='', mandatory=True)


class TestQueueDeclareAndDelete(BlockingTestCaseBase):

    def test(self):
        """BlockingChannel: Test queue_declare and queue_delete"""
        connection = self._connect()

        ch = connection.channel()

        q_name = 'TestQueueDeclareAndDelete_' + uuid.uuid1().hex

        # Declare a new queue
        frame = ch.queue_declare(q_name, auto_delete=True)
        self.addCleanup(lambda: self._connect().channel().queue_delete(q_name))

        self.assertIsInstance(frame.method, pika.spec.Queue.DeclareOk)

        # Check if it exists by declaring it passively
        frame = ch.queue_declare(q_name, passive=True)
        self.assertIsInstance(frame.method, pika.spec.Queue.DeclareOk)

        # Delete the queue
        frame = ch.queue_delete(q_name)
        self.assertIsInstance(frame.method, pika.spec.Queue.DeleteOk)

        # Verify that it's been deleted
        with self.assertRaises(pika.exceptions.ChannelClosedByBroker) as cm:
            ch.queue_declare(q_name, passive=True)

        self.assertEqual(cm.exception.args[0], 404)


class TestPassiveQueueDeclareOfUnknownQueueRaisesChannelClosed(
        BlockingTestCaseBase):
    def test(self):
        """BlockingChannel: ChannelClosed raised when passive-declaring unknown queue"""  # pylint: disable=C0301
        connection = self._connect()
        ch = connection.channel()

        q_name = ("TestPassiveQueueDeclareOfUnknownQueueRaisesChannelClosed_q_"
                  + uuid.uuid1().hex)

        with self.assertRaises(pika.exceptions.ChannelClosedByBroker) as ex_cm:
            ch.queue_declare(q_name, passive=True)

        self.assertEqual(ex_cm.exception.args[0], 404)


class TestQueueBindAndUnbindAndPurge(BlockingTestCaseBase):

    def test(self):
        """BlockingChannel: Test queue_bind and queue_unbind"""
        connection = self._connect()

        ch = connection.channel()

        q_name = 'TestQueueBindAndUnbindAndPurge_q' + uuid.uuid1().hex
        exg_name = 'TestQueueBindAndUnbindAndPurge_exg_' + uuid.uuid1().hex
        routing_key = 'TestQueueBindAndUnbindAndPurge'

        # Place channel in publisher-acknowledgments mode so that we may test
        # whether the queue is reachable by publishing with mandatory=True
        res = ch.confirm_delivery()
        self.assertIsNone(res)

        # Declare a new exchange
        ch.exchange_declare(exg_name, exchange_type=ExchangeType.direct)
        self.addCleanup(connection.channel().exchange_delete, exg_name)

        # Declare a new queue
        ch.queue_declare(q_name, auto_delete=True)
        self.addCleanup(lambda: self._connect().channel().queue_delete(q_name))

        # Bind the queue to the exchange using routing key
        frame = ch.queue_bind(q_name, exchange=exg_name,
                              routing_key=routing_key)
        self.assertIsInstance(frame.method, pika.spec.Queue.BindOk)

        # Check that the queue is empty
        frame = ch.queue_declare(q_name, passive=True)
        self.assertEqual(frame.method.message_count, 0)

        # Deposit a message in the queue
        ch.basic_publish(exg_name, routing_key, body='TestQueueBindAndUnbindAndPurge',
                         mandatory=True)

        # Check that the queue now has one message
        frame = ch.queue_declare(q_name, passive=True)
        self.assertEqual(frame.method.message_count, 1)

        # Unbind the queue
        frame = ch.queue_unbind(queue=q_name, exchange=exg_name,
                                routing_key=routing_key)
        self.assertIsInstance(frame.method, pika.spec.Queue.UnbindOk)

        # Verify that the queue is now unreachable via that binding
        with self.assertRaises(pika.exceptions.UnroutableError):
            ch.basic_publish(exg_name, routing_key,
                             body='TestQueueBindAndUnbindAndPurge-2',
                             mandatory=True)

        # Purge the queue and verify that 1 message was purged
        frame = ch.queue_purge(q_name)
        self.assertIsInstance(frame.method, pika.spec.Queue.PurgeOk)
        self.assertEqual(frame.method.message_count, 1)

        # Verify that the queue is now empty
        frame = ch.queue_declare(q_name, passive=True)
        self.assertEqual(frame.method.message_count, 0)


class TestBasicGet(BlockingTestCaseBase):

    def tearDown(self):
        LOGGER.info('%s TEARING DOWN (%s)', datetime.utcnow(), self)

    def test(self):
        """BlockingChannel.basic_get"""
        LOGGER.info('%s STARTED (%s)', datetime.utcnow(), self)

        connection = self._connect()
        LOGGER.info('%s CONNECTED (%s)', datetime.utcnow(), self)

        ch = connection.channel()
        LOGGER.info('%s CREATED CHANNEL (%s)', datetime.utcnow(), self)

        q_name = 'TestBasicGet_q' + uuid.uuid1().hex

        # Place channel in publisher-acknowledgments mode so that the message
        # may be delivered synchronously to the queue by publishing it with
        # mandatory=True
        ch.confirm_delivery()
        LOGGER.info('%s ENABLED PUB-ACKS (%s)', datetime.utcnow(), self)

        # Declare a new queue
        ch.queue_declare(q_name, auto_delete=True)
        self.addCleanup(lambda: self._connect().channel().queue_delete(q_name))
        LOGGER.info('%s DECLARED QUEUE (%s)', datetime.utcnow(), self)

        # Verify result of getting a message from an empty queue
        msg = ch.basic_get(q_name, auto_ack=False)
        self.assertTupleEqual(msg, (None, None, None))
        LOGGER.info('%s GOT FROM EMPTY QUEUE (%s)', datetime.utcnow(), self)

        body = 'TestBasicGet'
        # Deposit a message in the queue via default exchange
        ch.basic_publish(exchange='', routing_key=q_name,
                         body=body, mandatory=True)
        LOGGER.info('%s PUBLISHED (%s)', datetime.utcnow(), self)

        # Get the message
        (method, properties, body) = ch.basic_get(q_name, auto_ack=False)
        LOGGER.info('%s GOT FROM NON-EMPTY QUEUE (%s)', datetime.utcnow(), self)
        self.assertIsInstance(method, pika.spec.Basic.GetOk)
        self.assertEqual(method.delivery_tag, 1)
        self.assertFalse(method.redelivered)
        self.assertEqual(method.exchange, '')
        self.assertEqual(method.routing_key, q_name)
        self.assertEqual(method.message_count, 0)

        self.assertIsInstance(properties, pika.BasicProperties)
        self.assertIsNone(properties.headers)
        self.assertEqual(body, as_bytes(body))

        # Ack it
        ch.basic_ack(delivery_tag=method.delivery_tag)
        LOGGER.info('%s ACKED (%s)', datetime.utcnow(), self)

        # Verify that the queue is now empty
        self._assert_exact_message_count_with_retries(channel=ch,
                                                      queue=q_name,
                                                      expected_count=0)


class TestBasicReject(BlockingTestCaseBase):

    def test(self):
        """BlockingChannel.basic_reject"""
        connection = self._connect()

        ch = connection.channel()

        q_name = 'TestBasicReject_q' + uuid.uuid1().hex

        # Place channel in publisher-acknowledgments mode so that the message
        # may be delivered synchronously to the queue by publishing it with
        # mandatory=True
        ch.confirm_delivery()

        # Declare a new queue
        ch.queue_declare(q_name, auto_delete=True)
        self.addCleanup(lambda: self._connect().channel().queue_delete(q_name))

        # Deposit two messages in the queue via default exchange
        ch.basic_publish(exchange='', routing_key=q_name,
                         body='TestBasicReject1', mandatory=True)
        ch.basic_publish(exchange='', routing_key=q_name,
                         body='TestBasicReject2', mandatory=True)

        # Get the messages
        (rx_method, _, rx_body) = ch.basic_get(q_name, auto_ack=False)
        self.assertEqual(rx_body, as_bytes('TestBasicReject1'))

        (rx_method, _, rx_body) = ch.basic_get(q_name, auto_ack=False)
        self.assertEqual(rx_body, as_bytes('TestBasicReject2'))

        # Nack the second message
        ch.basic_reject(rx_method.delivery_tag, requeue=True)

        # Verify that exactly one message is present in the queue, namely the
        # second one
        self._assert_exact_message_count_with_retries(channel=ch,
                                                      queue=q_name,
                                                      expected_count=1)
        (rx_method, _, rx_body) = ch.basic_get(q_name, auto_ack=False)
        self.assertEqual(rx_body, as_bytes('TestBasicReject2'))


class TestBasicRejectNoRequeue(BlockingTestCaseBase):

    def test(self):
        """BlockingChannel.basic_reject with requeue=False"""
        connection = self._connect()

        ch = connection.channel()

        q_name = 'TestBasicRejectNoRequeue_q' + uuid.uuid1().hex

        # Place channel in publisher-acknowledgments mode so that the message
        # may be delivered synchronously to the queue by publishing it with
        # mandatory=True
        ch.confirm_delivery()

        # Declare a new queue
        ch.queue_declare(q_name, auto_delete=True)
        self.addCleanup(lambda: self._connect().channel().queue_delete(q_name))

        # Deposit two messages in the queue via default exchange
        ch.basic_publish(exchange='', routing_key=q_name,
                         body='TestBasicRejectNoRequeue1', mandatory=True)
        ch.basic_publish(exchange='', routing_key=q_name,
                         body='TestBasicRejectNoRequeue2', mandatory=True)

        # Get the messages
        (rx_method, _, rx_body) = ch.basic_get(q_name, auto_ack=False)
        self.assertEqual(rx_body,
                         as_bytes('TestBasicRejectNoRequeue1'))

        (rx_method, _, rx_body) = ch.basic_get(q_name, auto_ack=False)
        self.assertEqual(rx_body,
                         as_bytes('TestBasicRejectNoRequeue2'))

        # Nack the second message
        ch.basic_reject(rx_method.delivery_tag, requeue=False)

        # Verify that no messages are present in the queue
        self._assert_exact_message_count_with_retries(channel=ch,
                                                      queue=q_name,
                                                      expected_count=0)


class TestBasicNack(BlockingTestCaseBase):

    def test(self):
        """BlockingChannel.basic_nack single message"""
        connection = self._connect()

        ch = connection.channel()

        q_name = 'TestBasicNack_q' + uuid.uuid1().hex

        # Place channel in publisher-acknowledgments mode so that the message
        # may be delivered synchronously to the queue by publishing it with
        # mandatory=True
        ch.confirm_delivery()

        # Declare a new queue
        ch.queue_declare(q_name, auto_delete=True)
        self.addCleanup(lambda: self._connect().channel().queue_delete(q_name))

        # Deposit two messages in the queue via default exchange
        ch.basic_publish(exchange='', routing_key=q_name,
                         body='TestBasicNack1', mandatory=True)
        ch.basic_publish(exchange='', routing_key=q_name,
                         body='TestBasicNack2', mandatory=True)

        # Get the messages
        (rx_method, _, rx_body) = ch.basic_get(q_name, auto_ack=False)
        self.assertEqual(rx_body, as_bytes('TestBasicNack1'))

        (rx_method, _, rx_body) = ch.basic_get(q_name, auto_ack=False)
        self.assertEqual(rx_body, as_bytes('TestBasicNack2'))

        # Nack the second message
        ch.basic_nack(rx_method.delivery_tag, multiple=False, requeue=True)

        # Verify that exactly one message is present in the queue, namely the
        # second one
        self._assert_exact_message_count_with_retries(channel=ch,
                                                      queue=q_name,
                                                      expected_count=1)
        (rx_method, _, rx_body) = ch.basic_get(q_name, auto_ack=False)
        self.assertEqual(rx_body, as_bytes('TestBasicNack2'))


class TestBasicNackNoRequeue(BlockingTestCaseBase):

    def test(self):
        """BlockingChannel.basic_nack with requeue=False"""
        connection = self._connect()

        ch = connection.channel()

        q_name = 'TestBasicNackNoRequeue_q' + uuid.uuid1().hex

        # Place channel in publisher-acknowledgments mode so that the message
        # may be delivered synchronously to the queue by publishing it with
        # mandatory=True
        ch.confirm_delivery()

        # Declare a new queue
        ch.queue_declare(q_name, auto_delete=True)
        self.addCleanup(lambda: self._connect().channel().queue_delete(q_name))

        # Deposit two messages in the queue via default exchange
        ch.basic_publish(exchange='', routing_key=q_name,
                         body='TestBasicNackNoRequeue1', mandatory=True)
        ch.basic_publish(exchange='', routing_key=q_name,
                         body='TestBasicNackNoRequeue2', mandatory=True)

        # Get the messages
        (rx_method, _, rx_body) = ch.basic_get(q_name, auto_ack=False)
        self.assertEqual(rx_body,
                         as_bytes('TestBasicNackNoRequeue1'))

        (rx_method, _, rx_body) = ch.basic_get(q_name, auto_ack=False)
        self.assertEqual(rx_body,
                         as_bytes('TestBasicNackNoRequeue2'))

        # Nack the second message
        ch.basic_nack(rx_method.delivery_tag, requeue=False)

        # Verify that no messages are present in the queue
        self._assert_exact_message_count_with_retries(channel=ch,
                                                      queue=q_name,
                                                      expected_count=0)


class TestBasicNackMultiple(BlockingTestCaseBase):

    def test(self):
        """BlockingChannel.basic_nack multiple messages"""
        connection = self._connect()

        ch = connection.channel()

        q_name = 'TestBasicNackMultiple_q' + uuid.uuid1().hex

        # Place channel in publisher-acknowledgments mode so that the message
        # may be delivered synchronously to the queue by publishing it with
        # mandatory=True
        ch.confirm_delivery()

        # Declare a new queue
        ch.queue_declare(q_name, auto_delete=True)
        self.addCleanup(lambda: self._connect().channel().queue_delete(q_name))

        # Deposit two messages in the queue via default exchange
        ch.basic_publish(exchange='', routing_key=q_name,
                         body='TestBasicNackMultiple1', mandatory=True)
        ch.basic_publish(exchange='', routing_key=q_name,
                         body='TestBasicNackMultiple2', mandatory=True)

        # Get the messages
        (rx_method, _, rx_body) = ch.basic_get(q_name, auto_ack=False)
        self.assertEqual(rx_body,
                         as_bytes('TestBasicNackMultiple1'))

        (rx_method, _, rx_body) = ch.basic_get(q_name, auto_ack=False)
        self.assertEqual(rx_body,
                         as_bytes('TestBasicNackMultiple2'))

        # Nack both messages via the "multiple" option
        ch.basic_nack(rx_method.delivery_tag, multiple=True, requeue=True)

        # Verify that both messages are present in the queue
        self._assert_exact_message_count_with_retries(channel=ch,
                                                      queue=q_name,
                                                      expected_count=2)
        (rx_method, _, rx_body) = ch.basic_get(q_name, auto_ack=False)
        self.assertEqual(rx_body,
                         as_bytes('TestBasicNackMultiple1'))
        (rx_method, _, rx_body) = ch.basic_get(q_name, auto_ack=False)
        self.assertEqual(rx_body,
                         as_bytes('TestBasicNackMultiple2'))


class TestBasicRecoverWithRequeue(BlockingTestCaseBase):

    def test(self):
        """BlockingChannel.basic_recover with requeue=True.

        NOTE: the requeue=False option is not supported by RabbitMQ broker as
        of this writing (using RabbitMQ 3.5.1)
        """
        connection = self._connect()

        ch = connection.channel()

        q_name = (
            'TestBasicRecoverWithRequeue_q' + uuid.uuid1().hex)

        # Place channel in publisher-acknowledgments mode so that the message
        # may be delivered synchronously to the queue by publishing it with
        # mandatory=True
        ch.confirm_delivery()

        # Declare a new queue
        ch.queue_declare(q_name, auto_delete=True)
        self.addCleanup(lambda: self._connect().channel().queue_delete(q_name))

        # Deposit two messages in the queue via default exchange
        ch.basic_publish(exchange='', routing_key=q_name,
                         body='TestBasicRecoverWithRequeue1', mandatory=True)
        ch.basic_publish(exchange='', routing_key=q_name,
                         body='TestBasicRecoverWithRequeue2', mandatory=True)

        rx_messages = []
        num_messages = 0
        for msg in ch.consume(q_name, auto_ack=False):
            num_messages += 1

            if num_messages == 2:
                ch.basic_recover(requeue=True)

            if num_messages > 2:
                rx_messages.append(msg)

            if num_messages == 4:
                break
        else:
            self.fail('consumer aborted prematurely')

        # Get the messages
        (_, _, rx_body) = rx_messages[0]
        self.assertEqual(rx_body,
                         as_bytes('TestBasicRecoverWithRequeue1'))

        (_, _, rx_body) = rx_messages[1]
        self.assertEqual(rx_body,
                         as_bytes('TestBasicRecoverWithRequeue2'))


class TestTxCommit(BlockingTestCaseBase):

    def test(self):
        """BlockingChannel.tx_commit"""
        connection = self._connect()

        ch = connection.channel()

        q_name = 'TestTxCommit_q' + uuid.uuid1().hex

        # Declare a new queue
        ch.queue_declare(q_name, auto_delete=True)
        self.addCleanup(lambda: self._connect().channel().queue_delete(q_name))

        # Select standard transaction mode
        frame = ch.tx_select()
        self.assertIsInstance(frame.method, pika.spec.Tx.SelectOk)

        # Deposit a message in the queue via default exchange
        ch.basic_publish(exchange='', routing_key=q_name,
                         body='TestTxCommit1', mandatory=True)

        # Verify that queue is still empty
        frame = ch.queue_declare(q_name, passive=True)
        self.assertEqual(frame.method.message_count, 0)

        # Commit the transaction
        ch.tx_commit()

        # Verify that the queue has the expected message
        frame = ch.queue_declare(q_name, passive=True)
        self.assertEqual(frame.method.message_count, 1)

        (_, _, rx_body) = ch.basic_get(q_name, auto_ack=False)
        self.assertEqual(rx_body, as_bytes('TestTxCommit1'))


class TestTxRollback(BlockingTestCaseBase):

    def test(self):
        """BlockingChannel.tx_commit"""
        connection = self._connect()

        ch = connection.channel()

        q_name = 'TestTxRollback_q' + uuid.uuid1().hex

        # Declare a new queue
        ch.queue_declare(q_name, auto_delete=True)
        self.addCleanup(lambda: self._connect().channel().queue_delete(q_name))

        # Select standard transaction mode
        frame = ch.tx_select()
        self.assertIsInstance(frame.method, pika.spec.Tx.SelectOk)

        # Deposit a message in the queue via default exchange
        ch.basic_publish(exchange='', routing_key=q_name,
                         body='TestTxRollback1', mandatory=True)

        # Verify that queue is still empty
        frame = ch.queue_declare(q_name, passive=True)
        self.assertEqual(frame.method.message_count, 0)

        # Roll back the transaction
        ch.tx_rollback()

        # Verify that the queue continues to be empty
        frame = ch.queue_declare(q_name, passive=True)
        self.assertEqual(frame.method.message_count, 0)


class TestBasicConsumeFromUnknownQueueRaisesChannelClosed(BlockingTestCaseBase):
    def test(self):
        """ChannelClosed raised when consuming from unknown queue"""
        connection = self._connect()
        ch = connection.channel()

        q_name = ("TestBasicConsumeFromUnknownQueueRaisesChannelClosed_q_" +
                  uuid.uuid1().hex)

        with self.assertRaises(pika.exceptions.ChannelClosedByBroker) as ex_cm:
            ch.basic_consume(q_name, lambda *args: None)

        self.assertEqual(ex_cm.exception.args[0], 404)


class TestPublishAndBasicPublishWithPubacksUnroutable(BlockingTestCaseBase):

    def test(self):  # pylint: disable=R0914
        """BlockingChannel.publish amd basic_publish unroutable message with pubacks"""  # pylint: disable=C0301
        connection = self._connect()

        ch = connection.channel()

        exg_name = ('TestPublishAndBasicPublishUnroutable_exg_' +
                    uuid.uuid1().hex)
        routing_key = 'TestPublishAndBasicPublishUnroutable'

        # Place channel in publisher-acknowledgments mode so that publishing
        # with mandatory=True will be synchronous
        res = ch.confirm_delivery()
        self.assertIsNone(res)

        # Declare a new exchange
        ch.exchange_declare(exg_name, exchange_type=ExchangeType.direct)
        self.addCleanup(connection.channel().exchange_delete, exg_name)

        # Verify unroutable message handling using basic_publish
        msg2_headers = dict(
            test_name='TestPublishAndBasicPublishWithPubacksUnroutable')
        msg2_properties = pika.spec.BasicProperties(headers=msg2_headers)
        with self.assertRaises(pika.exceptions.UnroutableError) as cm:
            ch.basic_publish(exg_name, routing_key=routing_key, body='',
                             properties=msg2_properties, mandatory=True)
        (msg,) = cm.exception.messages
        self.assertIsInstance(msg, blocking_connection.ReturnedMessage)
        self.assertIsInstance(msg.method, pika.spec.Basic.Return)
        self.assertEqual(msg.method.reply_code, 312)
        self.assertEqual(msg.method.exchange, exg_name)
        self.assertEqual(msg.method.routing_key, routing_key)
        self.assertIsInstance(msg.properties, pika.BasicProperties)
        self.assertEqual(msg.properties.headers, msg2_headers)
        self.assertEqual(msg.body, as_bytes(''))


class TestConfirmDeliveryAfterUnroutableMessage(BlockingTestCaseBase):

    def test(self):  # pylint: disable=R0914
        """BlockingChannel.confirm_delivery following unroutable message"""
        connection = self._connect()

        ch = connection.channel()

        exg_name = ('TestConfirmDeliveryAfterUnroutableMessage_exg_' +
                    uuid.uuid1().hex)
        routing_key = 'TestConfirmDeliveryAfterUnroutableMessage'

        # Declare a new exchange
        ch.exchange_declare(exg_name, exchange_type=ExchangeType.direct)
        self.addCleanup(connection.channel().exchange_delete, exg_name)

        # Register on-return callback
        returned_messages = []
        ch.add_on_return_callback(lambda *args: returned_messages.append(args))

        # Emit unroutable message without pubacks
        ch.basic_publish(exg_name, routing_key=routing_key, body='', mandatory=True)

        # Select delivery confirmations
        ch.confirm_delivery()

        # Verify that unroutable message is in pending events
        self.assertEqual(len(ch._pending_events), 1)
        self.assertIsInstance(ch._pending_events[0],
                              blocking_connection._ReturnedMessageEvt)
        # Verify that repr of _ReturnedMessageEvt instance does crash
        repr(ch._pending_events[0])

        # Dispach events
        connection.process_data_events()

        self.assertEqual(len(ch._pending_events), 0)

        # Verify that unroutable message was dispatched
        ((channel, method, properties, body,),) = returned_messages  # pylint: disable=E0632
        self.assertIs(channel, ch)
        self.assertIsInstance(method, pika.spec.Basic.Return)
        self.assertEqual(method.reply_code, 312)
        self.assertEqual(method.exchange, exg_name)
        self.assertEqual(method.routing_key, routing_key)
        self.assertIsInstance(properties, pika.BasicProperties)
        self.assertEqual(body, as_bytes(''))


class TestUnroutableMessagesReturnedInNonPubackMode(BlockingTestCaseBase):

    def test(self):  # pylint: disable=R0914
        """BlockingChannel: unroutable messages is returned in non-puback mode"""  # pylint: disable=C0301
        connection = self._connect()

        ch = connection.channel()

        exg_name = (
            'TestUnroutableMessageReturnedInNonPubackMode_exg_'
            + uuid.uuid1().hex)
        routing_key = 'TestUnroutableMessageReturnedInNonPubackMode'

        # Declare a new exchange
        ch.exchange_declare(exg_name, exchange_type=ExchangeType.direct)
        self.addCleanup(connection.channel().exchange_delete, exg_name)

        # Register on-return callback
        returned_messages = []
        ch.add_on_return_callback(
            lambda *args: returned_messages.append(args))

        # Emit unroutable messages without pubacks
        ch.basic_publish(exg_name, routing_key=routing_key, body='msg1', mandatory=True)
        ch.basic_publish(exg_name, routing_key=routing_key, body='msg2', mandatory=True)

        # Process I/O until Basic.Return are dispatched
        while len(returned_messages) < 2:
            connection.process_data_events()

        self.assertEqual(len(returned_messages), 2)

        self.assertEqual(len(ch._pending_events), 0)

        # Verify returned messages
        (channel, method, properties, body,) = returned_messages[0]
        self.assertIs(channel, ch)
        self.assertIsInstance(method, pika.spec.Basic.Return)
        self.assertEqual(method.reply_code, 312)
        self.assertEqual(method.exchange, exg_name)
        self.assertEqual(method.routing_key, routing_key)
        self.assertIsInstance(properties, pika.BasicProperties)
        self.assertEqual(body, as_bytes('msg1'))

        (channel, method, properties, body,) = returned_messages[1]
        self.assertIs(channel, ch)
        self.assertIsInstance(method, pika.spec.Basic.Return)
        self.assertEqual(method.reply_code, 312)
        self.assertEqual(method.exchange, exg_name)
        self.assertEqual(method.routing_key, routing_key)
        self.assertIsInstance(properties, pika.BasicProperties)
        self.assertEqual(body, as_bytes('msg2'))


class TestUnroutableMessageReturnedInPubackMode(BlockingTestCaseBase):

    def test(self):  # pylint: disable=R0914
        """BlockingChannel: unroutable messages is returned in puback mode"""
        connection = self._connect()

        ch = connection.channel()

        exg_name = (
            'TestUnroutableMessageReturnedInPubackMode_exg_'
            + uuid.uuid1().hex)
        routing_key = 'TestUnroutableMessageReturnedInPubackMode'

        # Declare a new exchange
        ch.exchange_declare(exg_name, exchange_type=ExchangeType.direct)
        self.addCleanup(connection.channel().exchange_delete, exg_name)

        # Select delivery confirmations
        ch.confirm_delivery()

        # Register on-return callback
        returned_messages = []
        ch.add_on_return_callback(
            lambda *args: returned_messages.append(args))

        # Emit unroutable messages with pubacks
        with self.assertRaises(pika.exceptions.UnroutableError):
            ch.basic_publish(exg_name, routing_key=routing_key, body='msg1',
                             mandatory=True)
        with self.assertRaises(pika.exceptions.UnroutableError):
            ch.basic_publish(exg_name, routing_key=routing_key, body='msg2',
                             mandatory=True)

        # Verify that unroutable messages are already in pending events
        self.assertEqual(len(ch._pending_events), 2)
        self.assertIsInstance(ch._pending_events[0],
                              blocking_connection._ReturnedMessageEvt)
        self.assertIsInstance(ch._pending_events[1],
                              blocking_connection._ReturnedMessageEvt)
        # Verify that repr of _ReturnedMessageEvt instance does crash
        repr(ch._pending_events[0])
        repr(ch._pending_events[1])

        # Dispatch events
        connection.process_data_events()

        self.assertEqual(len(ch._pending_events), 0)

        # Verify returned messages
        (channel, method, properties, body,) = returned_messages[0]
        self.assertIs(channel, ch)
        self.assertIsInstance(method, pika.spec.Basic.Return)
        self.assertEqual(method.reply_code, 312)
        self.assertEqual(method.exchange, exg_name)
        self.assertEqual(method.routing_key, routing_key)
        self.assertIsInstance(properties, pika.BasicProperties)
        self.assertEqual(body, as_bytes('msg1'))

        (channel, method, properties, body,) = returned_messages[1]
        self.assertIs(channel, ch)
        self.assertIsInstance(method, pika.spec.Basic.Return)
        self.assertEqual(method.reply_code, 312)
        self.assertEqual(method.exchange, exg_name)
        self.assertEqual(method.routing_key, routing_key)
        self.assertIsInstance(properties, pika.BasicProperties)
        self.assertEqual(body, as_bytes('msg2'))


class TestBasicPublishDeliveredWhenPendingUnroutable(BlockingTestCaseBase):

    def test(self):  # pylint: disable=R0914
        """BlockingChannel.basic_publish msg delivered despite pending unroutable message"""  # pylint: disable=C0301
        connection = self._connect()

        ch = connection.channel()

        q_name = ('TestBasicPublishDeliveredWhenPendingUnroutable_q' +
                  uuid.uuid1().hex)
        exg_name = ('TestBasicPublishDeliveredWhenPendingUnroutable_exg_' +
                    uuid.uuid1().hex)
        routing_key = 'TestBasicPublishDeliveredWhenPendingUnroutable'

        # Declare a new exchange
        ch.exchange_declare(exg_name, exchange_type=ExchangeType.direct)
        self.addCleanup(connection.channel().exchange_delete, exg_name)

        # Declare a new queue
        ch.queue_declare(q_name, auto_delete=True)
        self.addCleanup(lambda: self._connect().channel().queue_delete(q_name))

        # Bind the queue to the exchange using routing key
        ch.queue_bind(q_name, exchange=exg_name, routing_key=routing_key)

        # Attempt to send an unroutable message in the queue via basic_publish
        ch.basic_publish(exg_name, routing_key='',
                         body='unroutable-message',
                         mandatory=True)

        # Flush connection to force Basic.Return
        connection.channel().close()

        # Deposit a routable message in the queue
        ch.basic_publish(exg_name, routing_key=routing_key,
                         body='routable-message',
                         mandatory=True)

        # Wait for the queue to get the routable message
        self._assert_exact_message_count_with_retries(channel=ch,
                                                      queue=q_name,
                                                      expected_count=1)

        msg = ch.basic_get(q_name)

        # Check the first message
        self.assertIsInstance(msg, tuple)
        rx_method, rx_properties, rx_body = msg
        self.assertIsInstance(rx_method, pika.spec.Basic.GetOk)
        self.assertEqual(rx_method.delivery_tag, 1)
        self.assertFalse(rx_method.redelivered)
        self.assertEqual(rx_method.exchange, exg_name)
        self.assertEqual(rx_method.routing_key, routing_key)

        self.assertIsInstance(rx_properties, pika.BasicProperties)
        self.assertEqual(rx_body, as_bytes('routable-message'))

        # There shouldn't be any more events now
        self.assertFalse(ch._pending_events)

        # Ack the message
        ch.basic_ack(delivery_tag=rx_method.delivery_tag, multiple=False)

        # Verify that the queue is now empty
        self._assert_exact_message_count_with_retries(channel=ch,
                                                      queue=q_name,
                                                      expected_count=0)


class TestPublishAndConsumeWithPubacksAndQosOfOne(BlockingTestCaseBase):

    def test(self):  # pylint: disable=R0914,R0915
        """BlockingChannel.basic_publish, publish, basic_consume, QoS, \
        Basic.Cancel from broker
        """
        connection = self._connect()

        ch = connection.channel()

        q_name = 'TestPublishAndConsumeAndQos_q' + uuid.uuid1().hex
        exg_name = 'TestPublishAndConsumeAndQos_exg_' + uuid.uuid1().hex
        routing_key = 'TestPublishAndConsumeAndQos'

        # Place channel in publisher-acknowledgments mode so that publishing
        # with mandatory=True will be synchronous
        res = ch.confirm_delivery()
        self.assertIsNone(res)

        # Declare a new exchange
        ch.exchange_declare(exg_name, exchange_type=ExchangeType.direct)
        self.addCleanup(connection.channel().exchange_delete, exg_name)

        # Declare a new queue
        ch.queue_declare(q_name, auto_delete=True)
        self.addCleanup(lambda: self._connect().channel().queue_delete(q_name))

        # Bind the queue to the exchange using routing key
        ch.queue_bind(q_name, exchange=exg_name, routing_key=routing_key)

        # Deposit a message in the queue
        msg1_headers = dict(
            test_name='TestPublishAndConsumeWithPubacksAndQosOfOne')
        msg1_properties = pika.spec.BasicProperties(headers=msg1_headers)
        ch.basic_publish(exg_name, routing_key=routing_key,
                         body='via-basic_publish',
                         properties=msg1_properties,
                         mandatory=True)

        # Deposit another message in the queue
        ch.basic_publish(exg_name, routing_key, body='via-publish',
                         mandatory=True)

        # Check that the queue now has two messages
        frame = ch.queue_declare(q_name, passive=True)
        self.assertEqual(frame.method.message_count, 2)

        # Configure QoS for one message
        ch.basic_qos(prefetch_size=0, prefetch_count=1, global_qos=False)

        # Create a consumer
        rx_messages = []
        consumer_tag = ch.basic_consume(
            q_name,
            lambda *args: rx_messages.append(args),
            auto_ack=False,
            exclusive=False,
            arguments=None)

        # Wait for first message to arrive
        while not rx_messages:
            connection.process_data_events(time_limit=None)

        self.assertEqual(len(rx_messages), 1)

        # Check the first message
        msg = rx_messages[0]
        self.assertIsInstance(msg, tuple)
        rx_ch, rx_method, rx_properties, rx_body = msg
        self.assertIs(rx_ch, ch)
        self.assertIsInstance(rx_method, pika.spec.Basic.Deliver)
        self.assertEqual(rx_method.consumer_tag, consumer_tag)
        self.assertEqual(rx_method.delivery_tag, 1)
        self.assertFalse(rx_method.redelivered)
        self.assertEqual(rx_method.exchange, exg_name)
        self.assertEqual(rx_method.routing_key, routing_key)

        self.assertIsInstance(rx_properties, pika.BasicProperties)
        self.assertEqual(rx_properties.headers, msg1_headers)
        self.assertEqual(rx_body, as_bytes('via-basic_publish'))

        # There shouldn't be any more events now
        self.assertFalse(ch._pending_events)

        # Ack the message so that the next one can arrive (we configured QoS
        # with prefetch_count=1)
        ch.basic_ack(delivery_tag=rx_method.delivery_tag, multiple=False)

        # Get the second message
        while len(rx_messages) < 2:
            connection.process_data_events(time_limit=None)

        self.assertEqual(len(rx_messages), 2)

        msg = rx_messages[1]
        self.assertIsInstance(msg, tuple)
        rx_ch, rx_method, rx_properties, rx_body = msg
        self.assertIs(rx_ch, ch)
        self.assertIsInstance(rx_method, pika.spec.Basic.Deliver)
        self.assertEqual(rx_method.consumer_tag, consumer_tag)
        self.assertEqual(rx_method.delivery_tag, 2)
        self.assertFalse(rx_method.redelivered)
        self.assertEqual(rx_method.exchange, exg_name)
        self.assertEqual(rx_method.routing_key, routing_key)

        self.assertIsInstance(rx_properties, pika.BasicProperties)
        self.assertEqual(rx_body, as_bytes('via-publish'))

        # There shouldn't be any more events now
        self.assertFalse(ch._pending_events)

        ch.basic_ack(delivery_tag=rx_method.delivery_tag, multiple=False)

        # Verify that the queue is now empty
        self._assert_exact_message_count_with_retries(channel=ch,
                                                      queue=q_name,
                                                      expected_count=0)

        # Attempt to consume again with a short timeout
        connection.process_data_events(time_limit=0.005)
        self.assertEqual(len(rx_messages), 2)

        # Delete the queue and wait for consumer cancellation
        rx_cancellations = []
        ch.add_on_cancel_callback(rx_cancellations.append)
        ch.queue_delete(q_name)
        ch.start_consuming()

        self.assertEqual(len(rx_cancellations), 1)
        frame, = rx_cancellations  # pylint: disable=E0632
        self.assertEqual(frame.method.consumer_tag, consumer_tag)


class TestBasicConsumeWithAckFromAnotherThread(BlockingTestCaseBase):

    def test(self):  # pylint: disable=R0914,R0915
        """BlockingChannel.basic_consume with ack from another thread and \
        requesting basic_ack via add_callback_threadsafe
        """
        # This test simulates processing of a message on another thread and
        # then requesting an ACK to be dispatched on the connection's thread
        # via BlockingConnection.add_callback_threadsafe

        connection = self._connect()

        ch = connection.channel()

        q_name = 'TestBasicConsumeWithAckFromAnotherThread_q' + uuid.uuid1().hex
        exg_name = ('TestBasicConsumeWithAckFromAnotherThread_exg' +
                    uuid.uuid1().hex)
        routing_key = 'TestBasicConsumeWithAckFromAnotherThread'

        # Place channel in publisher-acknowledgments mode so that publishing
        # with mandatory=True will be synchronous (for convenience)
        res = ch.confirm_delivery()
        self.assertIsNone(res)

        # Declare a new exchange
        ch.exchange_declare(exg_name, exchange_type=ExchangeType.direct)
        self.addCleanup(connection.channel().exchange_delete, exg_name)

        # Declare a new queue
        ch.queue_declare(q_name, auto_delete=True)
        self.addCleanup(lambda: self._connect().channel().queue_delete(q_name))

        # Bind the queue to the exchange using routing key
        ch.queue_bind(q_name, exchange=exg_name, routing_key=routing_key)

        # Publish 2 messages with mandatory=True for synchronous processing
        ch.basic_publish(exg_name, routing_key, body='msg1', mandatory=True)
        ch.basic_publish(exg_name, routing_key, body='last-msg', mandatory=True)

        # Configure QoS for one message so that the 2nd message will be
        # delivered only after the 1st one is ACKed
        ch.basic_qos(prefetch_size=0, prefetch_count=1, global_qos=False)

        # Create a consumer
        rx_messages = []
        def ackAndEnqueueMessageViaAnotherThread(rx_ch,
                                                 rx_method,
                                                 rx_properties, # pylint: disable=W0613
                                                 rx_body):
            LOGGER.debug(
                '%s: Got message body=%r; delivery-tag=%r',
                datetime.now(), rx_body, rx_method.delivery_tag)

            # Request ACK dispatch via add_callback_threadsafe from other
            # thread; if last message, cancel consumer so that start_consuming
            # can return

            def processOnConnectionThread():
                LOGGER.debug('%s: ACKing message body=%r; delivery-tag=%r',
                             datetime.now(),
                             rx_body,
                             rx_method.delivery_tag)
                ch.basic_ack(delivery_tag=rx_method.delivery_tag,
                             multiple=False)
                rx_messages.append(rx_body)

                # NOTE on python3, `b'last-msg' != 'last-msg'`
                if rx_body == b'last-msg':
                    LOGGER.debug('%s: Canceling consumer consumer-tag=%r',
                                 datetime.now(),
                                 rx_method.consumer_tag)
                    rx_ch.basic_cancel(rx_method.consumer_tag)

            # Spawn a thread to initiate ACKing
            timer = threading.Timer(0,
                                    lambda: connection.add_callback_threadsafe(
                                        processOnConnectionThread))
            self.addCleanup(timer.cancel)
            timer.start()

        consumer_tag = ch.basic_consume(
            q_name,
            ackAndEnqueueMessageViaAnotherThread,
            auto_ack=False,
            exclusive=False,
            arguments=None)

        # Wait for both messages
        LOGGER.debug('%s: calling start_consuming(); consumer tag=%r',
                     datetime.now(),
                     consumer_tag)
        ch.start_consuming()
        LOGGER.debug('%s: Returned from start_consuming(); consumer tag=%r',
                     datetime.now(),
                     consumer_tag)

        self.assertEqual(len(rx_messages), 2)
        self.assertEqual(rx_messages[0], b'msg1')
        self.assertEqual(rx_messages[1], b'last-msg')


class TestConsumeGeneratorWithAckFromAnotherThread(BlockingTestCaseBase):

    def test(self):  # pylint: disable=R0914,R0915
        """BlockingChannel.consume and requesting basic_ack from another \
        thread via add_callback_threadsafe
        """
        connection = self._connect()

        ch = connection.channel()

        q_name = ('TestConsumeGeneratorWithAckFromAnotherThread_q' +
                  uuid.uuid1().hex)
        exg_name = ('TestConsumeGeneratorWithAckFromAnotherThread_exg' +
                    uuid.uuid1().hex)
        routing_key = 'TestConsumeGeneratorWithAckFromAnotherThread'

        # Place channel in publisher-acknowledgments mode so that publishing
        # with mandatory=True will be synchronous (for convenience)
        res = ch.confirm_delivery()
        self.assertIsNone(res)

        # Declare a new exchange
        ch.exchange_declare(exg_name, exchange_type=ExchangeType.direct)
        self.addCleanup(connection.channel().exchange_delete, exg_name)

        # Declare a new queue
        ch.queue_declare(q_name, auto_delete=True)
        self.addCleanup(lambda: self._connect().channel().queue_delete(q_name))

        # Bind the queue to the exchange using routing key
        ch.queue_bind(q_name, exchange=exg_name, routing_key=routing_key)

        # Publish 2 messages with mandatory=True for synchronous processing
        ch.basic_publish(exg_name, routing_key, body='msg1', mandatory=True)
        ch.basic_publish(exg_name, routing_key, body='last-msg', mandatory=True)

        # Configure QoS for one message so that the 2nd message will be
        # delivered only after the 1st one is ACKed
        ch.basic_qos(prefetch_size=0, prefetch_count=1, global_qos=False)

        # Create a consumer
        rx_messages = []
        def ackAndEnqueueMessageViaAnotherThread(rx_ch,
                                                 rx_method,
                                                 rx_properties, # pylint: disable=W0613
                                                 rx_body):
            LOGGER.debug(
                '%s: Got message body=%r; delivery-tag=%r',
                datetime.now(), rx_body, rx_method.delivery_tag)

            # Request ACK dispatch via add_callback_threadsafe from other
            # thread; if last message, cancel consumer so that consumer
            # generator completes

            def processOnConnectionThread():
                LOGGER.debug('%s: ACKing message body=%r; delivery-tag=%r',
                             datetime.now(),
                             rx_body,
                             rx_method.delivery_tag)
                ch.basic_ack(delivery_tag=rx_method.delivery_tag,
                             multiple=False)
                rx_messages.append(rx_body)

                # NOTE on python3, `b'last-msg' != 'last-msg'`
                if rx_body == b'last-msg':
                    LOGGER.debug('%s: Canceling consumer consumer-tag=%r',
                                 datetime.now(),
                                 rx_method.consumer_tag)
                    # NOTE Need to use cancel() for the consumer generator
                    # instead of basic_cancel()
                    rx_ch.cancel()

            # Spawn a thread to initiate ACKing
            timer = threading.Timer(0,
                                    lambda: connection.add_callback_threadsafe(
                                        processOnConnectionThread))
            self.addCleanup(timer.cancel)
            timer.start()

        for method, properties, body in ch.consume(q_name, auto_ack=False):
            ackAndEnqueueMessageViaAnotherThread(rx_ch=ch,
                                                 rx_method=method,
                                                 rx_properties=properties,
                                                 rx_body=body)

        self.assertEqual(len(rx_messages), 2)
        self.assertEqual(rx_messages[0], b'msg1')
        self.assertEqual(rx_messages[1], b'last-msg')


class TestTwoBasicConsumersOnSameChannel(BlockingTestCaseBase):

    def test(self):  # pylint: disable=R0914
        """BlockingChannel: two basic_consume consumers on same channel
        """
        connection = self._connect()

        ch = connection.channel()

        exg_name = 'TestPublishAndConsumeAndQos_exg_' + uuid.uuid1().hex
        q1_name = 'TestTwoBasicConsumersOnSameChannel_q1' + uuid.uuid1().hex
        q2_name = 'TestTwoBasicConsumersOnSameChannel_q2' + uuid.uuid1().hex
        q1_routing_key = 'TestTwoBasicConsumersOnSameChannel1'
        q2_routing_key = 'TestTwoBasicConsumersOnSameChannel2'

        # Place channel in publisher-acknowledgments mode so that publishing
        # with mandatory=True will be synchronous
        ch.confirm_delivery()

        # Declare a new exchange
        ch.exchange_declare(exg_name, exchange_type=ExchangeType.direct)
        self.addCleanup(connection.channel().exchange_delete, exg_name)

        # Declare the two new queues and bind them to the exchange
        ch.queue_declare(q1_name, auto_delete=True)
        self.addCleanup(lambda: self._connect().channel().queue_delete(q1_name))
        ch.queue_bind(q1_name, exchange=exg_name, routing_key=q1_routing_key)

        ch.queue_declare(q2_name, auto_delete=True)
        self.addCleanup(lambda: self._connect().channel().queue_delete(q2_name))
        ch.queue_bind(q2_name, exchange=exg_name, routing_key=q2_routing_key)

        # Deposit messages in the queues
        q1_tx_message_bodies = ['q1_message+%s' % (i,)
                                for i in pika.compat.xrange(100)]
        for message_body in q1_tx_message_bodies:
            ch.basic_publish(exg_name, q1_routing_key, body=message_body, mandatory=True)

        q2_tx_message_bodies = ['q2_message+%s' % (i,)
                                for i in pika.compat.xrange(150)]
        for message_body in q2_tx_message_bodies:
            ch.basic_publish(exg_name, q2_routing_key, body=message_body, mandatory=True)

        # Create the consumers
        q1_rx_messages = []
        q1_consumer_tag = ch.basic_consume(
            q1_name,
            lambda *args: q1_rx_messages.append(args),
            auto_ack=False,
            exclusive=False,
            arguments=None)

        q2_rx_messages = []
        q2_consumer_tag = ch.basic_consume(
            q2_name,
            lambda *args: q2_rx_messages.append(args),
            auto_ack=False,
            exclusive=False,
            arguments=None)

        # Wait for all messages to be delivered
        while (len(q1_rx_messages) < len(q1_tx_message_bodies) or
               len(q2_rx_messages) < len(q2_tx_message_bodies)):
            connection.process_data_events(time_limit=None)

        self.assertEqual(len(q2_rx_messages), len(q2_tx_message_bodies))

        # Verify the messages
        def validate_messages(rx_messages,
                              routing_key,
                              consumer_tag,
                              tx_message_bodies):
            self.assertEqual(len(rx_messages), len(tx_message_bodies))

            for msg, expected_body in zip(rx_messages, tx_message_bodies):
                self.assertIsInstance(msg, tuple)
                rx_ch, rx_method, rx_properties, rx_body = msg
                self.assertIs(rx_ch, ch)
                self.assertIsInstance(rx_method, pika.spec.Basic.Deliver)
                self.assertEqual(rx_method.consumer_tag, consumer_tag)
                self.assertFalse(rx_method.redelivered)
                self.assertEqual(rx_method.exchange, exg_name)
                self.assertEqual(rx_method.routing_key, routing_key)

                self.assertIsInstance(rx_properties, pika.BasicProperties)
                self.assertEqual(rx_body, as_bytes(expected_body))

        # Validate q1 consumed messages
        validate_messages(rx_messages=q1_rx_messages,
                          routing_key=q1_routing_key,
                          consumer_tag=q1_consumer_tag,
                          tx_message_bodies=q1_tx_message_bodies)

        # Validate q2 consumed messages
        validate_messages(rx_messages=q2_rx_messages,
                          routing_key=q2_routing_key,
                          consumer_tag=q2_consumer_tag,
                          tx_message_bodies=q2_tx_message_bodies)

        # There shouldn't be any more events now
        self.assertFalse(ch._pending_events)


class TestBasicCancelPurgesPendingConsumerCancellationEvt(BlockingTestCaseBase):

    def test(self):
        """BlockingChannel.basic_cancel purges pending _ConsumerCancellationEvt""" # pylint: disable=C0301
        connection = self._connect()

        ch = connection.channel()

        q_name = ('TestBasicCancelPurgesPendingConsumerCancellationEvt_q' +
                  uuid.uuid1().hex)

        ch.queue_declare(q_name)

        self.addCleanup(lambda: self._connect().channel().queue_delete(q_name))

        ch.basic_publish('', routing_key=q_name, body='via-publish', mandatory=True)

        # Create a consumer. Not passing a 'callback' to test client-generated
        # consumer tags
        rx_messages = []
        consumer_tag = ch.basic_consume(
            q_name,
            lambda *args: rx_messages.append(args))

        # Wait for the published message to arrive, but don't consume it
        while not ch._pending_events:
            # Issue synchronous command that forces processing of incoming I/O
            connection.channel().close()

        self.assertEqual(len(ch._pending_events), 1)
        self.assertIsInstance(ch._pending_events[0],
                              blocking_connection._ConsumerDeliveryEvt)

        # Delete the queue and wait for broker-initiated consumer cancellation
        ch.queue_delete(q_name)
        while len(ch._pending_events) < 2:
            # Issue synchronous command that forces processing of incoming I/O
            connection.channel().close()

        self.assertEqual(len(ch._pending_events), 2)
        self.assertIsInstance(ch._pending_events[1],
                              blocking_connection._ConsumerCancellationEvt)

        # Issue consumer cancellation and verify that the pending
        # _ConsumerCancellationEvt instance was removed
        messages = ch.basic_cancel(consumer_tag)
        self.assertEqual(messages, [])

        self.assertEqual(len(ch._pending_events), 0)


class TestBasicPublishWithoutPubacks(BlockingTestCaseBase):

    def test(self):  # pylint: disable=R0914,R0915
        """BlockingChannel.basic_publish without pubacks"""
        connection = self._connect()

        ch = connection.channel()

        q_name = 'TestBasicPublishWithoutPubacks_q' + uuid.uuid1().hex
        exg_name = 'TestBasicPublishWithoutPubacks_exg_' + uuid.uuid1().hex
        routing_key = 'TestBasicPublishWithoutPubacks'

        # Declare a new exchange
        ch.exchange_declare(exg_name, exchange_type=ExchangeType.direct)
        self.addCleanup(connection.channel().exchange_delete, exg_name)

        # Declare a new queue
        ch.queue_declare(q_name, auto_delete=True)
        self.addCleanup(lambda: self._connect().channel().queue_delete(q_name))

        # Bind the queue to the exchange using routing key
        ch.queue_bind(q_name, exchange=exg_name, routing_key=routing_key)

        # Deposit a message in the queue with mandatory=True
        msg1_headers = dict(
            test_name='TestBasicPublishWithoutPubacks')
        msg1_properties = pika.spec.BasicProperties(headers=msg1_headers)
        ch.basic_publish(exg_name, routing_key=routing_key,
                         body='via-basic_publish_mandatory=True',
                         properties=msg1_properties,
                         mandatory=True)

        # Deposit a message in the queue with mandatory=False
        ch.basic_publish(exg_name, routing_key=routing_key,
                         body='via-basic_publish_mandatory=False',
                         mandatory=False)

        # Wait for the messages to arrive in queue
        self._assert_exact_message_count_with_retries(channel=ch,
                                                      queue=q_name,
                                                      expected_count=2)

        # Create a consumer. Not passing a 'callback' to test client-generated
        # consumer tags
        rx_messages = []
        consumer_tag = ch.basic_consume(
            q_name,
            lambda *args: rx_messages.append(args),
            auto_ack=False,
            exclusive=False,
            arguments=None)

        # Wait for first message to arrive
        while not rx_messages:
            connection.process_data_events(time_limit=None)

        self.assertGreaterEqual(len(rx_messages), 1)

        # Check the first message
        msg = rx_messages[0]
        self.assertIsInstance(msg, tuple)
        rx_ch, rx_method, rx_properties, rx_body = msg
        self.assertIs(rx_ch, ch)
        self.assertIsInstance(rx_method, pika.spec.Basic.Deliver)
        self.assertEqual(rx_method.consumer_tag, consumer_tag)
        self.assertEqual(rx_method.delivery_tag, 1)
        self.assertFalse(rx_method.redelivered)
        self.assertEqual(rx_method.exchange, exg_name)
        self.assertEqual(rx_method.routing_key, routing_key)

        self.assertIsInstance(rx_properties, pika.BasicProperties)
        self.assertEqual(rx_properties.headers, msg1_headers)
        self.assertEqual(rx_body, as_bytes('via-basic_publish_mandatory=True'))

        # There shouldn't be any more events now
        self.assertFalse(ch._pending_events)

        # Ack the message so that the next one can arrive (we configured QoS
        # with prefetch_count=1)
        ch.basic_ack(delivery_tag=rx_method.delivery_tag, multiple=False)

        # Get the second message
        while len(rx_messages) < 2:
            connection.process_data_events(time_limit=None)

        self.assertEqual(len(rx_messages), 2)

        msg = rx_messages[1]
        self.assertIsInstance(msg, tuple)
        rx_ch, rx_method, rx_properties, rx_body = msg
        self.assertIs(rx_ch, ch)
        self.assertIsInstance(rx_method, pika.spec.Basic.Deliver)
        self.assertEqual(rx_method.consumer_tag, consumer_tag)
        self.assertEqual(rx_method.delivery_tag, 2)
        self.assertFalse(rx_method.redelivered)
        self.assertEqual(rx_method.exchange, exg_name)
        self.assertEqual(rx_method.routing_key, routing_key)

        self.assertIsInstance(rx_properties, pika.BasicProperties)
        self.assertEqual(rx_body, as_bytes('via-basic_publish_mandatory=False'))

        # There shouldn't be any more events now
        self.assertFalse(ch._pending_events)

        ch.basic_ack(delivery_tag=rx_method.delivery_tag, multiple=False)

        # Verify that the queue is now empty
        self._assert_exact_message_count_with_retries(channel=ch,
                                                      queue=q_name,
                                                      expected_count=0)

        # Attempt to consume again with a short timeout
        connection.process_data_events(time_limit=0.005)
        self.assertEqual(len(rx_messages), 2)


class TestPublishFromBasicConsumeCallback(BlockingTestCaseBase):

    def test(self):
        """BlockingChannel.basic_publish from basic_consume callback
        """
        connection = self._connect()

        ch = connection.channel()

        src_q_name = (
            'TestPublishFromBasicConsumeCallback_src_q' + uuid.uuid1().hex)
        dest_q_name = (
            'TestPublishFromBasicConsumeCallback_dest_q' + uuid.uuid1().hex)

        # Place channel in publisher-acknowledgments mode so that publishing
        # with mandatory=True will be synchronous
        ch.confirm_delivery()

        # Declare source and destination queues
        ch.queue_declare(src_q_name, auto_delete=True)
        self.addCleanup(lambda: self._connect().channel().queue_delete(src_q_name))
        ch.queue_declare(dest_q_name, auto_delete=True)
        self.addCleanup(lambda: self._connect().channel().queue_delete(dest_q_name))

        # Deposit a message in the source queue
        ch.basic_publish('',
                         routing_key=src_q_name,
                         body='via-publish',
                         mandatory=True)

        # Create a consumer
        def on_consume(channel, method, props, body):
            channel.basic_publish(
                '', routing_key=dest_q_name, body=body,
                properties=props, mandatory=True)
            channel.basic_ack(method.delivery_tag)

        ch.basic_consume(src_q_name,
                         on_consume,
                         auto_ack=False,
                         exclusive=False,
                         arguments=None)

        # Consume from destination queue
        for _, _, rx_body in ch.consume(dest_q_name, auto_ack=True):
            self.assertEqual(rx_body, as_bytes('via-publish'))
            break
        else:
            self.fail('failed to consume a messages from destination q')


class TestStopConsumingFromBasicConsumeCallback(BlockingTestCaseBase):

    def test(self):
        """BlockingChannel.stop_consuming from basic_consume callback
        """
        connection = self._connect()

        ch = connection.channel()

        q_name = (
            'TestStopConsumingFromBasicConsumeCallback_q' + uuid.uuid1().hex)

        # Place channel in publisher-acknowledgments mode so that publishing
        # with mandatory=True will be synchronous
        ch.confirm_delivery()

        # Declare the queue
        ch.queue_declare(q_name, auto_delete=False)
        self.addCleanup(connection.channel().queue_delete, q_name)

        # Deposit two messages in the queue
        ch.basic_publish('',
                         routing_key=q_name,
                         body='via-publish1',
                         mandatory=True)

        ch.basic_publish('',
                         routing_key=q_name,
                         body='via-publish2',
                         mandatory=True)

        # Create a consumer
        def on_consume(channel, method, props, body):  # pylint: disable=W0613
            channel.stop_consuming()
            channel.basic_ack(method.delivery_tag)

        ch.basic_consume(q_name,
                         on_consume,
                         auto_ack=False,
                         exclusive=False,
                         arguments=None)

        ch.start_consuming()

        ch.close()

        ch = connection.channel()

        # Verify that only the second message is present in the queue
        _, _, rx_body = ch.basic_get(q_name)
        self.assertEqual(rx_body, as_bytes('via-publish2'))

        msg = ch.basic_get(q_name)
        self.assertTupleEqual(msg, (None, None, None))


class TestCloseChannelFromBasicConsumeCallback(BlockingTestCaseBase):

    def test(self):
        """BlockingChannel.close from basic_consume callback
        """
        connection = self._connect()

        ch = connection.channel()

        q_name = (
            'TestCloseChannelFromBasicConsumeCallback_q' + uuid.uuid1().hex)

        # Place channel in publisher-acknowledgments mode so that publishing
        # with mandatory=True will be synchronous
        ch.confirm_delivery()

        # Declare the queue
        ch.queue_declare(q_name, auto_delete=False)
        self.addCleanup(connection.channel().queue_delete, q_name)

        # Deposit two messages in the queue
        ch.basic_publish('',
                         routing_key=q_name,
                         body='via-publish1',
                         mandatory=True)

        ch.basic_publish('',
                         routing_key=q_name,
                         body='via-publish2',
                         mandatory=True)

        # Create a consumer
        def on_consume(channel, method, props, body):  # pylint: disable=W0613
            channel.close()

        ch.basic_consume(q_name,
                         on_consume,
                         auto_ack=False,
                         exclusive=False,
                         arguments=None)

        ch.start_consuming()

        self.assertTrue(ch.is_closed)


        # Verify that both messages are present in the queue
        ch = connection.channel()
        _, _, rx_body = ch.basic_get(q_name)
        self.assertEqual(rx_body, as_bytes('via-publish1'))
        _, _, rx_body = ch.basic_get(q_name)
        self.assertEqual(rx_body, as_bytes('via-publish2'))


class TestCloseConnectionFromBasicConsumeCallback(BlockingTestCaseBase):

    def test(self):
        """BlockingConnection.close from basic_consume callback
        """
        connection = self._connect()

        ch = connection.channel()

        q_name = (
            'TestCloseConnectionFromBasicConsumeCallback_q' + uuid.uuid1().hex)

        # Place channel in publisher-acknowledgments mode so that publishing
        # with mandatory=True will be synchronous
        ch.confirm_delivery()

        # Declare the queue
        ch.queue_declare(q_name, auto_delete=False)
        self.addCleanup(lambda: self._connect().channel().queue_delete(q_name))

        # Deposit two messages in the queue
        ch.basic_publish('',
                         routing_key=q_name,
                         body='via-publish1',
                         mandatory=True)

        ch.basic_publish('',
                         routing_key=q_name,
                         body='via-publish2',
                         mandatory=True)

        # Create a consumer
        def on_consume(channel, method, props, body):  # pylint: disable=W0613
            connection.close()

        ch.basic_consume(q_name,
                         on_consume,
                         auto_ack=False,
                         exclusive=False,
                         arguments=None)

        ch.start_consuming()

        self.assertTrue(ch.is_closed)
        self.assertTrue(connection.is_closed)


        # Verify that both messages are present in the queue
        ch = self._connect().channel()
        _, _, rx_body = ch.basic_get(q_name)
        self.assertEqual(rx_body, as_bytes('via-publish1'))
        _, _, rx_body = ch.basic_get(q_name)
        self.assertEqual(rx_body, as_bytes('via-publish2'))


class TestStartConsumingRaisesChannelClosedOnSameChannelFailure(BlockingTestCaseBase):

    def test(self):
        """start_consuming() exits with ChannelClosed exception on same channel failure
        """
        connection = self._connect()

        # Fail test if exception leaks back ito I/O loop
        self._instrument_io_loop_exception_leak_detection(connection)

        ch = connection.channel()

        q_name = (
            'TestStartConsumingPassesChannelClosedOnSameChannelFailure_q' +
            uuid.uuid1().hex)

        # Declare the queue
        ch.queue_declare(q_name, auto_delete=False)
        self.addCleanup(lambda: self._connect().channel().queue_delete(q_name))

        ch.basic_consume(q_name,
                         lambda *args, **kwargs: None,
                         auto_ack=False,
                         exclusive=False,
                         arguments=None)

        # Schedule a callback that will cause a channel error on the consumer's
        # channel by publishing to an unknown exchange. This will cause the
        # broker to close our channel.
        connection.add_callback_threadsafe(
            lambda: ch.basic_publish(
                exchange=q_name,
                routing_key='123',
                body=b'Nope this is wrong'))

        with self.assertRaises(pika.exceptions.ChannelClosedByBroker):
            ch.start_consuming()


class TestStartConsumingReturnsAfterCancelFromBroker(BlockingTestCaseBase):

    def test(self):
        """start_consuming() returns after Cancel from broker
        """
        connection = self._connect()

        ch = connection.channel()

        q_name = (
            'TestStartConsumingExitsOnCancelFromBroker_q' + uuid.uuid1().hex)

        # Declare the queue
        ch.queue_declare(q_name, auto_delete=False)
        self.addCleanup(lambda: self._connect().channel().queue_delete(q_name))

        consumer_tag = ch.basic_consume(q_name,
                                        lambda *args, **kwargs: None,
                                        auto_ack=False,
                                        exclusive=False,
                                        arguments=None)

        # Schedule a callback that will run while start_consuming() is
        # executing and delete the queue. This will cause the broker to cancel
        # our consumer
        connection.add_callback_threadsafe(
            lambda: self._connect().channel().queue_delete(q_name))

        ch.start_consuming()

        self.assertNotIn(consumer_tag, ch._consumer_infos)


class TestNonPubAckPublishAndConsumeHugeMessage(BlockingTestCaseBase):

    def test(self):
        """BlockingChannel.publish/consume huge message"""
        connection = self._connect()

        ch = connection.channel()

        q_name = 'TestPublishAndConsumeHugeMessage_q' + uuid.uuid1().hex
        body = 'a' * 1000000

        # Declare a new queue
        ch.queue_declare(q_name, auto_delete=False)
        self.addCleanup(lambda: self._connect().channel().queue_delete(q_name))

        # Publish a message to the queue by way of default exchange
        ch.basic_publish(exchange='', routing_key=q_name, body=body)
        LOGGER.info('Published message body size=%s', len(body))

        # Consume the message
        for rx_method, rx_props, rx_body in ch.consume(q_name, auto_ack=False,
                                                       exclusive=False,
                                                       arguments=None):
            self.assertIsInstance(rx_method, pika.spec.Basic.Deliver)
            self.assertEqual(rx_method.delivery_tag, 1)
            self.assertFalse(rx_method.redelivered)
            self.assertEqual(rx_method.exchange, '')
            self.assertEqual(rx_method.routing_key, q_name)

            self.assertIsInstance(rx_props, pika.BasicProperties)
            self.assertEqual(rx_body, as_bytes(body))

            # Ack the message
            ch.basic_ack(delivery_tag=rx_method.delivery_tag, multiple=False)

            break

        # There shouldn't be any more events now
        self.assertFalse(ch._queue_consumer_generator.pending_events)

        # Verify that the queue is now empty
        ch.close()
        ch = connection.channel()
        self._assert_exact_message_count_with_retries(channel=ch,
                                                      queue=q_name,
                                                      expected_count=0)


class TestNonPubAckPublishAndConsumeManyMessages(BlockingTestCaseBase):

    def test(self):
        """BlockingChannel non-pub-ack publish/consume many messages"""
        connection = self._connect()

        ch = connection.channel()

        q_name = ('TestNonPubackPublishAndConsumeManyMessages_q' +
                  uuid.uuid1().hex)
        body = 'b' * 1024

        num_messages_to_publish = 500

        # Declare a new queue
        ch.queue_declare(q_name, auto_delete=False)
        self.addCleanup(lambda: self._connect().channel().queue_delete(q_name))

        for _ in pika.compat.xrange(num_messages_to_publish):
            # Publish a message to the queue by way of default exchange
            ch.basic_publish(exchange='', routing_key=q_name, body=body)

        # Consume the messages
        num_consumed = 0
        for rx_method, rx_props, rx_body in ch.consume(q_name,
                                                       auto_ack=False,
                                                       exclusive=False,
                                                       arguments=None):
            num_consumed += 1
            self.assertIsInstance(rx_method, pika.spec.Basic.Deliver)
            self.assertEqual(rx_method.delivery_tag, num_consumed)
            self.assertFalse(rx_method.redelivered)
            self.assertEqual(rx_method.exchange, '')
            self.assertEqual(rx_method.routing_key, q_name)

            self.assertIsInstance(rx_props, pika.BasicProperties)
            self.assertEqual(rx_body, as_bytes(body))

            # Ack the message
            ch.basic_ack(delivery_tag=rx_method.delivery_tag, multiple=False)

            if num_consumed >= num_messages_to_publish:
                break

        # There shouldn't be any more events now
        self.assertFalse(ch._queue_consumer_generator.pending_events)

        ch.close()

        self.assertIsNone(ch._queue_consumer_generator)

        # Verify that the queue is now empty
        ch = connection.channel()
        self._assert_exact_message_count_with_retries(channel=ch,
                                                      queue=q_name,
                                                      expected_count=0)


class TestBasicCancelWithNonAckableConsumer(BlockingTestCaseBase):

    def test(self):
        """BlockingChannel user cancels non-ackable consumer via basic_cancel"""
        connection = self._connect()

        ch = connection.channel()

        q_name = (
            'TestBasicCancelWithNonAckableConsumer_q' + uuid.uuid1().hex)

        body1 = 'a' * 1024
        body2 = 'b' * 2048

        # Declare a new queue
        ch.queue_declare(q_name, auto_delete=False)
        self.addCleanup(lambda: self._connect().channel().queue_delete(q_name))

        # Publish two messages to the queue by way of default exchange
        ch.basic_publish(exchange='', routing_key=q_name, body=body1)
        ch.basic_publish(exchange='', routing_key=q_name, body=body2)

        # Wait for queue to contain both messages
        self._assert_exact_message_count_with_retries(channel=ch,
                                                      queue=q_name,
                                                      expected_count=2)

        # Create a consumer that uses automatic ack mode
        consumer_tag = ch.basic_consume(q_name, lambda *x: None, auto_ack=True,
                                        exclusive=False, arguments=None)

        # Wait for all messages to be sent by broker to client
        self._assert_exact_message_count_with_retries(channel=ch,
                                                      queue=q_name,
                                                      expected_count=0)

        # Cancel the consumer
        messages = ch.basic_cancel(consumer_tag)

        # Both messages should have been on their way when we cancelled
        self.assertEqual(len(messages), 2)

        _, _, rx_body1 = messages[0]
        self.assertEqual(rx_body1, as_bytes(body1))

        _, _, rx_body2 = messages[1]
        self.assertEqual(rx_body2, as_bytes(body2))

        ch.close()

        ch = connection.channel()

        # Verify that the queue is now empty
        frame = ch.queue_declare(q_name, passive=True)
        self.assertEqual(frame.method.message_count, 0)


class TestBasicCancelWithAckableConsumer(BlockingTestCaseBase):

    def test(self):
        """BlockingChannel user cancels ackable consumer via basic_cancel"""
        connection = self._connect()

        ch = connection.channel()

        q_name = (
            'TestBasicCancelWithAckableConsumer_q' + uuid.uuid1().hex)

        body1 = 'a' * 1024
        body2 = 'b' * 2048

        # Declare a new queue
        ch.queue_declare(q_name, auto_delete=False)
        self.addCleanup(lambda: self._connect().channel().queue_delete(q_name))

        # Publish two messages to the queue by way of default exchange
        ch.basic_publish(exchange='', routing_key=q_name, body=body1)
        ch.basic_publish(exchange='', routing_key=q_name, body=body2)

        # Wait for queue to contain both messages
        self._assert_exact_message_count_with_retries(channel=ch,
                                                      queue=q_name,
                                                      expected_count=2)

        # Create an ackable consumer
        consumer_tag = ch.basic_consume(q_name, lambda *x: None, auto_ack=False,
                                        exclusive=False, arguments=None)

        # Wait for all messages to be sent by broker to client
        self._assert_exact_message_count_with_retries(channel=ch,
                                                      queue=q_name,
                                                      expected_count=0)

        # Cancel the consumer
        messages = ch.basic_cancel(consumer_tag)

        # Both messages should have been on their way when we cancelled
        self.assertEqual(len(messages), 0)

        ch.close()

        ch = connection.channel()

        # Verify that canceling the ackable consumer restored both messages
        self._assert_exact_message_count_with_retries(channel=ch,
                                                      queue=q_name,
                                                      expected_count=2)


class TestUnackedMessageAutoRestoredToQueueOnChannelClose(BlockingTestCaseBase):

    def test(self):
        """BlockingChannel unacked message restored to q on channel close """
        connection = self._connect()

        ch = connection.channel()

        q_name = ('TestUnackedMessageAutoRestoredToQueueOnChannelClose_q' +
                  uuid.uuid1().hex)

        body1 = 'a' * 1024
        body2 = 'b' * 2048

        # Declare a new queue
        ch.queue_declare(q_name, auto_delete=False)
        self.addCleanup(lambda: self._connect().channel().queue_delete(q_name))

        # Publish two messages to the queue by way of default exchange
        ch.basic_publish(exchange='', routing_key=q_name, body=body1)
        ch.basic_publish(exchange='', routing_key=q_name, body=body2)

        # Consume the events, but don't ack
        rx_messages = []
        ch.basic_consume(q_name, lambda *args: rx_messages.append(args),
                         auto_ack=False, exclusive=False, arguments=None)
        while len(rx_messages) != 2:
            connection.process_data_events(time_limit=None)

        self.assertEqual(rx_messages[0][1].delivery_tag, 1)
        self.assertEqual(rx_messages[1][1].delivery_tag, 2)

        # Verify no more ready messages in queue
        frame = ch.queue_declare(q_name, passive=True)
        self.assertEqual(frame.method.message_count, 0)

        # Closing channel should restore messages back to queue
        ch.close()

        # Verify that there are two messages in q now
        ch = connection.channel()

        self._assert_exact_message_count_with_retries(channel=ch,
                                                      queue=q_name,
                                                      expected_count=2)


class TestNoAckMessageNotRestoredToQueueOnChannelClose(BlockingTestCaseBase):

    def test(self):
        """BlockingChannel unacked message restored to q on channel close """
        connection = self._connect()

        ch = connection.channel()

        q_name = ('TestNoAckMessageNotRestoredToQueueOnChannelClose_q' +
                  uuid.uuid1().hex)

        body1 = 'a' * 1024
        body2 = 'b' * 2048

        # Declare a new queue
        ch.queue_declare(q_name, auto_delete=False)
        self.addCleanup(lambda: self._connect().channel().queue_delete(q_name))

        # Publish two messages to the queue by way of default exchange
        ch.basic_publish(exchange='', routing_key=q_name, body=body1)
        ch.basic_publish(exchange='', routing_key=q_name, body=body2)

        # Consume, but don't ack
        num_messages = 0
        for rx_method, _, _ in ch.consume(q_name, auto_ack=True, exclusive=False):
            num_messages += 1

            self.assertEqual(rx_method.delivery_tag, num_messages)

            if num_messages == 2:
                break
        else:
            self.fail('expected 2 messages, but consumed %i' % (num_messages,))

        # Verify no more ready messages in queue
        frame = ch.queue_declare(q_name, passive=True)
        self.assertEqual(frame.method.message_count, 0)

        # Closing channel should not restore no-ack messages back to queue
        ch.close()

        # Verify that there are no messages in q now
        ch = connection.channel()
        frame = ch.queue_declare(q_name, passive=True)
        self.assertEqual(frame.method.message_count, 0)


class TestConsumeGeneratorInactivityTimeout(BlockingTestCaseBase):

    def test(self):
        """BlockingChannel consume returns 3-tuple of None values on inactivity timeout """
        connection = self._connect()

        ch = connection.channel()

        q_name = ('TestConsumeGeneratorInactivityTimeout_q' + uuid.uuid1().hex)

        # Declare a new queue
        ch.queue_declare(q_name, auto_delete=True)

        # Expect to get only (None, None, None) upon inactivity timeout, since
        # there are no messages in queue
        for msg in ch.consume(q_name, inactivity_timeout=0.1):
            self.assertEqual(msg, (None, None, None))
            break
        else:
            self.fail('expected (None, None, None), but iterator stopped')


class TestConsumeGeneratorInterruptedByCancelFromBroker(BlockingTestCaseBase):

    def test(self):
        """BlockingChannel consume generator is interrupted broker's Cancel """
        connection = self._connect()

        self.assertTrue(connection.consumer_cancel_notify_supported)

        ch = connection.channel()

        q_name = ('TestConsumeGeneratorInterruptedByCancelFromBroker_q' +
                  uuid.uuid1().hex)

        # Declare a new queue
        ch.queue_declare(q_name, auto_delete=True)

        queue_deleted = False
        for _ in ch.consume(q_name, auto_ack=False, inactivity_timeout=0.001):
            if not queue_deleted:
                # Delete the queue to force Basic.Cancel from the broker
                ch.queue_delete(q_name)
                queue_deleted = True

        self.assertIsNone(ch._queue_consumer_generator)


class TestConsumeGeneratorCancelEncountersCancelFromBroker(BlockingTestCaseBase):

    def test(self):
        """BlockingChannel consume generator cancel called when broker's Cancel is enqueued """
        connection = self._connect()

        self.assertTrue(connection.consumer_cancel_notify_supported)

        ch = connection.channel()

        q_name = ('TestConsumeGeneratorCancelEncountersCancelFromBroker_q' +
                  uuid.uuid1().hex)

        # Declare a new queue
        ch.queue_declare(q_name, auto_delete=True)

        for _ in ch.consume(q_name, auto_ack=False, inactivity_timeout=0.001):
            # Delete the queue to force Basic.Cancel from the broker
            ch.queue_delete(q_name)

            # Wait for server's Basic.Cancel
            while not ch._queue_consumer_generator.pending_events:
                connection.process_data_events()

            # Confirm it's Basic.Cancel
            self.assertIsInstance(ch._queue_consumer_generator.pending_events[0],
                                  blocking_connection._ConsumerCancellationEvt)

            # Now attempt to cancel the consumer generator
            ch.cancel()

            self.assertIsNone(ch._queue_consumer_generator)


class TestConsumeGeneratorPassesChannelClosedOnSameChannelFailure(BlockingTestCaseBase):

    def test(self):
        """consume() exits with ChannelClosed exception on same channel failure
        """
        connection = self._connect()

        # Fail test if exception leaks back ito I/O loop
        self._instrument_io_loop_exception_leak_detection(connection)

        ch = connection.channel()

        q_name = (
            'TestConsumeGeneratorPassesChannelClosedOnSameChannelFailure_q' +
            uuid.uuid1().hex)

        # Declare the queue
        ch.queue_declare(q_name, auto_delete=False)
        self.addCleanup(lambda: self._connect().channel().queue_delete(q_name))

        # Schedule a callback that will cause a channel error on the consumer's
        # channel by publishing to an unknown exchange. This will cause the
        # broker to close our channel.
        connection.add_callback_threadsafe(
            lambda: ch.basic_publish(
                exchange=q_name,
                routing_key='123',
                body=b'Nope this is wrong'))

        with self.assertRaises(pika.exceptions.ChannelClosedByBroker):
            for _ in ch.consume(q_name):
                pass


class TestChannelFlow(BlockingTestCaseBase):

    def test(self):
        """BlockingChannel Channel.Flow activate and deactivate """
        connection = self._connect()

        ch = connection.channel()

        q_name = ('TestChannelFlow_q' + uuid.uuid1().hex)

        # Declare a new queue
        ch.queue_declare(q_name, auto_delete=False)
        self.addCleanup(lambda: self._connect().channel().queue_delete(q_name))

        # Verify zero active consumers on the queue
        frame = ch.queue_declare(q_name, passive=True)
        self.assertEqual(frame.method.consumer_count, 0)

        # Create consumer
        ch.basic_consume(q_name, lambda *args: None)

        # Verify one active consumer on the queue now
        frame = ch.queue_declare(q_name, passive=True)
        self.assertEqual(frame.method.consumer_count, 1)

        # Activate flow from default state (active by default)
        active = ch.flow(True)
        self.assertEqual(active, True)

        # Verify still one active consumer on the queue now
        frame = ch.queue_declare(q_name, passive=True)
        self.assertEqual(frame.method.consumer_count, 1)


class TestChannelRaisesWrongStateWhenDeclaringQueueOnClosedChannel(BlockingTestCaseBase):
    def test(self):
        """BlockingConnection: Declaring queue on closed channel raises ChannelWrongStateError"""
        q_name = (
            'TestChannelRaisesWrongStateWhenDeclaringQueueOnClosedChannel_q' +
            uuid.uuid1().hex)

        channel = self._connect().channel()
        channel.close()
        with self.assertRaises(pika.exceptions.ChannelWrongStateError):
            channel.queue_declare(q_name)


class TestChannelRaisesWrongStateWhenClosingClosedChannel(BlockingTestCaseBase):
    def test(self):
        """BlockingConnection: Closing closed channel raises ChannelWrongStateError"""
        channel = self._connect().channel()
        channel.close()
        with self.assertRaises(pika.exceptions.ChannelWrongStateError):
            channel.close()


class TestChannelContextManagerClosesChannel(BlockingTestCaseBase):
    def test(self):
        """BlockingConnection: chanel context manager exit survives closed channel"""
        with self._connect().channel() as channel:
            self.assertTrue(channel.is_open)

        self.assertTrue(channel.is_closed)


class TestChannelContextManagerExitSurvivesClosedChannel(BlockingTestCaseBase):
    def test(self):
        """BlockingConnection: chanel context manager exit survives closed channel"""
        with self._connect().channel() as channel:
            self.assertTrue(channel.is_open)
            channel.close()
            self.assertTrue(channel.is_closed)

        self.assertTrue(channel.is_closed)


class TestChannelContextManagerDoesNotSuppressChannelClosedByBroker(
        BlockingTestCaseBase):
    def test(self):
        """BlockingConnection: chanel context manager doesn't suppress ChannelClosedByBroker exception"""

        exg_name = (
            "TestChannelContextManagerDoesNotSuppressChannelClosedByBroker" +
            uuid.uuid1().hex)

        with self.assertRaises(pika.exceptions.ChannelClosedByBroker):
            with self._connect().channel() as channel:
                # Passively declaring non-existent exchange should force broker
                # to close channel
                channel.exchange_declare(exg_name, passive=True)

        self.assertTrue(channel.is_closed)


if __name__ == '__main__':
    unittest.main()