<s> """<STR_LIT>""" <EOL> from collections import deque <EOL> import logging <EOL> try : <EOL> import mock <EOL> except ImportError : <EOL> from unittest import mock <EOL> try : <EOL> import unittest2 as unittest <EOL> except ImportError : <EOL> import unittest <EOL> from pika . adapters import blocking_connection <EOL> from pika import callback <EOL> from pika import channel <EOL> from pika import exceptions <EOL> from pika import frame <EOL> from pika import spec <EOL> BLOCKING_CHANNEL = '<STR_LIT>' <EOL> BLOCKING_CONNECTION = '<STR_LIT>' <EOL> class ChannelTemplate ( channel . Channel ) : <EOL> channel_number = <NUM_LIT:1> <EOL> class BlockingChannelTests ( unittest . TestCase ) : <EOL> @ mock . patch ( BLOCKING_CONNECTION ) <EOL> def _create_connection ( self , connection = None ) : <EOL> return connection <EOL> def setUp ( self ) : <EOL> self . connection = self . _create_connection ( ) <EOL> channelImplMock = mock . Mock ( spec = ChannelTemplate , <EOL> is_closing = False , <EOL> is_closed = False , <EOL> is_open = True ) <EOL> self . obj = blocking_connection . BlockingChannel ( channelImplMock , <EOL> self . connection ) <EOL> def tearDown ( self ) : <EOL> del self . connection <EOL> del self . obj <EOL> def test_init_initial_value_confirmation ( self ) : <EOL> self . assertFalse ( self . obj . _delivery_confirmation ) <EOL> def test_init_initial_value_pending_events ( self ) : <EOL> self . assertEqual ( self . obj . _pending_events , deque ( ) ) <EOL> def test_init_initial_value_buback_return ( self ) : <EOL> self . assertIsNone ( self . obj . _puback_return ) <EOL> def test_basic_consume ( self ) : <EOL> with mock . patch . object ( self . obj . _impl , '<STR_LIT>' ) : <EOL> self . obj . _impl . _generate_consumer_tag . return_value = '<STR_LIT>' <EOL> self . obj . _impl . basic_consume . return_value = '<STR_LIT>' <EOL> self . obj . basic_consume ( mock . Mock ( ) , "<STR_LIT>" ) <EOL> self . assertEqual ( self . obj . _consumer_infos [ '<STR_LIT>' ] . state , <EOL> blocking_connection . _ConsumerInfo . ACTIVE ) <EOL> def test_context_manager ( self ) : <EOL> with self . obj as channel : <EOL> self . assertFalse ( channel . _impl . close . called ) <EOL> channel . _impl . close . assert_called_once_with ( reply_code = <NUM_LIT:0> , reply_text = '<STR_LIT>' ) <EOL> def test_context_manager_does_not_suppress_exception ( self ) : <EOL> class TestException ( Exception ) : <EOL> pass <EOL> with self . assertRaises ( TestException ) : <EOL> with self . obj as channel : <EOL> self . assertFalse ( channel . _impl . close . called ) <EOL> raise TestException ( ) <EOL> channel . _impl . close . assert_called_once_with ( reply_code = <NUM_LIT:0> , reply_text = '<STR_LIT>' ) <EOL> def test_context_manager_exit_with_closed_channel ( self ) : <EOL> with self . obj as channel : <EOL> self . assertFalse ( channel . _impl . close . called ) <EOL> channel . close ( ) <EOL> channel . _impl . close . assert_called_with ( reply_code = <NUM_LIT:0> , reply_text = '<STR_LIT>' ) </s>
