<s> from oslo_utils import uuidutils <EOL> from taskflow . engines . action_engine import executor <EOL> from taskflow . engines . worker_based import protocol as pr <EOL> from taskflow import exceptions as excp <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 TestProtocolValidation ( test . TestCase ) : <EOL> def test_send_notify ( self ) : <EOL> msg = pr . Notify ( ) <EOL> pr . Notify . validate ( msg . to_dict ( ) , False ) <EOL> def test_send_notify_invalid ( self ) : <EOL> msg = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> self . assertRaises ( excp . InvalidFormat , <EOL> pr . Notify . validate , msg , False ) <EOL> def test_reply_notify ( self ) : <EOL> msg = pr . Notify ( topic = "<STR_LIT>" , tasks = [ '<STR_LIT:a>' , '<STR_LIT:b>' , '<STR_LIT:c>' ] ) <EOL> pr . Notify . validate ( msg . to_dict ( ) , True ) <EOL> def test_reply_notify_invalid ( self ) : <EOL> msg = { <EOL> '<STR_LIT>' : { } , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> self . assertRaises ( excp . InvalidFormat , <EOL> pr . Notify . validate , msg , True ) <EOL> def test_request ( self ) : <EOL> request = pr . Request ( utils . DummyTask ( "<STR_LIT>" ) , <EOL> uuidutils . generate_uuid ( ) , <EOL> pr . EXECUTE , { } , <NUM_LIT:1.0> ) <EOL> pr . Request . validate ( request . to_dict ( ) ) <EOL> def test_request_invalid ( self ) : <EOL> msg = { <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : [ ] , <EOL> } <EOL> self . assertRaises ( excp . InvalidFormat , pr . Request . validate , msg ) <EOL> def test_request_invalid_action ( self ) : <EOL> request = pr . Request ( utils . DummyTask ( "<STR_LIT>" ) , <EOL> uuidutils . generate_uuid ( ) , <EOL> pr . EXECUTE , { } , <NUM_LIT:1.0> ) <EOL> request = request . to_dict ( ) <EOL> request [ '<STR_LIT:action>' ] = '<STR_LIT>' <EOL> self . assertRaises ( excp . InvalidFormat , pr . Request . validate , request ) <EOL> def test_response_progress ( self ) : <EOL> msg = pr . Response ( pr . EVENT , details = { '<STR_LIT>' : <NUM_LIT:0.5> } , <EOL> event_type = '<STR_LIT>' ) <EOL> pr . Response . validate ( msg . to_dict ( ) ) <EOL> def test_response_completion ( self ) : <EOL> msg = pr . Response ( pr . SUCCESS , result = <NUM_LIT:1> ) <EOL> pr . Response . validate ( msg . to_dict ( ) ) <EOL> def test_response_mixed_invalid ( self ) : <EOL> msg = pr . Response ( pr . EVENT , <EOL> details = { '<STR_LIT>' : <NUM_LIT:0.5> } , <EOL> event_type = '<STR_LIT>' , result = <NUM_LIT:1> ) <EOL> self . assertRaises ( excp . InvalidFormat , pr . Response . validate , msg ) <EOL> def test_response_bad_state ( self ) : <EOL> msg = pr . Response ( '<STR_LIT>' ) <EOL> self . assertRaises ( excp . InvalidFormat , pr . Response . validate , msg ) <EOL> class TestProtocol ( test . TestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestProtocol , self ) . setUp ( ) <EOL> self . task = utils . DummyTask ( ) <EOL> self . task_uuid = '<STR_LIT>' <EOL> self . task_action = '<STR_LIT>' <EOL> self . task_args = { '<STR_LIT:a>' : '<STR_LIT:a>' } <EOL> self . timeout = <NUM_LIT> <EOL> def 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 = self . timeout ) <EOL> request_kwargs . update ( kwargs ) <EOL> return pr . Request ( ** request_kwargs ) <EOL> def request_to_dict ( self , ** kwargs ) : <EOL> to_dict = dict ( task_cls = self . task . name , <EOL> task_name = self . task . name , <EOL> task_version = self . task . version , <EOL> action = self . task_action , <EOL> arguments = self . task_args ) <EOL> to_dict . update ( kwargs ) <EOL> return to_dict <EOL> def test_request_transitions ( self ) : <EOL> request = self . request ( ) <EOL> self . assertEqual ( pr . WAITING , request . current_state ) <EOL> self . assertIn ( request . current_state , pr . WAITING_STATES ) <EOL> self . assertRaises ( excp . InvalidState , request . transition , pr . SUCCESS ) <EOL> self . assertFalse ( request . transition ( pr . WAITING ) ) <EOL> self . assertTrue ( request . transition ( pr . PENDING ) ) <EOL> self . assertTrue ( request . transition ( pr . RUNNING ) ) <EOL> self . assertTrue ( request . transition ( pr . SUCCESS ) ) <EOL> for s in ( pr . PENDING , pr . WAITING ) : <EOL> self . assertRaises ( excp . InvalidState , request . transition , s ) <EOL> def test_creation ( self ) : <EOL> request = self . request ( ) <EOL> self . assertEqual ( self . task_uuid , request . uuid ) <EOL> self . assertEqual ( self . task , request . task ) <EOL> self . assertFalse ( request . future . done ( ) ) <EOL> def test_to_dict_default ( self ) : <EOL> request = self . request ( ) <EOL> self . assertEqual ( self . request_to_dict ( ) , request . to_dict ( ) ) <EOL> def test_to_dict_with_result ( self ) : <EOL> request = self . request ( result = <NUM_LIT> ) <EOL> self . assertEqual ( self . request_to_dict ( result = ( '<STR_LIT:success>' , <NUM_LIT> ) ) , <EOL> request . to_dict ( ) ) <EOL> def test_to_dict_with_result_none ( self ) : <EOL> request = self . request ( result = None ) <EOL> self . assertEqual ( self . request_to_dict ( result = ( '<STR_LIT:success>' , None ) ) , <EOL> request . to_dict ( ) ) <EOL> def test_to_dict_with_result_failure ( self ) : <EOL> a_failure = failure . Failure . from_exception ( RuntimeError ( '<STR_LIT>' ) ) <EOL> expected = self . request_to_dict ( result = ( '<STR_LIT>' , <EOL> a_failure . to_dict ( ) ) ) <EOL> request = self . request ( result = a_failure ) <EOL> self . assertEqual ( expected , request . to_dict ( ) ) <EOL> def test_to_dict_with_failures ( self ) : <EOL> a_failure = failure . Failure . from_exception ( RuntimeError ( '<STR_LIT>' ) ) <EOL> request = self . request ( failures = { self . task . name : a_failure } ) <EOL> expected = self . request_to_dict ( <EOL> failures = { self . task . name : a_failure . to_dict ( ) } ) <EOL> self . assertEqual ( expected , request . to_dict ( ) ) <EOL> def test_to_dict_with_invalid_json_failures ( self ) : <EOL> exc = RuntimeError ( Exception ( "<STR_LIT>" ) ) <EOL> a_failure = failure . Failure . from_exception ( exc ) <EOL> request = self . request ( failures = { self . task . name : a_failure } ) <EOL> expected = self . request_to_dict ( <EOL> failures = { self . task . name : a_failure . to_dict ( include_args = False ) } ) <EOL> self . assertEqual ( expected , request . to_dict ( ) ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_pending_not_expired ( self , now ) : <EOL> now . return_value = <NUM_LIT:0> <EOL> request = self . request ( ) <EOL> now . return_value = self . timeout - <NUM_LIT:1> <EOL> self . assertFalse ( request . expired ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_pending_expired ( self , now ) : <EOL> now . return_value = <NUM_LIT:0> <EOL> request = self . request ( ) <EOL> now . return_value = self . timeout + <NUM_LIT:1> <EOL> self . assertTrue ( request . expired ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_running_not_expired ( self , now ) : <EOL> now . return_value = <NUM_LIT:0> <EOL> request = self . request ( ) <EOL> request . transition ( pr . PENDING ) <EOL> request . transition ( pr . RUNNING ) <EOL> now . return_value = self . timeout + <NUM_LIT:1> <EOL> self . assertFalse ( request . expired ) <EOL> def test_set_result ( self ) : <EOL> request = self . request ( ) <EOL> request . set_result ( <NUM_LIT> ) <EOL> result = request . future . result ( ) <EOL> self . assertEqual ( ( executor . EXECUTED , <NUM_LIT> ) , result ) </s>
