<s> import time <EOL> import uuid <EOL> from testenv . utils import wait_for_execution_to_end <EOL> from testenv . utils import do_retries <EOL> from testenv . utils import verify_deployment_environment_creation_complete <EOL> from testenv . utils import get_resource as resource <EOL> from testenv . utils import deploy_application as deploy <EOL> from testenv import TestCase <EOL> from cloudify_rest_client . executions import Execution <EOL> class ExecutionsTest ( TestCase ) : <EOL> def test_cancel_execution ( self ) : <EOL> execution , deployment_id = self . _execute_and_cancel_execution ( <EOL> '<STR_LIT>' ) <EOL> self . _assert_execution_cancelled ( execution , deployment_id ) <EOL> def test_force_cancel_execution ( self ) : <EOL> execution , deployment_id = self . _execute_and_cancel_execution ( <EOL> '<STR_LIT>' , True ) <EOL> self . _assert_execution_cancelled ( execution , deployment_id ) <EOL> def test_cancel_execution_with_graph_workflow ( self ) : <EOL> execution , deployment_id = self . _execute_and_cancel_execution ( <EOL> '<STR_LIT>' ) <EOL> self . _assert_execution_cancelled ( execution , deployment_id ) <EOL> def test_cancel_execution_and_then_force_cancel ( self ) : <EOL> execution , deployment_id = self . _execute_and_cancel_execution ( <EOL> '<STR_LIT>' , False , False ) <EOL> execution = self . client . executions . cancel ( execution . id , True ) <EOL> self . assertEquals ( Execution . FORCE_CANCELLING , execution . status ) <EOL> wait_for_execution_to_end ( execution ) <EOL> execution = self . client . executions . get ( execution . id ) <EOL> self . _assert_execution_cancelled ( execution , deployment_id ) <EOL> def test_legacy_cancel_execution ( self ) : <EOL> execution , deployment_id = self . _execute_and_cancel_execution ( <EOL> '<STR_LIT>' , <EOL> workflow_params = { '<STR_LIT>' : True } ) <EOL> self . _assert_execution_cancelled ( execution , deployment_id ) <EOL> def test_cancel_execution_before_it_started ( self ) : <EOL> execution , deployment_id = self . _execute_and_cancel_execution ( <EOL> '<STR_LIT>' , False , True , False ) <EOL> self . assertEquals ( Execution . CANCELLED , execution . status ) <EOL> data = self . get_plugin_data ( <EOL> plugin_name = '<STR_LIT>' , <EOL> deployment_id = deployment_id <EOL> ) <EOL> self . assertEqual ( data , { } ) <EOL> def test_get_deployments_executions_with_status ( self ) : <EOL> dsl_path = resource ( "<STR_LIT>" ) <EOL> deployment , execution_id = deploy ( dsl_path ) <EOL> def assertions ( ) : <EOL> deployments_executions = self . client . executions . list ( <EOL> deployment_id = deployment . id ) <EOL> self . assertEquals ( <NUM_LIT:2> , len ( deployments_executions ) ) <EOL> self . assertIn ( execution_id , [ deployments_executions [ <NUM_LIT:0> ] . id , <EOL> deployments_executions [ <NUM_LIT:1> ] . id ] ) <EOL> install_execution = deployments_executions [ <NUM_LIT:0> ] if execution_id == deployments_executions [ <NUM_LIT:0> ] . id else deployments_executions [ <NUM_LIT:1> ] <EOL> self . assertEquals ( Execution . TERMINATED , install_execution . status ) <EOL> self . assertEquals ( '<STR_LIT>' , install_execution . error ) <EOL> self . do_assertions ( assertions , timeout = <NUM_LIT:10> ) <EOL> def test_execution_parameters ( self ) : <EOL> dsl_path = resource ( '<STR_LIT>' ) <EOL> _id = uuid . uuid1 ( ) <EOL> blueprint_id = '<STR_LIT>' . format ( _id ) <EOL> deployment_id = '<STR_LIT>' . format ( _id ) <EOL> self . client . blueprints . upload ( dsl_path , blueprint_id ) <EOL> self . client . deployments . create ( blueprint_id , deployment_id ) <EOL> do_retries ( verify_deployment_environment_creation_complete , <NUM_LIT> , <EOL> deployment_id = deployment_id ) <EOL> execution_parameters = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:key>' : '<STR_LIT>' , <EOL> '<STR_LIT:value>' : '<STR_LIT>' <EOL> } , <EOL> '<STR_LIT>' : "<STR_LIT>" <EOL> } <EOL> execution = self . client . executions . start ( <EOL> deployment_id , '<STR_LIT>' , <EOL> parameters = execution_parameters , <EOL> allow_custom_parameters = True ) <EOL> wait_for_execution_to_end ( execution ) <EOL> invocations = self . get_plugin_data ( <EOL> plugin_name = '<STR_LIT>' , <EOL> deployment_id = deployment_id <EOL> ) [ '<STR_LIT>' ] <EOL> self . assertEqual ( <NUM_LIT:1> , len ( invocations ) ) <EOL> self . assertDictEqual ( invocations [ <NUM_LIT:0> ] , <EOL> { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> expected_params = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT:key>' : '<STR_LIT>' , <EOL> '<STR_LIT:value>' : '<STR_LIT>' <EOL> } , <EOL> '<STR_LIT>' : "<STR_LIT>" <EOL> } <EOL> self . assertEqual ( expected_params , execution . parameters ) <EOL> def test_update_execution_status ( self ) : <EOL> dsl_path = resource ( "<STR_LIT>" ) <EOL> _ , execution_id = deploy ( dsl_path , wait_for_execution = True ) <EOL> execution = self . client . executions . get ( execution_id ) <EOL> self . assertEquals ( Execution . TERMINATED , execution . status ) <EOL> self . es_db_client . update_execution_status ( <EOL> execution_id , '<STR_LIT>' , '<STR_LIT>' ) <EOL> execution = self . client . executions . get ( execution_id ) <EOL> self . assertEquals ( '<STR_LIT>' , execution . status ) <EOL> execution = self . client . executions . update ( execution_id , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ) <EOL> self . assertEquals ( '<STR_LIT>' , execution . status ) <EOL> self . assertEquals ( '<STR_LIT>' , execution . error ) <EOL> execution = self . client . executions . update ( execution_id , '<STR_LIT>' ) <EOL> self . assertEquals ( '<STR_LIT>' , execution . status ) <EOL> self . assertEquals ( '<STR_LIT>' , execution . error ) <EOL> def _execute_and_cancel_execution ( self , workflow_id , force = False , <EOL> wait_for_termination = True , <EOL> is_wait_for_asleep_node = True , <EOL> workflow_params = None ) : <EOL> dsl_path = resource ( '<STR_LIT>' ) <EOL> _id = uuid . uuid1 ( ) <EOL> blueprint_id = '<STR_LIT>' . format ( _id ) <EOL> deployment_id = '<STR_LIT>' . format ( _id ) <EOL> self . client . blueprints . upload ( dsl_path , blueprint_id ) <EOL> self . client . deployments . create ( blueprint_id , deployment_id ) <EOL> do_retries ( verify_deployment_environment_creation_complete , <NUM_LIT:30> , <EOL> deployment_id = deployment_id ) <EOL> execution = self . client . executions . start ( <EOL> deployment_id , workflow_id , parameters = workflow_params ) <EOL> node_inst_id = self . client . node_instances . list ( deployment_id ) [ <NUM_LIT:0> ] . id <EOL> if is_wait_for_asleep_node : <EOL> for retry in range ( <NUM_LIT:30> ) : <EOL> if self . client . node_instances . get ( <EOL> node_inst_id ) . state == '<STR_LIT>' : <EOL> break <EOL> time . sleep ( <NUM_LIT:1> ) <EOL> else : <EOL> raise RuntimeError ( "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> execution = self . client . executions . cancel ( execution . id , force ) <EOL> expected_status = Execution . FORCE_CANCELLING if force else Execution . CANCELLING <EOL> self . assertEquals ( expected_status , execution . status ) <EOL> if wait_for_termination : <EOL> wait_for_execution_to_end ( execution ) <EOL> execution = self . client . executions . get ( execution . id ) <EOL> return execution , deployment_id <EOL> def _assert_execution_cancelled ( self , execution , deployment_id ) : <EOL> self . assertEquals ( Execution . CANCELLED , execution . status ) <EOL> invocations = self . get_plugin_data ( <EOL> plugin_name = '<STR_LIT>' , <EOL> deployment_id = deployment_id <EOL> ) [ '<STR_LIT>' ] <EOL> self . assertEqual ( <NUM_LIT:1> , len ( invocations ) ) <EOL> self . assertDictEqual ( invocations [ <NUM_LIT:0> ] , { '<STR_LIT>' : None } ) </s>
