<s> from tempest_lib import exceptions <EOL> from mistralclient . tests . functional . cli import base <EOL> from mistralclient . tests . functional . cli . v2 import base_v2 <EOL> MISTRAL_URL = "<STR_LIT>" <EOL> class SimpleMistralCLITests ( base . MistralCLIAuth ) : <EOL> """<STR_LIT>""" <EOL> _mistral_url = MISTRAL_URL <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> super ( SimpleMistralCLITests , cls ) . setUpClass ( ) <EOL> def test_workbooks_list ( self ) : <EOL> workbooks = self . parser . listing ( self . mistral ( '<STR_LIT>' ) ) <EOL> self . assertTableStruct ( <EOL> workbooks , <EOL> [ '<STR_LIT:Name>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> ) <EOL> def test_workflow_list ( self ) : <EOL> workflows = self . parser . listing ( <EOL> self . mistral ( '<STR_LIT>' ) ) <EOL> self . assertTableStruct ( <EOL> workflows , <EOL> [ '<STR_LIT>' , '<STR_LIT:Name>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> ) <EOL> def test_executions_list ( self ) : <EOL> executions = self . parser . listing ( <EOL> self . mistral ( '<STR_LIT>' ) ) <EOL> self . assertTableStruct ( <EOL> executions , <EOL> [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> ) <EOL> def test_tasks_list ( self ) : <EOL> tasks = self . parser . listing ( <EOL> self . mistral ( '<STR_LIT>' ) ) <EOL> self . assertTableStruct ( <EOL> tasks , <EOL> [ '<STR_LIT>' , '<STR_LIT:Name>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> ) <EOL> def test_cron_trigger_list ( self ) : <EOL> triggers = self . parser . listing ( <EOL> self . mistral ( '<STR_LIT>' ) ) <EOL> self . assertTableStruct ( <EOL> triggers , <EOL> [ '<STR_LIT:Name>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> ) <EOL> def test_actions_list ( self ) : <EOL> actions = self . parser . listing ( <EOL> self . mistral ( '<STR_LIT>' ) ) <EOL> self . assertTableStruct ( <EOL> actions , <EOL> [ '<STR_LIT:Name>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> ) <EOL> def test_environments_list ( self ) : <EOL> envs = self . parser . listing ( <EOL> self . mistral ( '<STR_LIT>' ) ) <EOL> self . assertTableStruct ( <EOL> envs , <EOL> [ '<STR_LIT:Name>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> ) <EOL> def test_action_execution_list ( self ) : <EOL> act_execs = self . parser . listing ( <EOL> self . mistral ( '<STR_LIT>' ) ) <EOL> self . assertTableStruct ( <EOL> act_execs , <EOL> [ '<STR_LIT>' , '<STR_LIT:Name>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] <EOL> ) <EOL> class WorkbookCLITests ( base_v2 . MistralClientTestBase ) : <EOL> """<STR_LIT>""" <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> super ( WorkbookCLITests , cls ) . setUpClass ( ) <EOL> def test_workbook_create_delete ( self ) : <EOL> wb = self . mistral_admin ( <EOL> '<STR_LIT>' , params = self . wb_def ) <EOL> wb_name = self . get_value_of_field ( wb , "<STR_LIT:Name>" ) <EOL> self . assertTableStruct ( wb , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> wbs = self . mistral_admin ( '<STR_LIT>' ) <EOL> self . assertIn ( wb_name , [ workbook [ '<STR_LIT:Name>' ] for workbook in wbs ] ) <EOL> wbs = self . mistral_admin ( '<STR_LIT>' ) <EOL> self . assertIn ( wb_name , [ workbook [ '<STR_LIT:Name>' ] for workbook in wbs ] ) <EOL> self . mistral_admin ( '<STR_LIT>' , params = wb_name ) <EOL> wbs = self . mistral_admin ( '<STR_LIT>' ) <EOL> self . assertNotIn ( wb_name , [ workbook [ '<STR_LIT:Name>' ] for workbook in wbs ] ) <EOL> def test_workbook_create_with_tags ( self ) : <EOL> wb = self . workbook_create ( self . wb_with_tags_def ) <EOL> tags = self . get_value_of_field ( wb , '<STR_LIT>' ) <EOL> self . assertIn ( '<STR_LIT>' , tags ) <EOL> def test_workbook_update ( self ) : <EOL> wb = self . workbook_create ( self . wb_def ) <EOL> wb_name = self . get_value_of_field ( wb , "<STR_LIT:Name>" ) <EOL> init_update_at = self . get_value_of_field ( wb , "<STR_LIT>" ) <EOL> tags = self . get_value_of_field ( wb , '<STR_LIT>' ) <EOL> self . assertNotIn ( '<STR_LIT>' , tags ) <EOL> wb = self . mistral_admin ( <EOL> '<STR_LIT>' , params = self . wb_def ) <EOL> update_at = self . get_value_of_field ( wb , "<STR_LIT>" ) <EOL> name = self . get_value_of_field ( wb , '<STR_LIT:Name>' ) <EOL> tags = self . get_value_of_field ( wb , '<STR_LIT>' ) <EOL> self . assertEqual ( wb_name , name ) <EOL> self . assertNotIn ( '<STR_LIT>' , tags ) <EOL> self . assertEqual ( init_update_at , update_at ) <EOL> wb = self . mistral_admin ( <EOL> '<STR_LIT>' , params = self . wb_with_tags_def ) <EOL> self . assertTableStruct ( wb , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> update_at = self . get_value_of_field ( wb , "<STR_LIT>" ) <EOL> name = self . get_value_of_field ( wb , '<STR_LIT:Name>' ) <EOL> tags = self . get_value_of_field ( wb , '<STR_LIT>' ) <EOL> self . assertEqual ( wb_name , name ) <EOL> self . assertIn ( '<STR_LIT>' , tags ) <EOL> self . assertNotEqual ( init_update_at , update_at ) <EOL> def test_workbook_get ( self ) : <EOL> created = self . workbook_create ( self . wb_with_tags_def ) <EOL> wb_name = self . get_value_of_field ( created , "<STR_LIT:Name>" ) <EOL> fetched = self . mistral_admin ( '<STR_LIT>' , params = wb_name ) <EOL> created_wb_name = self . get_value_of_field ( created , '<STR_LIT:Name>' ) <EOL> fetched_wb_name = self . get_value_of_field ( fetched , '<STR_LIT:Name>' ) <EOL> self . assertEqual ( created_wb_name , fetched_wb_name ) <EOL> created_wb_tag = self . get_value_of_field ( created , '<STR_LIT>' ) <EOL> fetched_wb_tag = self . get_value_of_field ( fetched , '<STR_LIT>' ) <EOL> self . assertEqual ( created_wb_tag , fetched_wb_tag ) <EOL> def test_workbook_get_definition ( self ) : <EOL> wb = self . workbook_create ( self . wb_def ) <EOL> wb_name = self . get_value_of_field ( wb , "<STR_LIT:Name>" ) <EOL> definition = self . mistral_admin ( <EOL> '<STR_LIT>' , params = wb_name ) <EOL> self . assertNotIn ( '<STR_LIT>' , definition ) <EOL> def test_workbook_validate_with_valid_def ( self ) : <EOL> wb = self . mistral_admin ( <EOL> '<STR_LIT>' , params = self . wb_def ) <EOL> wb_valid = self . get_value_of_field ( wb , '<STR_LIT>' ) <EOL> wb_error = self . get_value_of_field ( wb , '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT:True>' , wb_valid ) <EOL> self . assertEqual ( '<STR_LIT:None>' , wb_error ) <EOL> def test_workbook_validate_with_invalid_def ( self ) : <EOL> self . create_file ( '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> wb = self . mistral_admin ( <EOL> '<STR_LIT>' , params = '<STR_LIT>' ) <EOL> wb_valid = self . get_value_of_field ( wb , '<STR_LIT>' ) <EOL> wb_error = self . get_value_of_field ( wb , '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT:False>' , wb_valid ) <EOL> self . assertNotEqual ( '<STR_LIT:None>' , wb_error ) <EOL> class WorkflowCLITests ( base_v2 . MistralClientTestBase ) : <EOL> """<STR_LIT>""" <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> super ( WorkflowCLITests , cls ) . setUpClass ( ) <EOL> def test_workflow_create_delete ( self ) : <EOL> init_wfs = self . mistral_admin ( <EOL> '<STR_LIT>' , params = self . wf_def ) <EOL> wf_names = [ wf [ '<STR_LIT:Name>' ] for wf in init_wfs ] <EOL> self . assertTableStruct ( init_wfs , [ '<STR_LIT:Name>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> wfs = self . mistral_admin ( '<STR_LIT>' ) <EOL> self . assertIn ( wf_names [ <NUM_LIT:0> ] , [ workflow [ '<STR_LIT:Name>' ] for workflow in wfs ] ) <EOL> for wf_name in wf_names : <EOL> self . mistral_admin ( '<STR_LIT>' , params = wf_name ) <EOL> wfs = self . mistral_admin ( '<STR_LIT>' ) <EOL> for wf in wf_names : <EOL> self . assertNotIn ( wf , [ workflow [ '<STR_LIT:Name>' ] for workflow in wfs ] ) <EOL> def test_create_wf_with_tags ( self ) : <EOL> init_wfs = self . workflow_create ( self . wf_def ) <EOL> wf_name = init_wfs [ <NUM_LIT:1> ] [ '<STR_LIT:Name>' ] <EOL> self . assertTableStruct ( init_wfs , [ '<STR_LIT:Name>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> created_wf_info = self . get_item_info ( <EOL> get_from = init_wfs , get_by = '<STR_LIT:Name>' , value = wf_name ) <EOL> self . assertEqual ( '<STR_LIT>' , created_wf_info [ '<STR_LIT>' ] ) <EOL> def test_workflow_update ( self ) : <EOL> wf = self . workflow_create ( self . wf_def ) <EOL> wf_name = wf [ <NUM_LIT:0> ] [ '<STR_LIT:Name>' ] <EOL> wf_id = wf [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> created_wf_info = self . get_item_info ( <EOL> get_from = wf , <EOL> get_by = '<STR_LIT:Name>' , <EOL> value = wf_name <EOL> ) <EOL> upd_wf = self . mistral_admin ( <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' . format ( self . wf_def ) <EOL> ) <EOL> self . assertTableStruct ( upd_wf , [ '<STR_LIT:Name>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> updated_wf_info = self . get_item_info ( <EOL> get_from = upd_wf , <EOL> get_by = '<STR_LIT:Name>' , <EOL> value = wf_name <EOL> ) <EOL> self . assertEqual ( wf_name , upd_wf [ <NUM_LIT:0> ] [ '<STR_LIT:Name>' ] ) <EOL> self . assertEqual ( <EOL> created_wf_info [ '<STR_LIT>' ] . split ( "<STR_LIT:.>" ) [ <NUM_LIT:0> ] , <EOL> updated_wf_info [ '<STR_LIT>' ] <EOL> ) <EOL> self . assertEqual ( <EOL> created_wf_info [ '<STR_LIT>' ] , <EOL> updated_wf_info [ '<STR_LIT>' ] <EOL> ) <EOL> upd_wf = self . mistral_admin ( <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' . format ( self . wf_with_delay_def ) <EOL> ) <EOL> self . assertTableStruct ( upd_wf , [ '<STR_LIT:Name>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> updated_wf_info = self . get_item_info ( <EOL> get_from = upd_wf , <EOL> get_by = '<STR_LIT:Name>' , <EOL> value = wf_name <EOL> ) <EOL> self . assertEqual ( wf_name , upd_wf [ <NUM_LIT:0> ] [ '<STR_LIT:Name>' ] ) <EOL> self . assertEqual ( <EOL> created_wf_info [ '<STR_LIT>' ] . split ( "<STR_LIT:.>" ) [ <NUM_LIT:0> ] , <EOL> updated_wf_info [ '<STR_LIT>' ] <EOL> ) <EOL> self . assertNotEqual ( <EOL> created_wf_info [ '<STR_LIT>' ] , <EOL> updated_wf_info [ '<STR_LIT>' ] <EOL> ) <EOL> upd_wf = self . mistral_admin ( <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' . format ( self . wf_with_delay_def , wf_id ) <EOL> ) <EOL> self . assertTableStruct ( upd_wf , [ '<STR_LIT:Name>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> updated_wf_info = self . get_item_info ( <EOL> get_from = upd_wf , <EOL> get_by = '<STR_LIT>' , <EOL> value = wf_id <EOL> ) <EOL> self . assertEqual ( wf_name , upd_wf [ <NUM_LIT:0> ] [ '<STR_LIT:Name>' ] ) <EOL> self . assertEqual ( <EOL> created_wf_info [ '<STR_LIT>' ] . split ( "<STR_LIT:.>" ) [ <NUM_LIT:0> ] , <EOL> updated_wf_info [ '<STR_LIT>' ] <EOL> ) <EOL> self . assertNotEqual ( <EOL> created_wf_info [ '<STR_LIT>' ] , <EOL> updated_wf_info [ '<STR_LIT>' ] <EOL> ) <EOL> def test_workflow_update_truncate_input ( self ) : <EOL> input_value = "<STR_LIT>" <EOL> wf_def = """<STR_LIT>""" . format ( input_value ) <EOL> self . create_file ( '<STR_LIT>' , wf_def ) <EOL> self . workflow_create ( '<STR_LIT>' ) <EOL> upd_wf = self . mistral_admin ( <EOL> '<STR_LIT>' , params = '<STR_LIT>' ) <EOL> upd_wf_info = self . get_item_info ( <EOL> get_from = upd_wf , get_by = '<STR_LIT:Name>' , value = '<STR_LIT>' ) <EOL> self . assertEqual ( upd_wf_info [ '<STR_LIT>' ] [ : - <NUM_LIT:3> ] , input_value [ : <NUM_LIT> ] ) <EOL> def test_workflow_get ( self ) : <EOL> created = self . workflow_create ( self . wf_def ) <EOL> wf_name = created [ <NUM_LIT:0> ] [ '<STR_LIT:Name>' ] <EOL> fetched = self . mistral_admin ( '<STR_LIT>' , params = wf_name ) <EOL> fetched_wf_name = self . get_value_of_field ( fetched , '<STR_LIT:Name>' ) <EOL> self . assertEqual ( wf_name , fetched_wf_name ) <EOL> def test_workflow_get_with_id ( self ) : <EOL> created = self . workflow_create ( self . wf_def ) <EOL> wf_name = created [ <NUM_LIT:0> ] [ '<STR_LIT:Name>' ] <EOL> wf_id = created [ <NUM_LIT:0> ] [ '<STR_LIT>' ] <EOL> fetched = self . mistral_admin ( '<STR_LIT>' , params = wf_id ) <EOL> fetched_wf_name = self . get_value_of_field ( fetched , '<STR_LIT:Name>' ) <EOL> self . assertEqual ( wf_name , fetched_wf_name ) <EOL> def test_workflow_get_definition ( self ) : <EOL> wf = self . workflow_create ( self . wf_def ) <EOL> wf_name = wf [ <NUM_LIT:0> ] [ '<STR_LIT:Name>' ] <EOL> definition = self . mistral_admin ( <EOL> '<STR_LIT>' , params = wf_name ) <EOL> self . assertNotIn ( '<STR_LIT>' , definition ) <EOL> def test_workflow_validate_with_valid_def ( self ) : <EOL> wf = self . mistral_admin ( <EOL> '<STR_LIT>' , params = self . wf_def ) <EOL> wf_valid = self . get_value_of_field ( wf , '<STR_LIT>' ) <EOL> wf_error = self . get_value_of_field ( wf , '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT:True>' , wf_valid ) <EOL> self . assertEqual ( '<STR_LIT:None>' , wf_error ) <EOL> def test_workflow_validate_with_invalid_def ( self ) : <EOL> self . create_file ( '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> wf = self . mistral_admin ( <EOL> '<STR_LIT>' , params = '<STR_LIT>' ) <EOL> wf_valid = self . get_value_of_field ( wf , '<STR_LIT>' ) <EOL> wf_error = self . get_value_of_field ( wf , '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT:False>' , wf_valid ) <EOL> self . assertNotEqual ( '<STR_LIT:None>' , wf_error ) <EOL> class ExecutionCLITests ( base_v2 . MistralClientTestBase ) : <EOL> """<STR_LIT>""" <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> super ( ExecutionCLITests , cls ) . setUpClass ( ) <EOL> def setUp ( self ) : <EOL> super ( ExecutionCLITests , self ) . setUp ( ) <EOL> wfs = self . workflow_create ( self . wf_def ) <EOL> self . direct_wf = wfs [ <NUM_LIT:0> ] <EOL> self . reverse_wf = wfs [ <NUM_LIT:1> ] <EOL> self . create_file ( '<STR_LIT:input>' , '<STR_LIT>' ) <EOL> self . create_file ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_execution_create_delete ( self ) : <EOL> execution = self . mistral_admin ( <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' . format ( self . direct_wf [ '<STR_LIT:Name>' ] ) <EOL> ) <EOL> exec_id = self . get_value_of_field ( execution , '<STR_LIT>' ) <EOL> self . assertTableStruct ( execution , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> wf_name = self . get_value_of_field ( execution , '<STR_LIT>' ) <EOL> wf_id = self . get_value_of_field ( execution , '<STR_LIT>' ) <EOL> created_at = self . get_value_of_field ( execution , '<STR_LIT>' ) <EOL> description = self . get_value_of_field ( execution , '<STR_LIT>' ) <EOL> self . assertEqual ( self . direct_wf [ '<STR_LIT:Name>' ] , wf_name ) <EOL> self . assertIsNotNone ( wf_id ) <EOL> self . assertIsNotNone ( created_at ) <EOL> self . assertEqual ( "<STR_LIT>" , description ) <EOL> execs = self . mistral_admin ( '<STR_LIT>' ) <EOL> self . assertIn ( exec_id , [ ex [ '<STR_LIT>' ] for ex in execs ] ) <EOL> self . assertIn ( wf_name , [ ex [ '<STR_LIT>' ] for ex in execs ] ) <EOL> self . mistral_admin ( '<STR_LIT>' , params = exec_id ) <EOL> def test_execution_create_with_input_and_start_task ( self ) : <EOL> execution = self . execution_create ( <EOL> "<STR_LIT>" % self . reverse_wf [ '<STR_LIT:Name>' ] ) <EOL> exec_id = self . get_value_of_field ( execution , '<STR_LIT>' ) <EOL> result = self . wait_execution_success ( exec_id ) <EOL> self . assertTrue ( result ) <EOL> def test_execution_update ( self ) : <EOL> execution = self . execution_create ( self . direct_wf [ '<STR_LIT:Name>' ] ) <EOL> exec_id = self . get_value_of_field ( execution , '<STR_LIT>' ) <EOL> status = self . get_value_of_field ( execution , '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , status ) <EOL> execution = self . mistral_admin ( <EOL> '<STR_LIT>' , params = '<STR_LIT>' . format ( exec_id ) ) <EOL> updated_exec_id = self . get_value_of_field ( execution , '<STR_LIT>' ) <EOL> status = self . get_value_of_field ( execution , '<STR_LIT>' ) <EOL> self . assertEqual ( exec_id , updated_exec_id ) <EOL> self . assertEqual ( '<STR_LIT>' , status ) <EOL> execution = self . mistral_admin ( <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' . format ( exec_id ) <EOL> ) <EOL> description = self . get_value_of_field ( execution , '<STR_LIT>' ) <EOL> self . assertEqual ( "<STR_LIT>" , description ) <EOL> def test_execution_get ( self ) : <EOL> execution = self . execution_create ( self . direct_wf [ '<STR_LIT:Name>' ] ) <EOL> exec_id = self . get_value_of_field ( execution , '<STR_LIT>' ) <EOL> execution = self . mistral_admin ( <EOL> '<STR_LIT>' , params = '<STR_LIT>' . format ( exec_id ) ) <EOL> gotten_id = self . get_value_of_field ( execution , '<STR_LIT>' ) <EOL> wf_name = self . get_value_of_field ( execution , '<STR_LIT>' ) <EOL> wf_id = self . get_value_of_field ( execution , '<STR_LIT>' ) <EOL> self . assertIsNotNone ( wf_id ) <EOL> self . assertEqual ( exec_id , gotten_id ) <EOL> self . assertEqual ( self . direct_wf [ '<STR_LIT:Name>' ] , wf_name ) <EOL> def test_execution_get_input ( self ) : <EOL> execution = self . execution_create ( self . direct_wf [ '<STR_LIT:Name>' ] ) <EOL> exec_id = self . get_value_of_field ( execution , '<STR_LIT>' ) <EOL> ex_input = self . mistral_admin ( '<STR_LIT>' , params = exec_id ) <EOL> self . assertEqual ( [ ] , ex_input ) <EOL> def test_execution_get_output ( self ) : <EOL> execution = self . execution_create ( self . direct_wf [ '<STR_LIT:Name>' ] ) <EOL> exec_id = self . get_value_of_field ( execution , '<STR_LIT>' ) <EOL> ex_output = self . mistral_admin ( <EOL> '<STR_LIT>' , params = exec_id ) <EOL> self . assertEqual ( [ ] , ex_output ) <EOL> def test_executions_list_with_pagination ( self ) : <EOL> execution1 = self . mistral_admin ( <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' . format ( self . direct_wf [ '<STR_LIT:Name>' ] ) <EOL> ) <EOL> execution2 = self . mistral_admin ( <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' . format ( self . direct_wf [ '<STR_LIT:Name>' ] ) <EOL> ) <EOL> executions = self . mistral_cli ( <EOL> True , <EOL> '<STR_LIT>' , <EOL> params = "<STR_LIT>" <EOL> ) <EOL> self . assertEqual ( <NUM_LIT:1> , len ( executions ) ) <EOL> exec_id_1 = self . get_value_of_field ( execution1 , '<STR_LIT>' ) <EOL> exec_id_2 = self . get_value_of_field ( execution2 , '<STR_LIT>' ) <EOL> executions = self . mistral_cli ( <EOL> True , <EOL> '<STR_LIT>' , <EOL> params = "<STR_LIT>" % exec_id_1 <EOL> ) <EOL> self . assertNotIn ( exec_id_1 , [ ex [ '<STR_LIT>' ] for ex in executions ] ) <EOL> self . assertIn ( exec_id_2 , [ ex [ '<STR_LIT>' ] for ex in executions ] ) <EOL> executions = self . mistral_cli ( <EOL> True , <EOL> '<STR_LIT>' , <EOL> params = "<STR_LIT>" <EOL> ) <EOL> self . assertIn ( exec_id_1 , [ ex [ '<STR_LIT>' ] for ex in executions ] ) <EOL> self . assertIn ( exec_id_2 , [ ex [ '<STR_LIT>' ] for ex in executions ] ) <EOL> ex1_index = - <NUM_LIT:1> <EOL> ex2_index = - <NUM_LIT:1> <EOL> for idx , ex in enumerate ( executions ) : <EOL> if ex [ '<STR_LIT>' ] == exec_id_1 : <EOL> ex1_index = idx <EOL> elif ex [ '<STR_LIT>' ] == exec_id_2 : <EOL> ex2_index = idx <EOL> self . assertTrue ( ex1_index < ex2_index ) <EOL> executions = self . mistral_cli ( <EOL> True , <EOL> '<STR_LIT>' , <EOL> params = "<STR_LIT>" <EOL> ) <EOL> self . assertIn ( exec_id_1 , [ ex [ '<STR_LIT>' ] for ex in executions ] ) <EOL> self . assertIn ( exec_id_2 , [ ex [ '<STR_LIT>' ] for ex in executions ] ) <EOL> ex1_index = - <NUM_LIT:1> <EOL> ex2_index = - <NUM_LIT:1> <EOL> for idx , ex in enumerate ( executions ) : <EOL> if ex [ '<STR_LIT>' ] == exec_id_1 : <EOL> ex1_index = idx <EOL> elif ex [ '<STR_LIT>' ] == exec_id_2 : <EOL> ex2_index = idx <EOL> self . assertTrue ( ex1_index > ex2_index ) <EOL> class CronTriggerCLITests ( base_v2 . MistralClientTestBase ) : <EOL> """<STR_LIT>""" <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> super ( CronTriggerCLITests , cls ) . setUpClass ( ) <EOL> def setUp ( self ) : <EOL> super ( CronTriggerCLITests , self ) . setUp ( ) <EOL> wf = self . workflow_create ( self . wf_def ) <EOL> self . wf_name = wf [ <NUM_LIT:0> ] [ '<STR_LIT:Name>' ] <EOL> def test_cron_trigger_create_delete ( self ) : <EOL> trigger = self . mistral_admin ( <EOL> '<STR_LIT>' , <EOL> params = ( '<STR_LIT>' <EOL> '<STR_LIT>' % self . wf_name ) ) <EOL> self . assertTableStruct ( trigger , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> tr_name = self . get_value_of_field ( trigger , '<STR_LIT:Name>' ) <EOL> wf_name = self . get_value_of_field ( trigger , '<STR_LIT>' ) <EOL> created_at = self . get_value_of_field ( trigger , '<STR_LIT>' ) <EOL> remain = self . get_value_of_field ( trigger , '<STR_LIT>' ) <EOL> next_time = self . get_value_of_field ( trigger , '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , tr_name ) <EOL> self . assertEqual ( self . wf_name , wf_name ) <EOL> self . assertIsNotNone ( created_at ) <EOL> self . assertEqual ( "<STR_LIT>" , next_time ) <EOL> self . assertEqual ( "<STR_LIT:5>" , remain ) <EOL> trgs = self . mistral_admin ( '<STR_LIT>' ) <EOL> self . assertIn ( tr_name , [ tr [ '<STR_LIT:Name>' ] for tr in trgs ] ) <EOL> self . assertIn ( wf_name , [ tr [ '<STR_LIT>' ] for tr in trgs ] ) <EOL> self . mistral ( '<STR_LIT>' , params = tr_name ) <EOL> trgs = self . mistral_admin ( '<STR_LIT>' ) <EOL> self . assertNotIn ( tr_name , [ tr [ '<STR_LIT:Name>' ] for tr in trgs ] ) <EOL> def test_two_cron_triggers_for_one_wf ( self ) : <EOL> self . cron_trigger_create ( <EOL> '<STR_LIT>' , self . wf_name , '<STR_LIT:{}>' , "<STR_LIT>" ) <EOL> self . cron_trigger_create ( <EOL> '<STR_LIT>' , self . wf_name , '<STR_LIT:{}>' , "<STR_LIT>" ) <EOL> trgs = self . mistral_admin ( '<STR_LIT>' ) <EOL> self . assertIn ( "<STR_LIT>" , [ tr [ '<STR_LIT:Name>' ] for tr in trgs ] ) <EOL> self . assertIn ( "<STR_LIT>" , [ tr [ '<STR_LIT:Name>' ] for tr in trgs ] ) <EOL> def test_cron_trigger_get ( self ) : <EOL> trigger = self . cron_trigger_create ( <EOL> '<STR_LIT>' , self . wf_name , '<STR_LIT:{}>' , "<STR_LIT>" ) <EOL> self . assertTableStruct ( trigger , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> tr_name = self . get_value_of_field ( trigger , '<STR_LIT:Name>' ) <EOL> fetched_tr = self . mistral_admin ( <EOL> '<STR_LIT>' , params = '<STR_LIT>' ) <EOL> self . assertTableStruct ( trigger , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> tr_name = self . get_value_of_field ( fetched_tr , '<STR_LIT:Name>' ) <EOL> wf_name = self . get_value_of_field ( fetched_tr , '<STR_LIT>' ) <EOL> created_at = self . get_value_of_field ( fetched_tr , '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , tr_name ) <EOL> self . assertEqual ( self . wf_name , wf_name ) <EOL> self . assertIsNotNone ( created_at ) <EOL> class TaskCLITests ( base_v2 . MistralClientTestBase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> super ( TaskCLITests , self ) . setUp ( ) <EOL> wfs = self . workflow_create ( self . wf_def ) <EOL> self . direct_wf = wfs [ <NUM_LIT:0> ] <EOL> self . reverse_wf = wfs [ <NUM_LIT:1> ] <EOL> self . create_file ( '<STR_LIT:input>' , '<STR_LIT>' ) <EOL> self . create_file ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def test_task_get ( self ) : <EOL> execution = self . execution_create ( self . direct_wf [ '<STR_LIT:Name>' ] ) <EOL> exec_id = self . get_value_of_field ( execution , '<STR_LIT>' ) <EOL> tasks = self . mistral_admin ( '<STR_LIT>' ) <EOL> created_task_id = tasks [ - <NUM_LIT:1> ] [ '<STR_LIT>' ] <EOL> fetched_task = self . mistral_admin ( '<STR_LIT>' , params = created_task_id ) <EOL> fetched_task_id = self . get_value_of_field ( fetched_task , '<STR_LIT>' ) <EOL> task_execution_id = self . get_value_of_field ( fetched_task , <EOL> '<STR_LIT>' ) <EOL> self . assertEqual ( created_task_id , fetched_task_id ) <EOL> self . assertEqual ( exec_id , task_execution_id ) <EOL> class ActionCLITests ( base_v2 . MistralClientTestBase ) : <EOL> """<STR_LIT>""" <EOL> @ classmethod <EOL> def setUpClass ( cls ) : <EOL> super ( ActionCLITests , cls ) . setUpClass ( ) <EOL> def test_action_create_delete ( self ) : <EOL> init_acts = self . mistral_admin ( <EOL> '<STR_LIT>' , params = self . act_def ) <EOL> self . assertTableStruct ( init_acts , [ '<STR_LIT:Name>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> self . assertIn ( '<STR_LIT>' , [ action [ '<STR_LIT:Name>' ] for action in init_acts ] ) <EOL> self . assertIn ( '<STR_LIT>' , [ action [ '<STR_LIT:Name>' ] for action in init_acts ] ) <EOL> action_1 = self . get_item_info ( <EOL> get_from = init_acts , get_by = '<STR_LIT:Name>' , value = '<STR_LIT>' ) <EOL> action_2 = self . get_item_info ( <EOL> get_from = init_acts , get_by = '<STR_LIT:Name>' , value = '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , action_1 [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( '<STR_LIT>' , action_2 [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( '<STR_LIT:False>' , action_1 [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( '<STR_LIT:False>' , action_2 [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( '<STR_LIT:name>' , action_1 [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( '<STR_LIT:None>' , action_2 [ '<STR_LIT>' ] ) <EOL> acts = self . mistral_admin ( '<STR_LIT>' ) <EOL> self . assertIn ( action_1 [ '<STR_LIT:Name>' ] , [ action [ '<STR_LIT:Name>' ] for action in acts ] ) <EOL> self . assertIn ( action_2 [ '<STR_LIT:Name>' ] , [ action [ '<STR_LIT:Name>' ] for action in acts ] ) <EOL> self . mistral_admin ( <EOL> '<STR_LIT>' , params = '<STR_LIT>' . format ( action_1 [ '<STR_LIT:Name>' ] ) ) <EOL> self . mistral_admin ( <EOL> '<STR_LIT>' , params = '<STR_LIT>' . format ( action_2 [ '<STR_LIT:Name>' ] ) ) <EOL> acts = self . mistral_admin ( '<STR_LIT>' ) <EOL> self . assertNotIn ( action_1 [ '<STR_LIT:Name>' ] , [ action [ '<STR_LIT:Name>' ] for action in acts ] ) <EOL> self . assertNotIn ( action_2 [ '<STR_LIT:Name>' ] , [ action [ '<STR_LIT:Name>' ] for action in acts ] ) <EOL> def test_action_update ( self ) : <EOL> acts = self . action_create ( self . act_def ) <EOL> created_action = self . get_item_info ( <EOL> get_from = acts , get_by = '<STR_LIT:Name>' , value = '<STR_LIT>' ) <EOL> acts = self . mistral_admin ( '<STR_LIT>' , params = self . act_def ) <EOL> updated_action = self . get_item_info ( <EOL> get_from = acts , get_by = '<STR_LIT:Name>' , value = '<STR_LIT>' ) <EOL> self . assertEqual ( created_action [ '<STR_LIT>' ] . split ( "<STR_LIT:.>" ) [ <NUM_LIT:0> ] , <EOL> updated_action [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( created_action [ '<STR_LIT:Name>' ] , updated_action [ '<STR_LIT:Name>' ] ) <EOL> self . assertEqual ( created_action [ '<STR_LIT>' ] , <EOL> updated_action [ '<STR_LIT>' ] ) <EOL> acts = self . mistral_admin ( '<STR_LIT>' , params = self . act_tag_def ) <EOL> updated_action = self . get_item_info ( <EOL> get_from = acts , get_by = '<STR_LIT:Name>' , value = '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , updated_action [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( created_action [ '<STR_LIT>' ] . split ( "<STR_LIT:.>" ) [ <NUM_LIT:0> ] , <EOL> updated_action [ '<STR_LIT>' ] ) <EOL> self . assertEqual ( created_action [ '<STR_LIT:Name>' ] , updated_action [ '<STR_LIT:Name>' ] ) <EOL> self . assertNotEqual ( created_action [ '<STR_LIT>' ] , <EOL> updated_action [ '<STR_LIT>' ] ) <EOL> def test_action_update_truncate_input ( self ) : <EOL> input_value = "<STR_LIT>" <EOL> act_def = """<STR_LIT>""" . format ( input_value ) <EOL> self . create_file ( '<STR_LIT>' , act_def ) <EOL> self . action_create ( '<STR_LIT>' ) <EOL> upd_act = self . mistral_admin ( <EOL> '<STR_LIT>' , params = '<STR_LIT>' ) <EOL> upd_act_info = self . get_item_info ( <EOL> get_from = upd_act , get_by = '<STR_LIT:Name>' , value = '<STR_LIT>' ) <EOL> self . assertEqual ( upd_act_info [ '<STR_LIT>' ] [ : - <NUM_LIT:3> ] , input_value [ : <NUM_LIT> ] ) <EOL> def test_action_get_definition ( self ) : <EOL> self . action_create ( self . act_def ) <EOL> definition = self . mistral_admin ( <EOL> '<STR_LIT>' , params = '<STR_LIT>' ) <EOL> self . assertNotIn ( '<STR_LIT>' , definition ) <EOL> class EnvironmentCLITests ( base_v2 . MistralClientTestBase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> super ( EnvironmentCLITests , self ) . setUp ( ) <EOL> self . create_file ( '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def test_environment_create ( self ) : <EOL> env = self . mistral_admin ( '<STR_LIT>' , params = '<STR_LIT>' ) <EOL> env_name = self . get_value_of_field ( env , '<STR_LIT:Name>' ) <EOL> env_desc = self . get_value_of_field ( env , '<STR_LIT>' ) <EOL> self . assertTableStruct ( env , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> envs = self . mistral_admin ( '<STR_LIT>' ) <EOL> self . assertIn ( env_name , [ en [ '<STR_LIT:Name>' ] for en in envs ] ) <EOL> self . assertIn ( env_desc , [ en [ '<STR_LIT>' ] for en in envs ] ) <EOL> self . mistral_admin ( '<STR_LIT>' , params = env_name ) <EOL> envs = self . mistral_admin ( '<STR_LIT>' ) <EOL> self . assertNotIn ( env_name , [ en [ '<STR_LIT:Name>' ] for en in envs ] ) <EOL> def test_environment_create_without_description ( self ) : <EOL> self . create_file ( '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> env = self . mistral_admin ( '<STR_LIT>' , <EOL> params = '<STR_LIT>' ) <EOL> env_name = self . get_value_of_field ( env , '<STR_LIT:Name>' ) <EOL> env_desc = self . get_value_of_field ( env , '<STR_LIT>' ) <EOL> self . assertTableStruct ( env , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> envs = self . mistral_admin ( '<STR_LIT>' ) <EOL> self . assertIn ( env_name , [ en [ '<STR_LIT:Name>' ] for en in envs ] ) <EOL> self . assertIn ( env_desc , '<STR_LIT:None>' ) <EOL> self . mistral_admin ( '<STR_LIT>' , params = '<STR_LIT>' ) <EOL> envs = self . mistral_admin ( '<STR_LIT>' ) <EOL> self . assertNotIn ( env_name , [ en [ '<STR_LIT:Name>' ] for en in envs ] ) <EOL> def test_environment_update ( self ) : <EOL> env = self . environment_create ( '<STR_LIT>' ) <EOL> env_name = self . get_value_of_field ( env , '<STR_LIT:Name>' ) <EOL> env_desc = self . get_value_of_field ( env , '<STR_LIT>' ) <EOL> env_created_at = self . get_value_of_field ( env , '<STR_LIT>' ) <EOL> env_updated_at = self . get_value_of_field ( env , '<STR_LIT>' ) <EOL> self . assertIsNotNone ( env_created_at ) <EOL> self . assertEqual ( '<STR_LIT:None>' , env_updated_at ) <EOL> self . create_file ( '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> env = self . mistral_admin ( '<STR_LIT>' , params = '<STR_LIT>' ) <EOL> self . assertTableStruct ( env , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> updated_env_name = self . get_value_of_field ( env , '<STR_LIT:Name>' ) <EOL> updated_env_desc = self . get_value_of_field ( env , '<STR_LIT>' ) <EOL> updated_env_created_at = self . get_value_of_field ( env , '<STR_LIT>' ) <EOL> updated_env_updated_at = self . get_value_of_field ( env , '<STR_LIT>' ) <EOL> self . assertEqual ( env_name , updated_env_name ) <EOL> self . assertNotEqual ( env_desc , updated_env_desc ) <EOL> self . assertEqual ( '<STR_LIT>' , updated_env_desc ) <EOL> self . assertEqual ( env_created_at . split ( '<STR_LIT:.>' ) [ <NUM_LIT:0> ] , updated_env_created_at ) <EOL> self . assertIsNotNone ( updated_env_updated_at ) <EOL> def test_environment_get ( self ) : <EOL> env = self . environment_create ( '<STR_LIT>' ) <EOL> env_name = self . get_value_of_field ( env , '<STR_LIT:Name>' ) <EOL> env_desc = self . get_value_of_field ( env , '<STR_LIT>' ) <EOL> env = self . mistral_admin ( '<STR_LIT>' , params = env_name ) <EOL> fetched_env_name = self . get_value_of_field ( env , '<STR_LIT:Name>' ) <EOL> fetched_env_desc = self . get_value_of_field ( env , '<STR_LIT>' ) <EOL> self . assertTableStruct ( env , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> self . assertEqual ( env_name , fetched_env_name ) <EOL> self . assertEqual ( env_desc , fetched_env_desc ) <EOL> class ActionExecutionCLITests ( base_v2 . MistralClientTestBase ) : <EOL> """<STR_LIT>""" <EOL> def setUp ( self ) : <EOL> super ( ActionExecutionCLITests , self ) . setUp ( ) <EOL> wfs = self . workflow_create ( self . wf_def ) <EOL> self . direct_wf = wfs [ <NUM_LIT:0> ] <EOL> direct_wf_exec = self . execution_create ( self . direct_wf [ '<STR_LIT:Name>' ] ) <EOL> self . direct_ex_id = self . get_value_of_field ( direct_wf_exec , '<STR_LIT>' ) <EOL> def test_act_execution_get ( self ) : <EOL> self . wait_execution_success ( self . direct_ex_id ) <EOL> task = self . mistral_admin ( <EOL> '<STR_LIT>' , params = self . direct_ex_id ) [ <NUM_LIT:0> ] <EOL> act_ex_from_list = self . mistral_admin ( <EOL> '<STR_LIT>' , params = task [ '<STR_LIT>' ] ) [ <NUM_LIT:0> ] <EOL> act_ex = self . mistral_admin ( <EOL> '<STR_LIT>' , params = act_ex_from_list [ '<STR_LIT>' ] ) <EOL> wf_name = self . get_value_of_field ( act_ex , '<STR_LIT>' ) <EOL> status = self . get_value_of_field ( act_ex , '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> act_ex_from_list [ '<STR_LIT>' ] , <EOL> self . get_value_of_field ( act_ex , '<STR_LIT>' ) <EOL> ) <EOL> self . assertEqual ( self . direct_wf [ '<STR_LIT:Name>' ] , wf_name ) <EOL> self . assertEqual ( '<STR_LIT>' , status ) <EOL> def test_act_execution_create_delete ( self ) : <EOL> action_ex = self . mistral_admin ( <EOL> '<STR_LIT>' , <EOL> params = "<STR_LIT>" . format ( <EOL> '<STR_LIT>' ) <EOL> ) <EOL> action_ex_id = self . get_value_of_field ( action_ex , '<STR_LIT>' ) <EOL> self . assertTableStruct ( action_ex , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> name = self . get_value_of_field ( action_ex , '<STR_LIT:Name>' ) <EOL> wf_name = self . get_value_of_field ( action_ex , '<STR_LIT>' ) <EOL> task_name = self . get_value_of_field ( action_ex , '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , name ) <EOL> self . assertEqual ( '<STR_LIT:None>' , wf_name ) <EOL> self . assertEqual ( '<STR_LIT:None>' , task_name ) <EOL> action_exs = self . mistral_admin ( '<STR_LIT>' ) <EOL> self . assertIn ( action_ex_id , [ ex [ '<STR_LIT>' ] for ex in action_exs ] ) <EOL> self . mistral_admin ( '<STR_LIT>' , params = action_ex_id ) <EOL> action_exs = self . mistral_admin ( '<STR_LIT>' ) <EOL> self . assertNotIn ( action_ex_id , [ ex [ '<STR_LIT>' ] for ex in action_exs ] ) <EOL> class NegativeCLITests ( base_v2 . MistralClientTestBase ) : <EOL> """<STR_LIT>""" <EOL> def test_wb_list_extra_param ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' ) <EOL> def test_wb_get_unexist_wb ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT:wb>' ) <EOL> def test_wb_get_without_param ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' ) <EOL> def test_wb_create_same_name ( self ) : <EOL> self . workbook_create ( self . wb_def ) <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . workbook_create , <EOL> self . wb_def ) <EOL> def test_wb_create_with_wrong_path_to_definition ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT:wb>' ) <EOL> def test_wb_delete_unexist_wb ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT:wb>' ) <EOL> def test_wb_update_wrong_path_to_def ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT:wb>' ) <EOL> def test_wb_update_nonexistant_wb ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = self . wb_with_tags_def ) <EOL> def test_wb_create_empty_def ( self ) : <EOL> self . create_file ( '<STR_LIT>' ) <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' ) <EOL> def test_wb_update_empty_def ( self ) : <EOL> self . create_file ( '<STR_LIT>' ) <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' ) <EOL> def test_wb_get_definition_unexist_wb ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT:wb>' ) <EOL> def test_wb_create_invalid_def ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = self . wf_def ) <EOL> def test_wb_update_invalid_def ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = self . wf_def ) <EOL> def test_wb_update_without_def ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' ) <EOL> def test_wf_list_extra_param ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' ) <EOL> def test_wf_get_unexist_wf ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' ) <EOL> def test_wf_get_without_param ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' ) <EOL> def test_wf_create_without_definition ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' ) <EOL> def test_wf_create_with_wrong_path_to_definition ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' ) <EOL> def test_wf_delete_unexist_wf ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' ) <EOL> def test_wf_update_unexist_wf ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' ) <EOL> def test_wf_get_definition_unexist_wf ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' ) <EOL> def test_wf_get_definition_missed_param ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' ) <EOL> def test_wf_create_invalid_def ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = self . wb_def ) <EOL> def test_wf_update_invalid_def ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = self . wb_def ) <EOL> def test_wf_create_empty_def ( self ) : <EOL> self . create_file ( '<STR_LIT>' ) <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' ) <EOL> def test_wf_update_empty_def ( self ) : <EOL> self . create_file ( '<STR_LIT>' ) <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' ) <EOL> def test_ex_list_extra_param ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' ) <EOL> def test_ex_create_unexist_wf ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' ) <EOL> def test_ex_create_unexist_task ( self ) : <EOL> wf = self . workflow_create ( self . wf_def ) <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' % wf [ <NUM_LIT:0> ] [ '<STR_LIT:Name>' ] ) <EOL> def test_ex_create_with_invalid_input ( self ) : <EOL> wf = self . workflow_create ( self . wf_def ) <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = "<STR_LIT>" % wf [ <NUM_LIT:0> ] [ '<STR_LIT:Name>' ] ) <EOL> def test_ex_get_nonexist_execution ( self ) : <EOL> wf = self . workflow_create ( self . wf_def ) <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' % wf [ <NUM_LIT:0> ] [ '<STR_LIT:Name>' ] ) <EOL> def test_ex_create_without_wf_name ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' ) <EOL> def test_ex_create_reverse_wf_without_start_task ( self ) : <EOL> wf = self . workflow_create ( self . wf_def ) <EOL> self . create_file ( '<STR_LIT:input>' , '<STR_LIT>' ) <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = wf [ <NUM_LIT:1> ] [ '<STR_LIT:Name>' ] ) <EOL> def test_ex_create_missed_input ( self ) : <EOL> self . create_file ( '<STR_LIT>' ) <EOL> wf = self . workflow_create ( self . wf_def ) <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = wf [ <NUM_LIT:1> ] [ '<STR_LIT:Name>' ] ) <EOL> def test_ex_update_both_state_and_description ( self ) : <EOL> wf = self . workflow_create ( self . wf_def ) <EOL> execution = self . execution_create ( params = wf [ <NUM_LIT:0> ] [ '<STR_LIT:Name>' ] ) <EOL> exec_id = self . get_value_of_field ( execution , '<STR_LIT>' ) <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' % exec_id ) <EOL> def test_ex_delete_nonexistent_execution ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , params = '<STR_LIT>' ) <EOL> def test_tr_create_without_pattern ( self ) : <EOL> wf = self . workflow_create ( self . wf_def ) <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' % wf [ <NUM_LIT:0> ] [ '<STR_LIT:Name>' ] ) <EOL> def test_tr_create_invalid_pattern ( self ) : <EOL> wf = self . workflow_create ( self . wf_def ) <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' % wf [ <NUM_LIT:0> ] [ '<STR_LIT:Name>' ] ) <EOL> def test_tr_create_invalid_pattern_value_out_of_range ( self ) : <EOL> wf = self . workflow_create ( self . wf_def ) <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' <EOL> '<STR_LIT>' % wf [ <NUM_LIT:0> ] [ '<STR_LIT:Name>' ] ) <EOL> def test_tr_create_nonexistent_wf ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' ) <EOL> def test_tr_delete_nonexistant_tr ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' ) <EOL> def test_tr_get_nonexistant_tr ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' ) <EOL> def test_tr_create_invalid_count ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def test_tr_create_negative_count ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def test_tr_create_invalid_first_date ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def test_tr_create_count_only ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' ) <EOL> def test_tr_create_date_and_count_without_pattern ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def test_action_get_nonexistent ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' ) <EOL> def test_action_double_creation ( self ) : <EOL> self . action_create ( self . act_def ) <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' . format ( self . act_def ) ) <EOL> def test_action_create_without_def ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' ) <EOL> def test_action_create_invalid_def ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' . format ( self . wb_def ) ) <EOL> def test_action_delete_nonexistent_act ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' ) <EOL> def test_action_delete_standard_action ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' ) <EOL> def test_action_get_definition_nonexistent_action ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' ) <EOL> def test_task_get_nonexistent_task ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' ) <EOL> def test_env_get_without_param ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' ) <EOL> def test_env_get_nonexistent ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' ) <EOL> def test_env_create_same_name ( self ) : <EOL> self . create_file ( '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . environment_create ( '<STR_LIT>' ) <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . environment_create , <EOL> '<STR_LIT>' ) <EOL> def test_env_create_empty ( self ) : <EOL> self . create_file ( '<STR_LIT>' ) <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . environment_create , <EOL> '<STR_LIT>' ) <EOL> def test_env_create_with_wrong_path_to_definition ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> def test_env_delete_unexist_env ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' ) <EOL> def test_env_update_wrong_path_to_def ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' ) <EOL> def test_env_update_empty ( self ) : <EOL> self . create_file ( '<STR_LIT>' ) <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' ) <EOL> def test_env_update_nonexistant_env ( self ) : <EOL> self . create_file ( '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' ) <EOL> def test_env_create_without_name ( self ) : <EOL> self . create_file ( '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' ) <EOL> def test_env_create_without_variables ( self ) : <EOL> self . create_file ( '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' ) <EOL> def test_action_execution_get_without_params ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' ) <EOL> def test_action_execution_get_unexistent_obj ( self ) : <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' ) <EOL> def test_action_execution_update ( self ) : <EOL> wfs = self . workflow_create ( self . wf_def ) <EOL> direct_wf_exec = self . execution_create ( wfs [ <NUM_LIT:0> ] [ '<STR_LIT:Name>' ] ) <EOL> direct_ex_id = self . get_value_of_field ( direct_wf_exec , '<STR_LIT>' ) <EOL> self . assertRaises ( exceptions . CommandFailed , <EOL> self . mistral_admin , <EOL> '<STR_LIT>' , <EOL> params = '<STR_LIT>' % direct_ex_id ) </s>
