<s> import asyncio <EOL> from datetime import datetime <EOL> import contexts <EOL> import asynqp <EOL> from asynqp import message <EOL> from asynqp import frames <EOL> from asynqp import spec <EOL> from asynqp import exceptions <EOL> from . base_contexts import OpenChannelContext , QueueContext , ExchangeContext , BoundQueueContext , ConsumerContext <EOL> from . util import testing_exception_handler <EOL> class WhenDeclaringAQueue ( OpenChannelContext ) : <EOL> def when_I_declare_a_queue ( self ) : <EOL> self . async_partial ( self . channel . declare_queue ( '<STR_LIT>' , durable = True , exclusive = True , auto_delete = True , arguments = { '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : <NUM_LIT:1000> } ) ) <EOL> def it_should_send_a_QueueDeclare_method ( self ) : <EOL> expected_method = spec . QueueDeclare ( <NUM_LIT:0> , '<STR_LIT>' , False , True , True , True , False , { '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : <NUM_LIT:1000> } ) <EOL> self . server . should_have_received_method ( self . channel . id , expected_method ) <EOL> class WhenQueueDeclareOKArrives ( OpenChannelContext ) : <EOL> def given_I_declared_a_queue ( self ) : <EOL> self . queue_name = '<STR_LIT>' <EOL> self . task = asyncio . async ( self . channel . declare_queue ( self . queue_name , durable = True , exclusive = True , auto_delete = True , arguments = { '<STR_LIT>' : <NUM_LIT> , '<STR_LIT>' : <NUM_LIT:1000> } ) ) <EOL> self . tick ( ) <EOL> def when_QueueDeclareOK_arrives ( self ) : <EOL> self . server . send_method ( self . channel . id , spec . QueueDeclareOK ( self . queue_name , <NUM_LIT> , <NUM_LIT> ) ) <EOL> self . result = self . task . result ( ) <EOL> def it_should_have_the_correct_queue_name ( self ) : <EOL> assert self . result . name == self . queue_name <EOL> def it_should_be_durable ( self ) : <EOL> assert self . result . durable <EOL> def it_should_be_exclusive ( self ) : <EOL> assert self . result . exclusive <EOL> def it_should_auto_delete ( self ) : <EOL> assert self . result . auto_delete <EOL> class WhenILetTheServerPickTheQueueName ( OpenChannelContext ) : <EOL> def given_I_declared_a_queue ( self ) : <EOL> self . task = asyncio . async ( self . channel . declare_queue ( '<STR_LIT>' , durable = True , exclusive = True , auto_delete = True ) , loop = self . loop ) <EOL> self . tick ( ) <EOL> self . queue_name = '<STR_LIT>' <EOL> def when_QueueDeclareOK_arrives ( self ) : <EOL> self . server . send_method ( self . channel . id , spec . QueueDeclareOK ( self . queue_name , <NUM_LIT> , <NUM_LIT> ) ) <EOL> self . result = self . task . result ( ) <EOL> def it_should_return_a_queue_with_the_correct_name ( self ) : <EOL> assert self . result . name == self . queue_name <EOL> class WhenIUseAnIllegalNameForAQueue ( OpenChannelContext ) : <EOL> @ classmethod <EOL> def examples_of_bad_names ( cls ) : <EOL> yield '<STR_LIT>' <EOL> yield '<STR_LIT>' <EOL> def when_I_declare_the_queue ( self , queue_name ) : <EOL> self . task = asyncio . async ( self . channel . declare_queue ( queue_name , durable = True , exclusive = True , auto_delete = True ) , <EOL> loop = self . loop ) <EOL> self . tick ( ) <EOL> def it_should_throw_ValueError ( self ) : <EOL> assert isinstance ( self . task . exception ( ) , ValueError ) <EOL> class WhenBindingAQueueToAnExchange ( QueueContext , ExchangeContext ) : <EOL> def when_I_bind_the_queue ( self ) : <EOL> self . async_partial ( self . queue . bind ( self . exchange , '<STR_LIT>' , arguments = { '<STR_LIT>' : '<STR_LIT>' } ) ) <EOL> def it_should_send_QueueBind ( self ) : <EOL> expected_method = spec . QueueBind ( <NUM_LIT:0> , self . queue . name , self . exchange . name , '<STR_LIT>' , False , { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . server . should_have_received_method ( self . channel . id , expected_method ) <EOL> class WhenQueueBindOKArrives ( QueueContext , ExchangeContext ) : <EOL> def given_I_sent_QueueBind ( self ) : <EOL> self . task = asyncio . async ( self . queue . bind ( self . exchange , '<STR_LIT>' ) ) <EOL> self . tick ( ) <EOL> def when_QueueBindOK_arrives ( self ) : <EOL> self . server . send_method ( self . channel . id , spec . QueueBindOK ( ) ) <EOL> self . binding = self . task . result ( ) <EOL> def then_the_returned_binding_should_have_the_correct_queue ( self ) : <EOL> assert self . binding . queue is self . queue <EOL> def and_the_returned_binding_should_have_the_correct_exchange ( self ) : <EOL> assert self . binding . exchange is self . exchange <EOL> class WhenUnbindingAQueue ( BoundQueueContext ) : <EOL> def when_I_unbind_the_queue ( self ) : <EOL> self . async_partial ( self . binding . unbind ( arguments = { '<STR_LIT>' : '<STR_LIT>' } ) ) <EOL> def it_should_send_QueueUnbind ( self ) : <EOL> expected_method = spec . QueueUnbind ( <NUM_LIT:0> , self . queue . name , self . exchange . name , '<STR_LIT>' , { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> self . server . should_have_received_method ( self . channel . id , expected_method ) <EOL> class WhenQueueUnbindOKArrives ( BoundQueueContext ) : <EOL> def given_I_unbound_the_queue ( self ) : <EOL> self . task = asyncio . async ( self . binding . unbind ( ) ) <EOL> self . tick ( ) <EOL> def when_QueueUnbindOK_arrives ( self ) : <EOL> self . server . send_method ( self . channel . id , spec . QueueUnbindOK ( ) ) <EOL> def it_should_be_ok ( self ) : <EOL> assert self . task . result ( ) is None <EOL> class WhenIUnbindAQueueTwice ( BoundQueueContext ) : <EOL> def given_an_unbound_queue ( self ) : <EOL> asyncio . async ( self . binding . unbind ( ) ) <EOL> self . tick ( ) <EOL> self . server . send_method ( self . channel . id , spec . QueueUnbindOK ( ) ) <EOL> def when_I_unbind_the_queue_again ( self ) : <EOL> self . task = asyncio . async ( self . binding . unbind ( ) ) <EOL> self . tick ( ) <EOL> def it_should_throw_Deleted ( self ) : <EOL> assert isinstance ( self . task . exception ( ) , asynqp . Deleted ) <EOL> class WhenIAskForAMessage ( QueueContext ) : <EOL> def when_I_get_a_message ( self ) : <EOL> self . async_partial ( self . queue . get ( no_ack = False ) ) <EOL> def it_should_send_BasicGet ( self ) : <EOL> self . server . should_have_received_method ( self . channel . id , spec . BasicGet ( <NUM_LIT:0> , self . queue . name , False ) ) <EOL> class WhenBasicGetEmptyArrives ( QueueContext ) : <EOL> def given_I_asked_for_a_message ( self ) : <EOL> self . task = asyncio . async ( self . queue . get ( no_ack = False ) ) <EOL> self . tick ( ) <EOL> def when_BasicGetEmpty_arrives ( self ) : <EOL> self . server . send_method ( self . channel . id , spec . BasicGetEmpty ( '<STR_LIT>' ) ) <EOL> def it_should_return_None ( self ) : <EOL> assert self . task . result ( ) is None <EOL> class WhenBasicGetOKArrives ( QueueContext ) : <EOL> def given_I_asked_for_a_message ( self ) : <EOL> self . expected_message = asynqp . Message ( '<STR_LIT:body>' , timestamp = datetime ( <NUM_LIT> , <NUM_LIT:5> , <NUM_LIT:5> ) ) <EOL> self . task = asyncio . async ( self . queue . get ( no_ack = False ) ) <EOL> self . tick ( ) <EOL> def when_BasicGetOK_arrives_with_content ( self ) : <EOL> method = spec . BasicGetOK ( <NUM_LIT> , False , '<STR_LIT>' , '<STR_LIT>' , <NUM_LIT:0> ) <EOL> self . server . send_method ( self . channel . id , method ) <EOL> header = message . get_header_payload ( self . expected_message , spec . BasicGet . method_type [ <NUM_LIT:0> ] ) <EOL> self . server . send_frame ( frames . ContentHeaderFrame ( self . channel . id , header ) ) <EOL> body = message . get_frame_payloads ( self . expected_message , <NUM_LIT:100> ) [ <NUM_LIT:0> ] <EOL> self . server . send_frame ( frames . ContentBodyFrame ( self . channel . id , body ) ) <EOL> self . tick ( ) <EOL> def it_should_return_the_expected_message ( self ) : <EOL> assert self . task . result ( ) == self . expected_message <EOL> def it_should_put_the_exchange_name_on_the_msg ( self ) : <EOL> assert self . task . result ( ) . exchange_name == '<STR_LIT>' <EOL> def it_should_put_the_routing_key_on_the_msg ( self ) : <EOL> assert self . task . result ( ) . routing_key == '<STR_LIT>' <EOL> class WhenConnectionClosedOnGet ( QueueContext ) : <EOL> def given_I_asked_for_a_message ( self ) : <EOL> self . task = asyncio . async ( self . queue . get ( no_ack = False ) ) <EOL> self . tick ( ) <EOL> def when_connection_is_closed ( self ) : <EOL> try : <EOL> self . server . protocol . connection_lost ( Exception ( ) ) <EOL> except Exception : <EOL> pass <EOL> self . tick ( ) <EOL> def it_should_raise_exception ( self ) : <EOL> assert self . task . exception ( ) is not None <EOL> class WhenISubscribeToAQueue ( QueueContext ) : <EOL> def when_I_start_a_consumer ( self ) : <EOL> self . async_partial ( self . queue . consume ( lambda msg : None , no_local = False , no_ack = False , exclusive = False , arguments = { '<STR_LIT>' : <NUM_LIT:1> } ) ) <EOL> def it_should_send_BasicConsume ( self ) : <EOL> self . server . should_have_received_method ( self . channel . id , spec . BasicConsume ( <NUM_LIT:0> , self . queue . name , '<STR_LIT>' , False , False , False , False , { '<STR_LIT>' : <NUM_LIT:1> } ) ) <EOL> class WhenConsumeOKArrives ( QueueContext ) : <EOL> def given_I_started_a_consumer ( self ) : <EOL> self . task = asyncio . async ( self . queue . consume ( lambda msg : None , no_local = False , no_ack = False , exclusive = False , arguments = { '<STR_LIT>' : <NUM_LIT:1> } ) ) <EOL> self . tick ( ) <EOL> def when_BasicConsumeOK_arrives ( self ) : <EOL> method = spec . BasicConsumeOK ( '<STR_LIT>' ) <EOL> self . server . send_method ( self . channel . id , method ) <EOL> def it_should_return_a_consumer_with_the_correct_consumer_tag ( self ) : <EOL> assert self . task . result ( ) . tag == '<STR_LIT>' <EOL> class WhenBasicDeliverArrives ( ConsumerContext ) : <EOL> def given_a_message ( self ) : <EOL> self . expected_message = asynqp . Message ( '<STR_LIT:body>' , timestamp = datetime ( <NUM_LIT> , <NUM_LIT:5> , <NUM_LIT:5> ) ) <EOL> def when_BasicDeliver_arrives_with_content ( self ) : <EOL> method = spec . BasicDeliver ( self . consumer . tag , <NUM_LIT> , False , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . server . send_method ( self . channel . id , method ) <EOL> header = message . get_header_payload ( self . expected_message , spec . BasicGet . method_type [ <NUM_LIT:0> ] ) <EOL> self . server . send_frame ( frames . ContentHeaderFrame ( self . channel . id , header ) ) <EOL> body = message . get_frame_payloads ( self . expected_message , <NUM_LIT:100> ) [ <NUM_LIT:0> ] <EOL> self . server . send_frame ( frames . ContentBodyFrame ( self . channel . id , body ) ) <EOL> self . tick ( ) <EOL> def it_should_send_the_message_to_the_callback ( self ) : <EOL> self . callback . assert_called_once_with ( self . expected_message ) <EOL> class WhenAConsumerThrowsAnExceptionAndAnotherMessageArrives ( ConsumerContext ) : <EOL> def given_a_consumer_has_thrown_an_exception ( self ) : <EOL> self . loop . set_exception_handler ( lambda l , c : None ) <EOL> self . expected_message = asynqp . Message ( '<STR_LIT:body>' , timestamp = datetime ( <NUM_LIT> , <NUM_LIT:5> , <NUM_LIT:5> ) ) <EOL> self . consumer . callback . side_effect = Exception <EOL> self . deliver_msg ( ) <EOL> def when_another_message_arrives ( self ) : <EOL> self . deliver_msg ( ) <EOL> def it_should_correctly_call_the_consumer_again ( self ) : <EOL> assert self . callback . call_count == <NUM_LIT:2> <EOL> def deliver_msg ( self ) : <EOL> method = spec . BasicDeliver ( self . consumer . tag , <NUM_LIT> , False , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . server . send_method ( self . channel . id , method ) <EOL> header = message . get_header_payload ( self . expected_message , spec . BasicGet . method_type [ <NUM_LIT:0> ] ) <EOL> self . server . send_frame ( frames . ContentHeaderFrame ( self . channel . id , header ) ) <EOL> body = message . get_frame_payloads ( self . expected_message , <NUM_LIT:100> ) [ <NUM_LIT:0> ] <EOL> self . server . send_frame ( frames . ContentBodyFrame ( self . channel . id , body ) ) <EOL> self . tick ( ) <EOL> def cleanup_the_exception_handler ( self ) : <EOL> self . loop . set_exception_handler ( testing_exception_handler ) <EOL> class WhenICancelAConsumer ( ConsumerContext ) : <EOL> def when_I_cancel_the_consumer ( self ) : <EOL> self . async_partial ( self . consumer . cancel ( ) ) <EOL> def it_should_send_a_BasicCancel_method ( self ) : <EOL> self . server . should_have_received_method ( self . channel . id , spec . BasicCancel ( self . consumer . tag , False ) ) <EOL> class WhenCancelOKArrives ( ConsumerContext ) : <EOL> def given_I_cancelled_a_consumer ( self ) : <EOL> asyncio . async ( self . consumer . cancel ( ) ) <EOL> self . tick ( ) <EOL> def when_BasicCancelOK_arrives ( self ) : <EOL> self . server . send_method ( self . channel . id , spec . BasicCancelOK ( self . consumer . tag ) ) <EOL> def it_should_be_cancelled ( self ) : <EOL> assert self . consumer . cancelled <EOL> class WhenCancelOKArrivesForAConsumerWithAnOnCancelMethod ( QueueContext ) : <EOL> def given_I_started_and_cancelled_a_consumer ( self ) : <EOL> self . consumer = self . ConsumerWithOnCancel ( ) <EOL> task = asyncio . async ( self . queue . consume ( self . consumer , no_local = False , no_ack = False , exclusive = False , arguments = { '<STR_LIT>' : <NUM_LIT:1> } ) ) <EOL> self . tick ( ) <EOL> self . server . send_method ( self . channel . id , spec . BasicConsumeOK ( '<STR_LIT>' ) ) <EOL> self . tick ( ) <EOL> asyncio . async ( task . result ( ) . cancel ( ) ) <EOL> self . tick ( ) <EOL> def when_BasicCancelOK_arrives ( self ) : <EOL> self . server . send_method ( self . channel . id , spec . BasicCancelOK ( '<STR_LIT>' ) ) <EOL> def it_should_call_on_cancel ( self ) : <EOL> assert self . consumer . on_cancel_called <EOL> class ConsumerWithOnCancel : <EOL> def __init__ ( self ) : <EOL> self . on_cancel_called = False <EOL> def __call__ ( self ) : <EOL> pass <EOL> def on_cancel ( self ) : <EOL> self . on_cancel_called = True <EOL> class WhenAConsumerWithAnOnCancelMethodIsKilledDueToAnError ( QueueContext ) : <EOL> def given_I_started_a_consumer ( self ) : <EOL> self . consumer = self . ConsumerWithOnError ( ) <EOL> asyncio . async ( self . queue . consume ( self . consumer , no_local = False , no_ack = False , exclusive = False , arguments = { '<STR_LIT>' : <NUM_LIT:1> } ) ) <EOL> self . tick ( ) <EOL> self . server . send_method ( self . channel . id , spec . BasicConsumeOK ( '<STR_LIT>' ) ) <EOL> self . tick ( ) <EOL> self . exception = Exception ( ) <EOL> def when_the_connection_dies ( self ) : <EOL> contexts . catch ( self . protocol . connection_lost , self . exception ) <EOL> self . tick ( ) <EOL> def it_should_call_on_error ( self ) : <EOL> assert self . consumer . exc . original_exc is self . exception <EOL> class ConsumerWithOnError : <EOL> def __init__ ( self ) : <EOL> self . exc = None <EOL> def __call__ ( self ) : <EOL> pass <EOL> def on_error ( self , exc ) : <EOL> self . exc = exc <EOL> class WhenIPurgeAQueue ( QueueContext ) : <EOL> def because_I_purge_the_queue ( self ) : <EOL> self . async_partial ( self . queue . purge ( ) ) <EOL> def it_should_send_a_QueuePurge_method ( self ) : <EOL> self . server . should_have_received_method ( self . channel . id , spec . QueuePurge ( <NUM_LIT:0> , self . queue . name , False ) ) <EOL> class WhenQueuePurgeOKArrives ( QueueContext ) : <EOL> def given_I_called_queue_purge ( self ) : <EOL> self . task = asyncio . async ( self . queue . purge ( ) ) <EOL> self . tick ( ) <EOL> def when_QueuePurgeOK_arrives ( self ) : <EOL> self . server . send_method ( self . channel . id , spec . QueuePurgeOK ( <NUM_LIT> ) ) <EOL> def it_should_return ( self ) : <EOL> self . task . result ( ) <EOL> class WhenDeletingAQueue ( QueueContext ) : <EOL> def because_I_delete_the_queue ( self ) : <EOL> self . async_partial ( self . queue . delete ( if_unused = False , if_empty = False ) ) <EOL> def it_should_send_a_QueueDelete_method ( self ) : <EOL> self . server . should_have_received_method ( self . channel . id , spec . QueueDelete ( <NUM_LIT:0> , self . queue . name , False , False , False ) ) <EOL> class WhenQueueDeleteOKArrives ( QueueContext ) : <EOL> def given_I_deleted_a_queue ( self ) : <EOL> asyncio . async ( self . queue . delete ( if_unused = False , if_empty = False ) , loop = self . loop ) <EOL> self . tick ( ) <EOL> def when_QueueDeleteOK_arrives ( self ) : <EOL> method = spec . QueueDeleteOK ( <NUM_LIT> ) <EOL> self . server . send_method ( self . channel . id , method ) <EOL> def it_should_be_deleted ( self ) : <EOL> assert self . queue . deleted <EOL> class WhenITryToUseADeletedQueue ( QueueContext ) : <EOL> def given_a_deleted_queue ( self ) : <EOL> asyncio . async ( self . queue . delete ( if_unused = False , if_empty = False ) , loop = self . loop ) <EOL> self . tick ( ) <EOL> self . server . send_method ( self . channel . id , spec . QueueDeleteOK ( <NUM_LIT> ) ) <EOL> def when_I_try_to_use_the_queue ( self ) : <EOL> self . task = asyncio . async ( self . queue . get ( ) ) <EOL> self . tick ( ) <EOL> def it_should_throw_Deleted ( self ) : <EOL> assert isinstance ( self . task . exception ( ) , asynqp . Deleted ) <EOL> class WhenAConnectionIsClosedCancelConsuming ( QueueContext , ExchangeContext ) : <EOL> def given_a_consumer ( self ) : <EOL> task = asyncio . async ( self . queue . consume ( <EOL> lambda x : None , no_local = False , no_ack = False , <EOL> exclusive = False , arguments = { '<STR_LIT>' : <NUM_LIT:1> } ) ) <EOL> self . tick ( ) <EOL> self . server . send_method ( self . channel . id , spec . BasicConsumeOK ( '<STR_LIT>' ) ) <EOL> self . tick ( ) <EOL> self . consumer = task . result ( ) <EOL> def when_connection_is_closed ( self ) : <EOL> try : <EOL> self . connection . protocol . connection_lost ( Exception ( ) ) <EOL> except Exception : <EOL> pass <EOL> def it_should_not_hang ( self ) : <EOL> self . loop . run_until_complete ( asyncio . wait_for ( self . consumer . cancel ( ) , <NUM_LIT> ) ) <EOL> class WhenIDeclareQueueWithPassiveAndOKArrives ( OpenChannelContext ) : <EOL> def given_I_declared_a_queue_with_passive ( self ) : <EOL> self . task = asyncio . async ( self . channel . declare_queue ( <EOL> '<STR_LIT>' , durable = True , exclusive = True , auto_delete = False , <EOL> passive = True ) , loop = self . loop ) <EOL> self . tick ( ) <EOL> def when_QueueDeclareOK_arrives ( self ) : <EOL> self . server . send_method ( <EOL> self . channel . id , spec . QueueDeclareOK ( '<STR_LIT>' , <NUM_LIT> , <NUM_LIT> ) ) <EOL> def it_should_return_queue_object ( self ) : <EOL> result = self . task . result ( ) <EOL> assert result <EOL> assert result . name == '<STR_LIT>' <EOL> def it_should_have_sent_passive_in_frame ( self ) : <EOL> self . server . should_have_received_method ( <EOL> self . channel . id , spec . QueueDeclare ( <EOL> <NUM_LIT:0> , '<STR_LIT>' , True , True , True , False , False , { } ) ) <EOL> class WhenIDeclareQueueWithPassiveAndErrorArrives ( OpenChannelContext ) : <EOL> def given_I_declared_a_queue_with_passive ( self ) : <EOL> self . task = asyncio . async ( self . channel . declare_queue ( <EOL> '<STR_LIT>' , durable = True , exclusive = True , auto_delete = True , <EOL> passive = True ) , loop = self . loop ) <EOL> self . tick ( ) <EOL> def when_error_arrives ( self ) : <EOL> self . server . send_method ( <EOL> self . channel . id , spec . ChannelClose ( <NUM_LIT> , '<STR_LIT>' , <NUM_LIT> , <NUM_LIT:50> ) ) <EOL> def it_should_raise_exception ( self ) : <EOL> assert isinstance ( self . task . exception ( ) , exceptions . NotFound ) <EOL> class WhenIDeclareQueueWithNoWait ( OpenChannelContext ) : <EOL> def given_I_declared_a_queue_with_passive ( self ) : <EOL> self . task = asyncio . async ( self . channel . declare_queue ( <EOL> '<STR_LIT>' , durable = True , exclusive = True , auto_delete = False , <EOL> nowait = True ) , loop = self . loop ) <EOL> self . tick ( ) <EOL> def it_should_return_queue_object_without_wait ( self ) : <EOL> result = self . task . result ( ) <EOL> assert result <EOL> assert result . name == '<STR_LIT>' <EOL> def it_should_have_sent_nowait_in_frame ( self ) : <EOL> self . server . should_have_received_method ( <EOL> self . channel . id , spec . QueueDeclare ( <EOL> <NUM_LIT:0> , '<STR_LIT>' , False , True , True , False , True , { } ) ) </s>
