<s> import uuid <EOL> import os <EOL> import json <EOL> from testenv import TestCase <EOL> from testenv . utils import get_resource as resource <EOL> from testenv . utils import verify_deployment_environment_creation_complete <EOL> from testenv . utils import do_retries <EOL> from testenv . utils import wait_for_execution_to_end <EOL> from testenv . utils import deploy_application as deploy <EOL> from cloudify_rest_client . exceptions import CloudifyClientError <EOL> class TestDeploymentWorkflows ( TestCase ) : <EOL> def test_deployment_workflows ( self ) : <EOL> dsl_path = resource ( "<STR_LIT>" ) <EOL> deployment , _ = deploy ( dsl_path ) <EOL> deployment_id = deployment . id <EOL> workflows = self . client . deployments . get ( deployment_id ) . workflows <EOL> self . assertEqual ( <NUM_LIT:8> , len ( workflows ) ) <EOL> wf_ids = [ x . name for x in workflows ] <EOL> self . assertIn ( '<STR_LIT>' , wf_ids ) <EOL> self . assertIn ( '<STR_LIT>' , wf_ids ) <EOL> self . assertIn ( '<STR_LIT>' , wf_ids ) <EOL> self . assertIn ( '<STR_LIT>' , wf_ids ) <EOL> self . assertIn ( '<STR_LIT>' , wf_ids ) <EOL> self . assertIn ( '<STR_LIT>' , wf_ids ) <EOL> self . assertIn ( '<STR_LIT>' , wf_ids ) <EOL> self . assertIn ( '<STR_LIT>' , wf_ids ) <EOL> def test_workflow_parameters_pass_from_blueprint ( 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:30> , <EOL> deployment_id = deployment_id ) <EOL> execution = self . client . executions . start ( deployment_id , <EOL> '<STR_LIT>' ) <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> ] , { '<STR_LIT>' : '<STR_LIT>' } ) <EOL> def test_get_workflow_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> workflows = self . client . deployments . get ( deployment_id ) . workflows <EOL> execute_op_workflow = next ( wf for wf in workflows if <EOL> wf . name == '<STR_LIT>' ) <EOL> expected_params = { <EOL> u'<STR_LIT>' : { u'<STR_LIT:default>' : u'<STR_LIT>' } , <EOL> u'<STR_LIT>' : { } , <EOL> u'<STR_LIT>' : { <EOL> u'<STR_LIT:default>' : { <EOL> u'<STR_LIT:key>' : u'<STR_LIT>' , <EOL> u'<STR_LIT:value>' : u'<STR_LIT>' <EOL> } <EOL> } <EOL> } <EOL> self . assertEqual ( expected_params , execute_op_workflow . parameters ) <EOL> def test_delete_botched_deployment ( self ) : <EOL> from testenv import testenv_instance <EOL> storage_file_path = os . path . join ( <EOL> testenv_instance . plugins_storage_dir , <EOL> '<STR_LIT>' <EOL> ) <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> data = { <EOL> deployment_id : { '<STR_LIT>' : True } <EOL> } <EOL> with open ( storage_file_path , '<STR_LIT:w>' ) as f : <EOL> json . dump ( data , f ) <EOL> self . client . blueprints . upload ( dsl_path , blueprint_id ) <EOL> self . client . deployments . create ( blueprint_id , deployment_id ) <EOL> execution = self . client . executions . list ( deployment_id , <EOL> include_system_workflows = True ) [ <NUM_LIT:0> ] <EOL> wait_for_execution_to_end ( execution ) <EOL> self . client . deployments . delete ( deployment_id ) <EOL> try : <EOL> self . client . deployments . get ( deployment_id ) <EOL> self . fail ( "<STR_LIT>" ) <EOL> except CloudifyClientError as e : <EOL> self . assertEquals ( <NUM_LIT> , e . status_code ) </s>
