<s> import copy <EOL> import uuid <EOL> import dateutil . parser <EOL> from datetime import datetime , timedelta <EOL> from nose . plugins . attrib import attr <EOL> from base_test import CLIENT_API_VERSION <EOL> from manager_rest . test import base_test <EOL> from cloudify_rest_client import exceptions <EOL> from cloudify_rest_client . deployment_modifications import ( <EOL> DeploymentModification ) <EOL> @ attr ( client_min_version = <NUM_LIT:1> , client_max_version = base_test . LATEST_API_VERSION ) <EOL> class ModifyTests ( base_test . BaseServerTestCase ) : <EOL> def test_data_model_with_finish ( self ) : <EOL> def expected_after_end_func ( _ , before_end ) : <EOL> return before_end <EOL> self . _test_data_model_impl ( <EOL> end_func = self . client . deployment_modifications . finish , <EOL> expected_end_status = DeploymentModification . FINISHED , <EOL> expected_end_node_counts = { <EOL> '<STR_LIT>' : <NUM_LIT:2> , '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT>' : <NUM_LIT:2> } , <EOL> expected_before_end_func = lambda before_end : [ ] , <EOL> expected_after_end_func = expected_after_end_func , <EOL> expected_after_end_count = <NUM_LIT:3> , <EOL> expected_after_end_runtime_property = '<STR_LIT>' ) <EOL> def test_data_model_with_rollback ( self ) : <EOL> def expected_after_end_func ( before_modification , _ ) : <EOL> return before_modification <EOL> self . _test_data_model_impl ( <EOL> end_func = self . client . deployment_modifications . rollback , <EOL> expected_end_status = DeploymentModification . ROLLEDBACK , <EOL> expected_end_node_counts = { <EOL> '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT>' : <NUM_LIT:1> , '<STR_LIT>' : <NUM_LIT:1> } , <EOL> expected_before_end_func = lambda before_end : before_end , <EOL> expected_after_end_func = expected_after_end_func , <EOL> expected_after_end_count = <NUM_LIT:2> , <EOL> expected_after_end_runtime_property = '<STR_LIT>' ) <EOL> def _test_data_model_impl ( <EOL> self , <EOL> end_func , <EOL> expected_end_status , <EOL> expected_end_node_counts , <EOL> expected_before_end_func , <EOL> expected_after_end_func , <EOL> expected_after_end_count , <EOL> expected_after_end_runtime_property ) : <EOL> def node_assertions ( num , deploy_num , planned_num ) : <EOL> node = self . client . nodes . get ( deployment . id , '<STR_LIT>' ) <EOL> self . assertEqual ( node . number_of_instances , num ) <EOL> self . assertEqual ( node . planned_number_of_instances , planned_num ) <EOL> self . assertEqual ( node . deploy_number_of_instances , deploy_num ) <EOL> _ , _ , _ , deployment = self . put_deployment ( <EOL> deployment_id = str ( uuid . uuid4 ( ) ) , <EOL> blueprint_file_name = '<STR_LIT>' ) <EOL> node_assertions ( num = <NUM_LIT:1> , deploy_num = <NUM_LIT:1> , planned_num = <NUM_LIT:1> ) <EOL> mock_context = { '<STR_LIT>' : '<STR_LIT:data>' } <EOL> node1_instance = self . client . node_instances . list ( <EOL> deployment_id = deployment . id , node_name = '<STR_LIT>' ) [ <NUM_LIT:0> ] <EOL> self . client . node_instances . update ( <EOL> node1_instance . id , <EOL> runtime_properties = { '<STR_LIT:test>' : '<STR_LIT>' } , <EOL> version = <NUM_LIT:0> ) <EOL> node2_instance = self . client . node_instances . list ( <EOL> deployment_id = deployment . id , node_name = '<STR_LIT>' ) [ <NUM_LIT:0> ] <EOL> self . client . node_instances . update ( <EOL> node2_instance . id , <EOL> runtime_properties = { '<STR_LIT:test>' : '<STR_LIT>' } , <EOL> version = <NUM_LIT:0> ) <EOL> before_modification = self . list_items ( self . client . node_instances . list , <EOL> deployment . id ) <EOL> modified_nodes = { '<STR_LIT>' : { '<STR_LIT>' : <NUM_LIT:2> } } <EOL> modification = self . client . deployment_modifications . start ( <EOL> deployment . id , nodes = modified_nodes , context = mock_context ) <EOL> self . assertEqual ( modification . node_instances . before_modification , <EOL> before_modification ) <EOL> self . assertIsNone ( modification . ended_at ) <EOL> self . client . node_instances . update ( <EOL> node1_instance . id , <EOL> runtime_properties = { '<STR_LIT:test>' : '<STR_LIT>' } , <EOL> version = <NUM_LIT:1> ) <EOL> self . client . node_instances . update ( <EOL> node2_instance . id , <EOL> runtime_properties = { '<STR_LIT:test>' : '<STR_LIT>' } , <EOL> version = <NUM_LIT:1> ) <EOL> node_assertions ( num = <NUM_LIT:1> , deploy_num = <NUM_LIT:1> , planned_num = <NUM_LIT:2> ) <EOL> modification_id = modification . id <EOL> self . assertEqual ( modification . status , <EOL> DeploymentModification . STARTED ) <EOL> before_end = self . list_items ( self . client . node_instances . list , <EOL> deployment . id ) <EOL> end_func ( modification_id ) <EOL> after_end = self . list_items ( self . client . node_instances . list , <EOL> deployment . id ) <EOL> node_assertions ( ** expected_end_node_counts ) <EOL> modification = self . client . deployment_modifications . get ( <EOL> modification . id ) <EOL> self . assertEqual ( modification . id , modification_id ) <EOL> self . assertEqual ( modification . status , expected_end_status ) <EOL> self . assertEqual ( modification . deployment_id , deployment . id ) <EOL> self . assertEqual ( modification . modified_nodes , modified_nodes ) <EOL> created_at = dateutil . parser . parse ( modification . created_at ) <EOL> ended_at = dateutil . parser . parse ( modification . ended_at ) <EOL> self . assertTrue ( datetime . now ( ) - timedelta ( seconds = <NUM_LIT:5> ) <= <EOL> created_at <= ended_at <= datetime . now ( ) ) <EOL> all_modifications = self . list_items ( <EOL> self . client . deployment_modifications . list ) <EOL> dep_modifications = self . list_items ( <EOL> self . client . deployment_modifications . list , <EOL> deployment_id = deployment . id ) <EOL> self . assertEqual ( len ( dep_modifications ) , <NUM_LIT:1> ) <EOL> self . assertEqual ( dep_modifications [ <NUM_LIT:0> ] , modification ) <EOL> self . assertEqual ( all_modifications , dep_modifications ) <EOL> self . assertEqual ( [ ] , self . list_items ( <EOL> self . client . deployment_modifications . list , <EOL> deployment_id = '<STR_LIT>' ) ) <EOL> self . assertEqual ( modification . node_instances . before_modification , <EOL> before_modification ) <EOL> self . assertEqual ( modification . node_instances . before_rollback , <EOL> expected_before_end_func ( before_end ) ) <EOL> self . assertEqual ( after_end , <EOL> expected_after_end_func ( before_modification , <EOL> before_end ) ) <EOL> self . assertEqual ( modification . context , mock_context ) <EOL> self . assertEqual ( expected_after_end_count , len ( after_end ) ) <EOL> self . assertEqual ( <EOL> self . client . node_instances . get ( <EOL> node1_instance . id ) . runtime_properties [ '<STR_LIT:test>' ] , <EOL> expected_after_end_runtime_property ) <EOL> self . assertEqual ( <EOL> self . client . node_instances . get ( <EOL> node2_instance . id ) . runtime_properties [ '<STR_LIT:test>' ] , <EOL> expected_after_end_runtime_property ) <EOL> def list_items ( self , list_func , * args , ** kwargs ) : <EOL> if CLIENT_API_VERSION != '<STR_LIT>' : <EOL> return list_func ( * args , ** kwargs ) . items <EOL> else : <EOL> return list_func ( * args , ** kwargs ) <EOL> def test_no_concurrent_modifications ( self ) : <EOL> _ , _ , _ , deployment = self . put_deployment ( <EOL> deployment_id = str ( uuid . uuid4 ( ) ) , <EOL> blueprint_file_name = '<STR_LIT>' ) <EOL> modification = self . client . deployment_modifications . start ( <EOL> deployment . id , nodes = { } ) <EOL> with self . assertRaises ( <EOL> exceptions . ExistingStartedDeploymentModificationError ) as e : <EOL> self . client . deployment_modifications . start ( deployment . id , nodes = { } ) <EOL> self . assertIn ( modification . id , str ( e . exception ) ) <EOL> self . client . deployment_modifications . finish ( modification . id ) <EOL> self . client . deployment_modifications . start ( deployment . id , nodes = { } ) <EOL> def test_finish_and_rollback_on_ended_modification ( self ) : <EOL> def test ( end_function ) : <EOL> _ , _ , _ , deployment = self . put_deployment ( <EOL> deployment_id = str ( uuid . uuid4 ( ) ) , <EOL> blueprint_id = str ( uuid . uuid4 ( ) ) , <EOL> blueprint_file_name = '<STR_LIT>' ) <EOL> modification = self . client . deployment_modifications . start ( <EOL> deployment . id , nodes = { } ) <EOL> end_function ( modification . id ) <EOL> with self . assertRaises ( <EOL> exceptions . DeploymentModificationAlreadyEndedError ) : <EOL> self . client . deployment_modifications . finish ( modification . id ) <EOL> with self . assertRaises ( <EOL> exceptions . DeploymentModificationAlreadyEndedError ) : <EOL> self . client . deployment_modifications . rollback ( modification . id ) <EOL> test ( self . client . deployment_modifications . finish ) <EOL> test ( self . client . deployment_modifications . rollback ) <EOL> def test_finish_and_rollback_on_non_existent_modification ( self ) : <EOL> with self . assertRaises ( exceptions . CloudifyClientError ) as scope : <EOL> self . client . deployment_modifications . finish ( '<STR_LIT>' ) <EOL> self . assertEqual ( scope . exception . status_code , <NUM_LIT> ) <EOL> self . assertRegexpMatches ( str ( scope . exception ) , <EOL> '<STR_LIT>' , ) <EOL> with self . assertRaises ( exceptions . CloudifyClientError ) as scope : <EOL> self . client . deployment_modifications . rollback ( '<STR_LIT>' ) <EOL> self . assertEqual ( scope . exception . status_code , <NUM_LIT> ) <EOL> self . assertRegexpMatches ( str ( scope . exception ) , <EOL> '<STR_LIT>' , ) <EOL> def test_modify_add_instance ( self ) : <EOL> _ , _ , _ , deployment = self . put_deployment ( <EOL> deployment_id = str ( uuid . uuid4 ( ) ) , <EOL> blueprint_file_name = '<STR_LIT>' ) <EOL> node_instances1 = self . client . node_instances . list ( ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( node_instances1 ) ) <EOL> self . _assert_number_of_instances ( deployment . id , '<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT:1> ) <EOL> modified_nodes = { '<STR_LIT>' : { '<STR_LIT>' : <NUM_LIT:2> } } <EOL> modification = self . client . deployment_modifications . start ( <EOL> deployment . id , nodes = modified_nodes ) <EOL> self . _assert_number_of_instances ( deployment . id , '<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT:1> ) <EOL> node_instances2 = self . client . node_instances . list ( ) <EOL> self . assertEqual ( <NUM_LIT:3> , len ( node_instances2 ) ) <EOL> initial_instance_ids = [ i2 . id for i2 in node_instances1 ] <EOL> new_instances = [ i for i in node_instances2 <EOL> if i . id not in initial_instance_ids ] <EOL> old_instances = [ i for i in node_instances2 <EOL> if i . id in initial_instance_ids ] <EOL> self . assertEqual ( <NUM_LIT:1> , len ( new_instances ) ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( old_instances ) ) <EOL> new_instance = new_instances [ <NUM_LIT:0> ] <EOL> self . assertEqual ( '<STR_LIT>' , new_instance . node_id ) <EOL> expected_old_instances = copy . deepcopy ( node_instances1 ) <EOL> for instance in expected_old_instances : <EOL> if instance . node_id == '<STR_LIT>' : <EOL> current_relationship = instance . relationships [ <NUM_LIT:0> ] <EOL> new_relationship = copy . deepcopy ( current_relationship ) <EOL> new_relationship [ '<STR_LIT>' ] = new_instance . id <EOL> instance . relationships . append ( new_relationship ) <EOL> self . assertEqual ( sorted ( old_instances , key = lambda _i : _i . id ) , <EOL> sorted ( expected_old_instances , key = lambda _i : _i . id ) ) <EOL> added_and_related = modification . node_instances . added_and_related <EOL> self . assertEqual ( <NUM_LIT:2> , len ( added_and_related ) ) <EOL> self . client . deployment_modifications . finish ( modification . id ) <EOL> self . _assert_number_of_instances ( deployment . id , '<STR_LIT>' , <NUM_LIT:2> , <NUM_LIT:1> ) <EOL> node_instances3 = self . client . node_instances . list ( ) <EOL> self . assertEqual ( <NUM_LIT:3> , len ( node_instances3 ) ) <EOL> node1_instance_ids = [ i . id for i in node_instances3 <EOL> if i . node_id == '<STR_LIT>' ] <EOL> node2_instance = [ i for i in node_instances3 <EOL> if i . node_id == '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> node2_target_ids = [ rel [ '<STR_LIT>' ] for rel <EOL> in node2_instance . relationships ] <EOL> self . assertEqual ( set ( node1_instance_ids ) , set ( node2_target_ids ) ) <EOL> def test_modify_remove_instance ( self ) : <EOL> _ , _ , _ , deployment = self . put_deployment ( <EOL> deployment_id = str ( uuid . uuid4 ( ) ) , <EOL> blueprint_file_name = '<STR_LIT>' ) <EOL> node_instances1 = self . client . node_instances . list ( ) <EOL> self . assertEqual ( <NUM_LIT:3> , len ( node_instances1 ) ) <EOL> self . _assert_number_of_instances ( deployment . id , '<STR_LIT>' , <NUM_LIT:2> , <NUM_LIT:2> ) <EOL> modified_nodes = { '<STR_LIT>' : { '<STR_LIT>' : <NUM_LIT:1> } } <EOL> modification = self . client . deployment_modifications . start ( <EOL> deployment . id , nodes = modified_nodes ) <EOL> self . _assert_number_of_instances ( deployment . id , '<STR_LIT>' , <NUM_LIT:2> , <NUM_LIT:2> ) <EOL> node_instances2 = self . client . node_instances . list ( ) <EOL> self . assertEqual ( <NUM_LIT:3> , len ( node_instances2 ) ) <EOL> initial_instance_ids = [ i2 . id for i2 in node_instances1 ] <EOL> new_instances = [ i for i in node_instances2 <EOL> if i . id not in initial_instance_ids ] <EOL> old_instances = [ i for i in node_instances2 <EOL> if i . id in initial_instance_ids ] <EOL> self . assertEqual ( <NUM_LIT:0> , len ( new_instances ) ) <EOL> self . assertEqual ( <NUM_LIT:3> , len ( old_instances ) ) <EOL> self . assertEqual ( sorted ( old_instances , key = lambda _i : _i . id ) , <EOL> sorted ( node_instances1 , key = lambda _i : _i . id ) ) <EOL> removed_and_related = modification . node_instances . removed_and_related <EOL> self . assertEqual ( <NUM_LIT:2> , len ( removed_and_related ) ) <EOL> self . client . deployment_modifications . finish ( modification . id ) <EOL> self . _assert_number_of_instances ( deployment . id , '<STR_LIT>' , <NUM_LIT:1> , <NUM_LIT:2> ) <EOL> node_instances3 = self . client . node_instances . list ( ) <EOL> self . assertEqual ( <NUM_LIT:2> , len ( node_instances3 ) ) <EOL> node1_instance_id = [ i . id for i in node_instances3 <EOL> if i . node_id == '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> node2_instance = [ i for i in node_instances3 <EOL> if i . node_id == '<STR_LIT>' ] [ <NUM_LIT:0> ] <EOL> node2_target_ids = [ rel [ '<STR_LIT>' ] for rel <EOL> in node2_instance . relationships ] <EOL> self . assertEqual ( <NUM_LIT:1> , len ( node2_target_ids ) ) <EOL> self . assertEqual ( node1_instance_id , node2_target_ids [ <NUM_LIT:0> ] ) <EOL> def _assert_number_of_instances ( self , <EOL> deployment_id , node_id , <EOL> expected_number_of_instances , <EOL> expected_deploy_number_of_instances ) : <EOL> node = self . client . nodes . get ( deployment_id , node_id ) <EOL> self . assertEqual ( expected_deploy_number_of_instances , <EOL> node . deploy_number_of_instances ) <EOL> self . assertEqual ( expected_number_of_instances , <EOL> node . number_of_instances ) </s>
