<s> import socket <EOL> import uuid <EOL> import eventlet <EOL> import pytest <EOL> from eventlet . event import Event <EOL> from kombu import Connection , Exchange , Queue <EOL> from kombu . exceptions import TimeoutError <EOL> from mock import ANY , call , Mock , patch <EOL> from nameko . constants import AMQP_URI_CONFIG_KEY <EOL> from nameko . messaging import QueueConsumer <EOL> from nameko . rpc import rpc , RpcConsumer <EOL> from nameko . standalone . rpc import ServiceRpcProxy <EOL> from nameko . testing . utils import ( <EOL> assert_stops_raising , get_extension , get_rabbit_connections ) <EOL> TIMEOUT = <NUM_LIT:5> <EOL> exchange = Exchange ( '<STR_LIT>' ) <EOL> ham_queue = Queue ( '<STR_LIT>' , exchange = exchange , auto_delete = False ) <EOL> @ pytest . yield_fixture <EOL> def logger ( ) : <EOL> with patch ( '<STR_LIT>' ) as logger : <EOL> yield logger <EOL> class MessageHandler ( object ) : <EOL> queue = ham_queue <EOL> def __init__ ( self ) : <EOL> self . handle_message_called = Event ( ) <EOL> def handle_message ( self , body , message ) : <EOL> self . handle_message_called . send ( message ) <EOL> def wait ( self ) : <EOL> return self . handle_message_called . wait ( ) <EOL> def spawn_thread ( method , protected ) : <EOL> return eventlet . spawn ( method ) <EOL> def test_lifecycle ( rabbit_manager , rabbit_config , mock_container ) : <EOL> container = mock_container <EOL> container . shared_extensions = { } <EOL> container . config = rabbit_config <EOL> container . max_workers = <NUM_LIT:3> <EOL> container . spawn_managed_thread . side_effect = spawn_thread <EOL> content_type = '<STR_LIT>' <EOL> container . accept = [ content_type ] <EOL> queue_consumer = QueueConsumer ( ) . bind ( container ) <EOL> handler = MessageHandler ( ) <EOL> queue_consumer . register_provider ( handler ) <EOL> queue_consumer . setup ( ) <EOL> queue_consumer . start ( ) <EOL> container . spawn_managed_thread . assert_called_once_with ( ANY , protected = True ) <EOL> vhost = rabbit_config [ '<STR_LIT>' ] <EOL> rabbit_manager . publish ( vhost , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> properties = dict ( content_type = content_type ) ) <EOL> message = handler . wait ( ) <EOL> gt = eventlet . spawn ( queue_consumer . unregister_provider , handler ) <EOL> with eventlet . Timeout ( TIMEOUT ) : <EOL> while len ( queue_consumer . _consumers ) : <EOL> eventlet . sleep ( ) <EOL> assert not gt . dead <EOL> rabbit_manager . publish ( vhost , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> queue_consumer . ack_message ( message ) <EOL> with eventlet . Timeout ( TIMEOUT ) : <EOL> gt . wait ( ) <EOL> messages = rabbit_manager . get_messages ( vhost , '<STR_LIT>' ) <EOL> assert [ '<STR_LIT>' ] == [ msg [ '<STR_LIT>' ] for msg in messages ] <EOL> queue_consumer . kill ( ) <EOL> def test_reentrant_start_stops ( mock_container ) : <EOL> container = mock_container <EOL> container . shared_extensions = { } <EOL> container . config = { AMQP_URI_CONFIG_KEY : '<STR_LIT>' } <EOL> container . max_workers = <NUM_LIT:3> <EOL> container . spawn_managed_thread = spawn_thread <EOL> queue_consumer = QueueConsumer ( ) . bind ( container ) <EOL> queue_consumer . setup ( ) <EOL> queue_consumer . start ( ) <EOL> gt = queue_consumer . _gt <EOL> queue_consumer . start ( ) <EOL> assert gt is queue_consumer . _gt <EOL> queue_consumer . kill ( ) <EOL> def test_stop_while_starting ( rabbit_config , mock_container ) : <EOL> started = Event ( ) <EOL> container = mock_container <EOL> container . shared_extensions = { } <EOL> container . config = rabbit_config <EOL> container . max_workers = <NUM_LIT:3> <EOL> container . spawn_managed_thread = spawn_thread <EOL> class BrokenConnConsumer ( QueueConsumer ) : <EOL> def consume ( self , * args , ** kwargs ) : <EOL> started . send ( None ) <EOL> started . reset ( ) <EOL> return super ( BrokenConnConsumer , self ) . consume ( * args , ** kwargs ) <EOL> queue_consumer = BrokenConnConsumer ( ) . bind ( container ) <EOL> queue_consumer . setup ( ) <EOL> handler = MessageHandler ( ) <EOL> queue_consumer . register_provider ( handler ) <EOL> with eventlet . Timeout ( TIMEOUT ) : <EOL> with patch . object ( Connection , '<STR_LIT>' , autospec = True ) as connect : <EOL> connect . side_effect = TimeoutError ( '<STR_LIT:test>' ) <EOL> gt = eventlet . spawn ( queue_consumer . start ) <EOL> started . wait ( ) <EOL> with eventlet . Timeout ( TIMEOUT ) : <EOL> queue_consumer . unregister_provider ( handler ) <EOL> queue_consumer . stop ( ) <EOL> with eventlet . Timeout ( TIMEOUT ) : <EOL> while not gt . dead : <EOL> eventlet . sleep ( ) <EOL> assert queue_consumer . _gt . dead <EOL> def test_error_stops_consumer_thread ( mock_container ) : <EOL> container = mock_container <EOL> container . shared_extensions = { } <EOL> container . config = { AMQP_URI_CONFIG_KEY : '<STR_LIT>' } <EOL> container . max_workers = <NUM_LIT:3> <EOL> container . spawn_managed_thread = spawn_thread <EOL> queue_consumer = QueueConsumer ( ) . bind ( container ) <EOL> queue_consumer . setup ( ) <EOL> handler = MessageHandler ( ) <EOL> queue_consumer . register_provider ( handler ) <EOL> with eventlet . Timeout ( TIMEOUT ) : <EOL> with patch . object ( <EOL> Connection , '<STR_LIT>' , autospec = True ) as drain_events : <EOL> drain_events . side_effect = Exception ( '<STR_LIT:test>' ) <EOL> queue_consumer . start ( ) <EOL> with pytest . raises ( Exception ) as exc_info : <EOL> queue_consumer . _gt . wait ( ) <EOL> assert exc_info . value . args == ( '<STR_LIT:test>' , ) <EOL> def test_on_consume_error_kills_consumer ( mock_container ) : <EOL> container = mock_container <EOL> container . shared_extensions = { } <EOL> container . config = { AMQP_URI_CONFIG_KEY : '<STR_LIT>' } <EOL> container . max_workers = <NUM_LIT:1> <EOL> container . spawn_managed_thread = spawn_thread <EOL> queue_consumer = QueueConsumer ( ) . bind ( container ) <EOL> queue_consumer . setup ( ) <EOL> handler = MessageHandler ( ) <EOL> queue_consumer . register_provider ( handler ) <EOL> with patch . object ( queue_consumer , '<STR_LIT>' ) as on_consume_ready : <EOL> on_consume_ready . side_effect = Exception ( '<STR_LIT>' ) <EOL> queue_consumer . start ( ) <EOL> with pytest . raises ( Exception ) : <EOL> queue_consumer . _gt . wait ( ) <EOL> def test_reconnect_on_socket_error ( rabbit_config , mock_container ) : <EOL> container = mock_container <EOL> container . shared_extensions = { } <EOL> container . config = rabbit_config <EOL> container . max_workers = <NUM_LIT:1> <EOL> container . spawn_managed_thread = spawn_thread <EOL> connection_revived = Mock ( ) <EOL> queue_consumer = QueueConsumer ( ) . bind ( container ) <EOL> queue_consumer . setup ( ) <EOL> queue_consumer . on_connection_revived = connection_revived <EOL> handler = MessageHandler ( ) <EOL> queue_consumer . register_provider ( handler ) <EOL> queue_consumer . start ( ) <EOL> with patch . object ( <EOL> Connection , '<STR_LIT>' , autospec = True ) as drain_events : <EOL> drain_events . side_effect = socket . error ( '<STR_LIT>' ) <EOL> def check_reconnected ( ) : <EOL> assert connection_revived . call_count > <NUM_LIT:1> <EOL> assert_stops_raising ( check_reconnected ) <EOL> queue_consumer . unregister_provider ( handler ) <EOL> queue_consumer . stop ( ) <EOL> def test_prefetch_count ( rabbit_manager , rabbit_config , mock_container ) : <EOL> container = mock_container <EOL> container . shared_extensions = { } <EOL> container . config = rabbit_config <EOL> container . max_workers = <NUM_LIT:1> <EOL> container . spawn_managed_thread = spawn_thread <EOL> content_type = '<STR_LIT>' <EOL> container . accept = [ content_type ] <EOL> class NonShared ( QueueConsumer ) : <EOL> @ property <EOL> def sharing_key ( self ) : <EOL> return uuid . uuid4 ( ) <EOL> queue_consumer1 = NonShared ( ) . bind ( container ) <EOL> queue_consumer1 . setup ( ) <EOL> queue_consumer2 = NonShared ( ) . bind ( container ) <EOL> queue_consumer2 . setup ( ) <EOL> consumer_continue = Event ( ) <EOL> class Handler1 ( object ) : <EOL> queue = ham_queue <EOL> def handle_message ( self , body , message ) : <EOL> consumer_continue . wait ( ) <EOL> queue_consumer1 . ack_message ( message ) <EOL> messages = [ ] <EOL> class Handler2 ( object ) : <EOL> queue = ham_queue <EOL> def handle_message ( self , body , message ) : <EOL> messages . append ( body ) <EOL> queue_consumer2 . ack_message ( message ) <EOL> handler1 = Handler1 ( ) <EOL> handler2 = Handler2 ( ) <EOL> queue_consumer1 . register_provider ( handler1 ) <EOL> queue_consumer2 . register_provider ( handler2 ) <EOL> queue_consumer1 . start ( ) <EOL> queue_consumer2 . start ( ) <EOL> vhost = rabbit_config [ '<STR_LIT>' ] <EOL> rabbit_manager . publish ( vhost , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> properties = dict ( content_type = content_type ) ) <EOL> rabbit_manager . publish ( vhost , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> properties = dict ( content_type = content_type ) ) <EOL> rabbit_manager . publish ( vhost , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> properties = dict ( content_type = content_type ) ) <EOL> with eventlet . Timeout ( TIMEOUT ) : <EOL> while len ( messages ) < <NUM_LIT:2> : <EOL> eventlet . sleep ( ) <EOL> consumer_continue . send ( None ) <EOL> assert messages == [ '<STR_LIT>' , '<STR_LIT>' ] <EOL> queue_consumer1 . unregister_provider ( handler1 ) <EOL> queue_consumer2 . unregister_provider ( handler2 ) <EOL> queue_consumer1 . kill ( ) <EOL> queue_consumer2 . kill ( ) <EOL> def test_kill_closes_connections ( rabbit_manager , rabbit_config , <EOL> mock_container ) : <EOL> container = mock_container <EOL> container . shared_extensions = { } <EOL> container . config = rabbit_config <EOL> container . max_workers = <NUM_LIT:1> <EOL> container . spawn_managed_thread = spawn_thread <EOL> queue_consumer = QueueConsumer ( ) . bind ( container ) <EOL> queue_consumer . setup ( ) <EOL> class Handler ( object ) : <EOL> queue = ham_queue <EOL> def handle_message ( self , body , message ) : <EOL> pass <EOL> queue_consumer . register_provider ( Handler ( ) ) <EOL> queue_consumer . start ( ) <EOL> queue_consumer . kill ( ) <EOL> vhost = rabbit_config [ '<STR_LIT>' ] <EOL> connections = get_rabbit_connections ( vhost , rabbit_manager ) <EOL> if connections : <EOL> for connection in connections : <EOL> assert connection [ '<STR_LIT>' ] != vhost <EOL> def test_greenthread_raise_in_kill ( container_factory , rabbit_config , logger ) : <EOL> class Service ( object ) : <EOL> name = "<STR_LIT>" <EOL> @ rpc <EOL> def echo ( self , arg ) : <EOL> return arg <EOL> container = container_factory ( Service , rabbit_config ) <EOL> queue_consumer = get_extension ( container , QueueConsumer ) <EOL> rpc_consumer = get_extension ( container , RpcConsumer ) <EOL> exc = Exception ( '<STR_LIT>' ) <EOL> with patch . object ( rpc_consumer , '<STR_LIT>' ) as handle_message : <EOL> handle_message . side_effect = exc <EOL> container . start ( ) <EOL> with ServiceRpcProxy ( '<STR_LIT>' , rabbit_config ) as service_rpc : <EOL> eventlet . spawn ( service_rpc . echo , "<STR_LIT:foo>" ) <EOL> with pytest . raises ( Exception ) as exc_info : <EOL> container . wait ( ) <EOL> assert str ( exc_info . value ) == "<STR_LIT>" <EOL> assert logger . warn . call_args_list == [ <EOL> call ( "<STR_LIT>" , queue_consumer , exc ) ] </s>
