<s> import six <EOL> from taskflow . engines . worker_based import endpoint as ep <EOL> from taskflow . engines . worker_based import protocol as pr <EOL> from taskflow . engines . worker_based import server <EOL> from taskflow import task as task_atom <EOL> from taskflow import test <EOL> from taskflow . test import mock <EOL> from taskflow . tests import utils <EOL> from taskflow . types import failure <EOL> class TestServer ( test . MockTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestServer , self ) . setUp ( ) <EOL> self . server_topic = '<STR_LIT>' <EOL> self . server_exchange = '<STR_LIT>' <EOL> self . broker_url = '<STR_LIT>' <EOL> self . task = utils . TaskOneArgOneReturn ( ) <EOL> self . task_uuid = '<STR_LIT>' <EOL> self . task_args = { '<STR_LIT:x>' : <NUM_LIT:1> } <EOL> self . task_action = '<STR_LIT>' <EOL> self . reply_to = '<STR_LIT>' <EOL> self . endpoints = [ ep . Endpoint ( task_cls = utils . TaskOneArgOneReturn ) , <EOL> ep . Endpoint ( task_cls = utils . TaskWithFailure ) , <EOL> ep . Endpoint ( task_cls = utils . ProgressingTask ) ] <EOL> self . proxy_mock , self . proxy_inst_mock = self . patchClass ( <EOL> server . proxy , '<STR_LIT>' ) <EOL> self . response_mock , self . response_inst_mock = self . patchClass ( <EOL> server . pr , '<STR_LIT>' ) <EOL> self . proxy_inst_mock . is_running = True <EOL> self . executor_mock = mock . MagicMock ( name = '<STR_LIT>' ) <EOL> self . message_mock = mock . MagicMock ( name = '<STR_LIT:message>' ) <EOL> self . message_mock . properties = { '<STR_LIT>' : self . task_uuid , <EOL> '<STR_LIT>' : self . reply_to , <EOL> '<STR_LIT:type>' : pr . REQUEST } <EOL> self . master_mock . attach_mock ( self . executor_mock , '<STR_LIT>' ) <EOL> self . master_mock . attach_mock ( self . message_mock , '<STR_LIT:message>' ) <EOL> def server ( self , reset_master_mock = False , ** kwargs ) : <EOL> server_kwargs = dict ( topic = self . server_topic , <EOL> exchange = self . server_exchange , <EOL> executor = self . executor_mock , <EOL> endpoints = self . endpoints , <EOL> url = self . broker_url ) <EOL> server_kwargs . update ( kwargs ) <EOL> s = server . Server ( ** server_kwargs ) <EOL> if reset_master_mock : <EOL> self . resetMasterMock ( ) <EOL> return s <EOL> def make_request ( self , ** kwargs ) : <EOL> request_kwargs = dict ( task = self . task , <EOL> uuid = self . task_uuid , <EOL> action = self . task_action , <EOL> arguments = self . task_args , <EOL> timeout = <NUM_LIT> ) <EOL> request_kwargs . update ( kwargs ) <EOL> request = pr . Request ( ** request_kwargs ) <EOL> return request . to_dict ( ) <EOL> def test_creation ( self ) : <EOL> s = self . server ( ) <EOL> master_mock_calls = [ <EOL> mock . call . Proxy ( self . server_topic , self . server_exchange , <EOL> type_handlers = mock . ANY , url = self . broker_url , <EOL> transport = mock . ANY , transport_options = mock . ANY , <EOL> retry_options = mock . ANY ) <EOL> ] <EOL> self . master_mock . assert_has_calls ( master_mock_calls ) <EOL> self . assertEqual ( <NUM_LIT:3> , len ( s . _endpoints ) ) <EOL> def test_creation_with_endpoints ( self ) : <EOL> s = self . server ( endpoints = self . endpoints ) <EOL> master_mock_calls = [ <EOL> mock . call . Proxy ( self . server_topic , self . server_exchange , <EOL> type_handlers = mock . ANY , url = self . broker_url , <EOL> transport = mock . ANY , transport_options = mock . ANY , <EOL> retry_options = mock . ANY ) <EOL> ] <EOL> self . master_mock . assert_has_calls ( master_mock_calls ) <EOL> self . assertEqual ( len ( self . endpoints ) , len ( s . _endpoints ) ) <EOL> def test_parse_request ( self ) : <EOL> request = self . make_request ( ) <EOL> bundle = pr . Request . from_dict ( request ) <EOL> task_cls , task_name , action , task_args = bundle <EOL> self . assertEqual ( ( self . task . name , self . task . name , self . task_action , <EOL> dict ( arguments = self . task_args ) ) , <EOL> ( task_cls , task_name , action , task_args ) ) <EOL> def test_parse_request_with_success_result ( self ) : <EOL> request = self . make_request ( action = '<STR_LIT>' , result = <NUM_LIT:1> ) <EOL> bundle = pr . Request . from_dict ( request ) <EOL> task_cls , task_name , action , task_args = bundle <EOL> self . assertEqual ( ( self . task . name , self . task . name , '<STR_LIT>' , <EOL> dict ( arguments = self . task_args , <EOL> result = <NUM_LIT:1> ) ) , <EOL> ( task_cls , task_name , action , task_args ) ) <EOL> def test_parse_request_with_failure_result ( self ) : <EOL> a_failure = failure . Failure . from_exception ( Exception ( '<STR_LIT:test>' ) ) <EOL> request = self . make_request ( action = '<STR_LIT>' , result = a_failure ) <EOL> bundle = pr . Request . from_dict ( request ) <EOL> task_cls , task_name , action , task_args = bundle <EOL> self . assertEqual ( ( self . task . name , self . task . name , '<STR_LIT>' , <EOL> dict ( arguments = self . task_args , <EOL> result = utils . FailureMatcher ( a_failure ) ) ) , <EOL> ( task_cls , task_name , action , task_args ) ) <EOL> def test_parse_request_with_failures ( self ) : <EOL> failures = { '<STR_LIT:0>' : failure . Failure . from_exception ( Exception ( '<STR_LIT>' ) ) , <EOL> '<STR_LIT:1>' : failure . Failure . from_exception ( Exception ( '<STR_LIT>' ) ) } <EOL> request = self . make_request ( action = '<STR_LIT>' , failures = failures ) <EOL> bundle = pr . Request . from_dict ( request ) <EOL> task_cls , task_name , action , task_args = bundle <EOL> self . assertEqual ( <EOL> ( self . task . name , self . task . name , '<STR_LIT>' , <EOL> dict ( arguments = self . task_args , <EOL> failures = dict ( ( i , utils . FailureMatcher ( f ) ) <EOL> for i , f in six . iteritems ( failures ) ) ) ) , <EOL> ( task_cls , task_name , action , task_args ) ) <EOL> @ mock . patch ( "<STR_LIT>" ) <EOL> def test_reply_publish_failure ( self , mocked_exception ) : <EOL> self . proxy_inst_mock . publish . side_effect = RuntimeError ( '<STR_LIT>' ) <EOL> s = self . server ( reset_master_mock = True ) <EOL> s . _reply ( True , self . reply_to , self . task_uuid ) <EOL> self . master_mock . assert_has_calls ( [ <EOL> mock . call . Response ( pr . FAILURE ) , <EOL> mock . call . proxy . publish ( self . response_inst_mock , self . reply_to , <EOL> correlation_id = self . task_uuid ) <EOL> ] ) <EOL> self . assertTrue ( mocked_exception . called ) <EOL> def test_on_run_reply_failure ( self ) : <EOL> request = self . make_request ( task = utils . ProgressingTask ( ) , arguments = { } ) <EOL> self . proxy_inst_mock . publish . side_effect = RuntimeError ( '<STR_LIT>' ) <EOL> s = self . server ( reset_master_mock = True ) <EOL> s . _process_request ( request , self . message_mock ) <EOL> self . assertEqual ( <NUM_LIT:1> , self . proxy_inst_mock . publish . call_count ) <EOL> def test_on_update_progress ( self ) : <EOL> request = self . make_request ( task = utils . ProgressingTask ( ) , arguments = { } ) <EOL> s = self . server ( reset_master_mock = True ) <EOL> s . _process_request ( request , self . message_mock ) <EOL> master_mock_calls = [ <EOL> mock . call . Response ( pr . RUNNING ) , <EOL> mock . call . proxy . publish ( self . response_inst_mock , self . reply_to , <EOL> correlation_id = self . task_uuid ) , <EOL> mock . call . Response ( pr . EVENT , details = { '<STR_LIT>' : <NUM_LIT:0.0> } , <EOL> event_type = task_atom . EVENT_UPDATE_PROGRESS ) , <EOL> mock . call . proxy . publish ( self . response_inst_mock , self . reply_to , <EOL> correlation_id = self . task_uuid ) , <EOL> mock . call . Response ( pr . EVENT , details = { '<STR_LIT>' : <NUM_LIT:1.0> } , <EOL> event_type = task_atom . EVENT_UPDATE_PROGRESS ) , <EOL> mock . call . proxy . publish ( self . response_inst_mock , self . reply_to , <EOL> correlation_id = self . task_uuid ) , <EOL> mock . call . Response ( pr . SUCCESS , result = <NUM_LIT:5> ) , <EOL> mock . call . proxy . publish ( self . response_inst_mock , self . reply_to , <EOL> correlation_id = self . task_uuid ) <EOL> ] <EOL> self . master_mock . assert_has_calls ( master_mock_calls ) <EOL> def test_process_request ( self ) : <EOL> s = self . server ( reset_master_mock = True ) <EOL> s . _process_request ( self . make_request ( ) , self . message_mock ) <EOL> master_mock_calls = [ <EOL> mock . call . Response ( pr . RUNNING ) , <EOL> mock . call . proxy . publish ( self . response_inst_mock , self . reply_to , <EOL> correlation_id = self . task_uuid ) , <EOL> mock . call . Response ( pr . SUCCESS , result = <NUM_LIT:1> ) , <EOL> mock . call . proxy . publish ( self . response_inst_mock , self . reply_to , <EOL> correlation_id = self . task_uuid ) <EOL> ] <EOL> self . master_mock . assert_has_calls ( master_mock_calls ) <EOL> @ mock . patch ( "<STR_LIT>" ) <EOL> def test_process_request_parse_message_failure ( self , mocked_exception ) : <EOL> self . message_mock . properties = { } <EOL> request = self . make_request ( ) <EOL> s = self . server ( reset_master_mock = True ) <EOL> s . _process_request ( request , self . message_mock ) <EOL> self . assertTrue ( mocked_exception . called ) <EOL> @ mock . patch . object ( failure . Failure , '<STR_LIT>' ) <EOL> @ mock . patch . object ( failure . Failure , '<STR_LIT>' ) <EOL> def test_process_request_parse_request_failure ( self , to_mock , from_mock ) : <EOL> failure_dict = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> a_failure = failure . Failure . from_exception ( RuntimeError ( '<STR_LIT>' ) ) <EOL> to_mock . return_value = failure_dict <EOL> from_mock . side_effect = ValueError ( '<STR_LIT>' ) <EOL> request = self . make_request ( result = a_failure ) <EOL> s = self . server ( reset_master_mock = True ) <EOL> s . _process_request ( request , self . message_mock ) <EOL> master_mock_calls = [ <EOL> mock . call . Response ( pr . FAILURE , result = failure_dict ) , <EOL> mock . call . proxy . publish ( self . response_inst_mock , <EOL> self . reply_to , <EOL> correlation_id = self . task_uuid ) <EOL> ] <EOL> self . master_mock . assert_has_calls ( master_mock_calls ) <EOL> @ mock . patch . object ( failure . Failure , '<STR_LIT>' ) <EOL> def test_process_request_endpoint_not_found ( self , to_mock ) : <EOL> failure_dict = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> to_mock . return_value = failure_dict <EOL> request = self . make_request ( task = mock . MagicMock ( name = '<STR_LIT>' ) ) <EOL> s = self . server ( reset_master_mock = True ) <EOL> s . _process_request ( request , self . message_mock ) <EOL> master_mock_calls = [ <EOL> mock . call . Response ( pr . FAILURE , result = failure_dict ) , <EOL> mock . call . proxy . publish ( self . response_inst_mock , <EOL> self . reply_to , <EOL> correlation_id = self . task_uuid ) <EOL> ] <EOL> self . master_mock . assert_has_calls ( master_mock_calls ) <EOL> @ mock . patch . object ( failure . Failure , '<STR_LIT>' ) <EOL> def test_process_request_execution_failure ( self , to_mock ) : <EOL> failure_dict = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> to_mock . return_value = failure_dict <EOL> request = self . make_request ( ) <EOL> request [ '<STR_LIT:action>' ] = '<STR_LIT>' <EOL> s = self . server ( reset_master_mock = True ) <EOL> s . _process_request ( request , self . message_mock ) <EOL> master_mock_calls = [ <EOL> mock . call . Response ( pr . FAILURE , result = failure_dict ) , <EOL> mock . call . proxy . publish ( self . response_inst_mock , <EOL> self . reply_to , <EOL> correlation_id = self . task_uuid ) <EOL> ] <EOL> self . master_mock . assert_has_calls ( master_mock_calls ) <EOL> @ mock . patch . object ( failure . Failure , '<STR_LIT>' ) <EOL> def test_process_request_task_failure ( self , to_mock ) : <EOL> failure_dict = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> to_mock . return_value = failure_dict <EOL> request = self . make_request ( task = utils . TaskWithFailure ( ) , arguments = { } ) <EOL> s = self . server ( reset_master_mock = True ) <EOL> s . _process_request ( request , self . message_mock ) <EOL> master_mock_calls = [ <EOL> mock . call . Response ( pr . RUNNING ) , <EOL> mock . call . proxy . publish ( self . response_inst_mock , self . reply_to , <EOL> correlation_id = self . task_uuid ) , <EOL> mock . call . Response ( pr . FAILURE , result = failure_dict ) , <EOL> mock . call . proxy . publish ( self . response_inst_mock , <EOL> self . reply_to , <EOL> correlation_id = self . task_uuid ) <EOL> ] <EOL> self . master_mock . assert_has_calls ( master_mock_calls ) <EOL> def test_start ( self ) : <EOL> self . server ( reset_master_mock = True ) . start ( ) <EOL> master_mock_calls = [ <EOL> mock . call . proxy . start ( ) <EOL> ] <EOL> self . master_mock . assert_has_calls ( master_mock_calls ) <EOL> def test_wait ( self ) : <EOL> server = self . server ( reset_master_mock = True ) <EOL> server . start ( ) <EOL> server . wait ( ) <EOL> master_mock_calls = [ <EOL> mock . call . proxy . start ( ) , <EOL> mock . call . proxy . wait ( ) <EOL> ] <EOL> self . master_mock . assert_has_calls ( master_mock_calls ) <EOL> def test_stop ( self ) : <EOL> self . server ( reset_master_mock = True ) . stop ( ) <EOL> master_mock_calls = [ <EOL> mock . call . proxy . stop ( ) <EOL> ] <EOL> self . master_mock . assert_has_calls ( master_mock_calls ) </s>
