#NEWLINE# @ mock . patch . object ( workflows . WorkflowManager , 'list' , mock . MagicMock ( return_value = [ ] ) ) #NEWLINE# @ mock . patch . object ( workflows . WorkflowManager , 'get' , mock . MagicMock ( return_value = WF1 ) ) #NEWLINE# @ mock . patch . object ( workflows . WorkflowManager , 'create' , mock . MagicMock ( return_value = [ WF1 ] ) ) #NEWLINE# @ mock . patch . object ( executions . ExecutionManager , 'create' , mock . MagicMock ( return_value = executions . Execution ( None , WF1_EXEC ) ) ) #NEWLINE# def test_launch_workflow( self ) : #NEWLINE# #INDENT# MistralRunner . entry_point = mock . PropertyMock ( return_value = WF1_YAML_FILE_PATH ) #NEWLINE# liveaction = LiveActionDB ( action = WF1_NAME , parameters = ACTION_PARAMS ) #NEWLINE# ( liveaction , execution ) = action_service . request ( liveaction ) #NEWLINE# liveaction = LiveAction . get_by_id ( str ( liveaction . id ) ) #NEWLINE# self . assertEqual ( liveaction . status , action_constants . LIVEACTION_STATUS_RUNNING ) #NEWLINE# mistral_context = liveaction . context . get ( 'mistral' , None ) #NEWLINE# self . assertIsNotNone ( mistral_context ) #NEWLINE# self . assertEqual ( mistral_context [ 'execution_id' ] , WF1_EXEC . get ( 'id' ) ) #NEWLINE# self . assertEqual ( mistral_context [ 'workflow_name' ] , WF1_EXEC . get ( 'workflow_name' ) ) #NEWLINE# workflow_input = copy . deepcopy ( ACTION_PARAMS ) #NEWLINE# workflow_input . update ( { #NEWLINE# #INDENT# 'count' : '3' , #NEWLINE# #UNINDENT# } ) #NEWLINE# env = { #NEWLINE# #INDENT# 'st2_execution_id' : str ( liveaction . id ) , #NEWLINE# 'st2_liveaction_id' : str ( liveaction . id ) , #NEWLINE# 'st2_action_api_url' : 'http://0.0.0.0:9101/v1' , #NEWLINE# '__actions' : { #NEWLINE# #INDENT# 'st2.action' : { #NEWLINE# #INDENT# 'st2_context' : { #NEWLINE# #INDENT# 'endpoint' : 'http://0.0.0.0:9101/v1/actionexecutions' , #NEWLINE# 'parent' : { #NEWLINE# #INDENT# 'execution_id' : str ( execution . id ) , #NEWLINE# #UNINDENT# } , #NEWLINE# 'notify' : { #NEWLINE# #INDENT# #NEWLINE# #UNINDENT# } , #NEWLINE# 'skip_notify_tasks' : [ ] , #NEWLINE# #UNINDENT# } , #NEWLINE# #UNINDENT# } , #NEWLINE# #UNINDENT# } , #NEWLINE# #UNINDENT# } #NEWLINE# executions . ExecutionManager . create . assert_called_with ( WF1_NAME , workflow_input = workflow_input , env = env )