<s> import json <EOL> import uuid <EOL> from kombu . serialization import register <EOL> from mock import Mock , call <EOL> import pytest <EOL> from nameko . constants import SERIALIZER_CONFIG_KEY <EOL> from nameko . events import event_handler , EventDispatcher <EOL> from nameko . exceptions import RemoteError <EOL> from nameko . rpc import rpc <EOL> from nameko . standalone . rpc import ServiceRpcProxy <EOL> from nameko . testing . services import entrypoint_hook , entrypoint_waiter <EOL> entrypoint_called = Mock ( ) <EOL> test_data = { <EOL> "<STR_LIT:hello>" : ( "<STR_LIT>" , ) , <EOL> <NUM_LIT> : <NUM_LIT> , <EOL> '<STR_LIT:abc>' : [ <NUM_LIT:7> , <NUM_LIT:8> , <NUM_LIT:9> ] , <EOL> '<STR_LIT>' : <NUM_LIT> , <EOL> } <EOL> serialized_info = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT:application/json>' , <EOL> '<STR_LIT>' : '<STR_LIT:utf-8>' , <EOL> '<STR_LIT:data>' : json . loads ( json . dumps ( test_data ) ) <EOL> } , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT:data>' : test_data <EOL> } <EOL> } <EOL> def unserializable ( ) : <EOL> pass <EOL> class Service ( object ) : <EOL> name = "<STR_LIT>" <EOL> @ rpc <EOL> def echo ( self , arg ) : <EOL> entrypoint_called ( arg ) <EOL> return arg <EOL> @ rpc <EOL> def broken ( self ) : <EOL> return unserializable <EOL> @ event_handler ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def event ( self , evt_data ) : <EOL> entrypoint_called ( evt_data ) <EOL> @ pytest . fixture <EOL> def sniffer_queue_factory ( rabbit_config , rabbit_manager ) : <EOL> """<STR_LIT>""" <EOL> vhost = rabbit_config [ '<STR_LIT>' ] <EOL> def make ( exchange , routing_key = "<STR_LIT:*>" ) : <EOL> """<STR_LIT>""" <EOL> queue_name = "<STR_LIT>" . format ( uuid . uuid4 ( ) ) <EOL> rabbit_manager . create_queue ( vhost , queue_name , auto_delete = True ) <EOL> rabbit_manager . create_queue_binding ( <EOL> vhost , exchange , queue_name , routing_key = routing_key ) <EOL> def get_messages ( ) : <EOL> """<STR_LIT>""" <EOL> return rabbit_manager . get_messages ( vhost , queue_name ) <EOL> return get_messages <EOL> return make <EOL> @ pytest . mark . parametrize ( "<STR_LIT>" , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def test_rpc_serialization ( container_factory , rabbit_config , <EOL> sniffer_queue_factory , serializer ) : <EOL> config = rabbit_config <EOL> config [ SERIALIZER_CONFIG_KEY ] = serializer <EOL> container = container_factory ( Service , config ) <EOL> container . start ( ) <EOL> get_messages = sniffer_queue_factory ( '<STR_LIT>' ) <EOL> serialized = serialized_info [ serializer ] <EOL> with ServiceRpcProxy ( '<STR_LIT>' , rabbit_config ) as proxy : <EOL> assert proxy . echo ( test_data ) == serialized [ '<STR_LIT:data>' ] <EOL> assert entrypoint_called . call_args == call ( serialized [ '<STR_LIT:data>' ] ) <EOL> msg = get_messages ( ) [ <NUM_LIT:0> ] <EOL> assert msg [ '<STR_LIT>' ] [ '<STR_LIT>' ] == serialized [ '<STR_LIT>' ] <EOL> def test_rpc_result_serialization_error ( container_factory , rabbit_config ) : <EOL> container = container_factory ( Service , rabbit_config ) <EOL> container . start ( ) <EOL> with ServiceRpcProxy ( '<STR_LIT>' , rabbit_config ) as proxy : <EOL> with pytest . raises ( RemoteError ) as exc : <EOL> proxy . broken ( ) <EOL> assert exc . value . exc_type == "<STR_LIT>" <EOL> assert proxy . echo ( '<STR_LIT:foo>' ) == "<STR_LIT:foo>" <EOL> def test_rpc_arg_serialization_error ( container_factory , rabbit_config ) : <EOL> container = container_factory ( Service , rabbit_config ) <EOL> container . start ( ) <EOL> with ServiceRpcProxy ( '<STR_LIT>' , rabbit_config ) as proxy : <EOL> with pytest . raises ( Exception ) : <EOL> proxy . echo ( unserializable ) <EOL> assert proxy . echo ( '<STR_LIT:foo>' ) == "<STR_LIT:foo>" <EOL> @ pytest . mark . parametrize ( "<STR_LIT>" , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def test_event_serialization ( <EOL> container_factory , rabbit_config , sniffer_queue_factory , serializer <EOL> ) : <EOL> handler_called = Mock ( ) <EOL> class Service ( object ) : <EOL> name = "<STR_LIT>" <EOL> dispatch = EventDispatcher ( ) <EOL> @ rpc <EOL> def dispatch_event ( self , payload ) : <EOL> self . dispatch ( "<STR_LIT>" , payload ) <EOL> @ event_handler ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def handle_event ( self , event_data ) : <EOL> handler_called ( event_data ) <EOL> rabbit_config [ SERIALIZER_CONFIG_KEY ] = serializer <EOL> container = container_factory ( Service , rabbit_config ) <EOL> container . start ( ) <EOL> get_messages = sniffer_queue_factory ( <EOL> "<STR_LIT>" , routing_key = "<STR_LIT>" ) <EOL> serialized = serialized_info [ serializer ] <EOL> with entrypoint_waiter ( container , "<STR_LIT>" ) : <EOL> with entrypoint_hook ( container , "<STR_LIT>" ) as dispatch_event : <EOL> dispatch_event ( test_data ) <EOL> assert handler_called . call_args == call ( serialized [ '<STR_LIT:data>' ] ) <EOL> msg = get_messages ( ) [ <NUM_LIT:0> ] <EOL> assert msg [ '<STR_LIT>' ] [ '<STR_LIT>' ] == serialized [ '<STR_LIT>' ] <EOL> def test_custom_serializer ( container_factory , rabbit_config , <EOL> sniffer_queue_factory ) : <EOL> def encode ( value ) : <EOL> value = json . dumps ( value ) <EOL> return value . upper ( ) <EOL> def decode ( value ) : <EOL> value = value . lower ( ) <EOL> return json . loads ( value ) <EOL> register ( "<STR_LIT>" , encode , decode , "<STR_LIT>" , "<STR_LIT:utf-8>" ) <EOL> class Service ( object ) : <EOL> name = "<STR_LIT>" <EOL> @ rpc <EOL> def echo ( self , arg ) : <EOL> return arg <EOL> rabbit_config [ SERIALIZER_CONFIG_KEY ] = "<STR_LIT>" <EOL> container = container_factory ( Service , rabbit_config ) <EOL> container . start ( ) <EOL> get_messages = sniffer_queue_factory ( '<STR_LIT>' ) <EOL> with ServiceRpcProxy ( '<STR_LIT>' , rabbit_config ) as proxy : <EOL> assert proxy . echo ( "<STR_LIT:hello>" ) == "<STR_LIT:hello>" <EOL> msg = get_messages ( ) [ <NUM_LIT:0> ] <EOL> assert '<STR_LIT>' in msg [ '<STR_LIT>' ] <EOL> assert msg [ '<STR_LIT>' ] [ '<STR_LIT>' ] == "<STR_LIT>" </s>
