<s> import copy <EOL> import os <EOL> import json <EOL> from awscli . customizations . emr . emrfsutils import CONSISTENT_OPTION_NAME <EOL> from awscli . customizations . emr . emrfsutils import CSE_CUSTOM_OPTION_NAME <EOL> from awscli . customizations . emr . emrfsutils import CSE_KMS_OPTION_NAME <EOL> from awscli . customizations . emr . emrfsutils import CSE_OPTION_NAME <EOL> from tests . unit . customizations . emr import EMRBaseAWSCommandParamsTest as BaseAWSCommandParamsTest <EOL> DEFAULT_INSTANCES = { <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : False , <EOL> '<STR_LIT>' : [ { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : <NUM_LIT:1> , <EOL> '<STR_LIT:Name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } ] <EOL> } <EOL> DEFAULT_CMD = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> DEFAULT_RESULT = { <EOL> '<STR_LIT:Name>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : DEFAULT_INSTANCES , <EOL> '<STR_LIT>' : True , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : "<STR_LIT>" , <EOL> '<STR_LIT>' : [ ] <EOL> } <EOL> EMPTY_EMRFS_CONFIGURATION = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : { } <EOL> } <EOL> DEFAULT_CONFIGURATIONS = [ <EOL> { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ ] , <EOL> '<STR_LIT>' : { '<STR_LIT>' : '<STR_LIT>' } <EOL> } <EOL> ] <EOL> class TestEmrfsUtils ( BaseAWSCommandParamsTest ) : <EOL> def test_consistent ( self ) : <EOL> emrfs_option_value = '<STR_LIT>' <EOL> expected_emrfs_properties = { '<STR_LIT>' : '<STR_LIT:true>' } <EOL> expected_emrfs_ba_key_values = [ <EOL> '<STR_LIT>' <EOL> ] <EOL> self . _assert_bootstrap_actions ( <EOL> emrfs_option_value , expected_emrfs_ba_key_values , <EOL> expected_emrfs_properties ) <EOL> def test_consistent_w_optional_args ( self ) : <EOL> emrfs_option_value = '<STR_LIT>' <EOL> expected_emrfs_properties = { '<STR_LIT>' : '<STR_LIT:true>' , <EOL> '<STR_LIT>' : '<STR_LIT:5>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> expected_emrfs_ba_key_values = [ <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> ] <EOL> self . _assert_bootstrap_actions ( <EOL> emrfs_option_value , expected_emrfs_ba_key_values , <EOL> expected_emrfs_properties ) <EOL> def test_consistent_false_w_optional_args ( self ) : <EOL> emrfs_option_value = '<STR_LIT>' <EOL> expected_emrfs_properties = { <EOL> '<STR_LIT>' : '<STR_LIT:false>' , <EOL> '<STR_LIT>' : '<STR_LIT:5>' } <EOL> expected_emrfs_ba_key_values = [ <EOL> '<STR_LIT>' , '<STR_LIT>' <EOL> ] <EOL> self . _assert_bootstrap_actions ( <EOL> emrfs_option_value , expected_emrfs_ba_key_values , <EOL> expected_emrfs_properties ) <EOL> def test_sse ( self ) : <EOL> emrfs_option_value = '<STR_LIT>' <EOL> expected_emrfs_ba_key_values = [ <EOL> '<STR_LIT>' <EOL> ] <EOL> expected_emrfs_properties = { <EOL> '<STR_LIT>' : '<STR_LIT:true>' } <EOL> self . _assert_bootstrap_actions ( <EOL> emrfs_option_value , expected_emrfs_ba_key_values , <EOL> expected_emrfs_properties ) <EOL> emrfs_option_value = '<STR_LIT>' <EOL> expected_emrfs_ba_key_values = [ <EOL> '<STR_LIT>' <EOL> ] <EOL> expected_emrfs_properties = { <EOL> '<STR_LIT>' : '<STR_LIT:true>' } <EOL> self . _assert_bootstrap_actions ( <EOL> emrfs_option_value , expected_emrfs_ba_key_values , <EOL> expected_emrfs_properties ) <EOL> def test_cse_kms ( self ) : <EOL> emrfs_option_value = '<STR_LIT>' '<STR_LIT>' <EOL> expected_emrfs_ba_key_values = [ <EOL> '<STR_LIT>' , '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> ] <EOL> expected_emrfs_properties = { <EOL> '<STR_LIT>' : '<STR_LIT:true>' , <EOL> '<STR_LIT>' : <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> self . _assert_bootstrap_actions ( <EOL> emrfs_option_value , expected_emrfs_ba_key_values , <EOL> expected_emrfs_properties ) <EOL> def test_cse_custom ( self ) : <EOL> emrfs_option_value = '<STR_LIT>' '<STR_LIT>' <EOL> expected_emrfs_ba_key_values = [ <EOL> '<STR_LIT>' , '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> ] <EOL> expected_emrfs_properties = { <EOL> '<STR_LIT>' : '<STR_LIT:true>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> self . _assert_bootstrap_actions ( <EOL> emrfs_option_value , expected_emrfs_ba_key_values , <EOL> expected_emrfs_properties , '<STR_LIT>' ) <EOL> def test_sse_and_consistent ( self ) : <EOL> emrfs_option_value = '<STR_LIT>' <EOL> expected_emrfs_ba_key_values = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> expected_emrfs_properties = { <EOL> '<STR_LIT>' : '<STR_LIT:true>' , <EOL> '<STR_LIT>' : '<STR_LIT:true>' } <EOL> self . _assert_bootstrap_actions ( <EOL> emrfs_option_value , expected_emrfs_ba_key_values , <EOL> expected_emrfs_properties ) <EOL> emrfs_option_value = '<STR_LIT>' <EOL> expected_emrfs_ba_key_values = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> expected_emrfs_properties = { <EOL> '<STR_LIT>' : '<STR_LIT:false>' , <EOL> '<STR_LIT>' : '<STR_LIT:true>' } <EOL> self . _assert_bootstrap_actions ( <EOL> emrfs_option_value , expected_emrfs_ba_key_values , <EOL> expected_emrfs_properties ) <EOL> def test_cse_and_consistent ( self ) : <EOL> emrfs_option_value = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> expected_emrfs_ba_key_values = [ <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> expected_emrfs_properties = { <EOL> '<STR_LIT>' : '<STR_LIT:true>' , <EOL> '<STR_LIT>' : '<STR_LIT:true>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> self . _assert_bootstrap_actions ( <EOL> emrfs_option_value , expected_emrfs_ba_key_values , <EOL> expected_emrfs_properties ) <EOL> def test_args_and_sse ( self ) : <EOL> emrfs_option_value = '<STR_LIT>' <EOL> expected_emrfs_ba_key_values = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> expected_emrfs_properties = { <EOL> '<STR_LIT>' : '<STR_LIT:true>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> self . _assert_bootstrap_actions ( <EOL> emrfs_option_value , expected_emrfs_ba_key_values , <EOL> expected_emrfs_properties ) <EOL> def test_args_and_cse ( self ) : <EOL> emrfs_option_value = ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> expected_emrfs_ba_key_values = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] <EOL> expected_emrfs_properties = { <EOL> '<STR_LIT>' : '<STR_LIT:true>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> self . _assert_bootstrap_actions ( <EOL> emrfs_option_value , expected_emrfs_ba_key_values , <EOL> expected_emrfs_properties ) <EOL> def test_args_and_consistent ( self ) : <EOL> emrfs_option_value = '<STR_LIT>' <EOL> expected_emrfs_ba_key_values = [ '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ] <EOL> expected_emrfs_properties = { <EOL> '<STR_LIT>' : '<STR_LIT:true>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> self . _assert_bootstrap_actions ( <EOL> emrfs_option_value , expected_emrfs_ba_key_values , <EOL> expected_emrfs_properties ) <EOL> def test_only_args ( self ) : <EOL> emrfs_option_value = '<STR_LIT>' <EOL> expected_emrfs_ba_key_values = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> expected_emrfs_properties = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> self . _assert_bootstrap_actions ( <EOL> emrfs_option_value , expected_emrfs_ba_key_values , <EOL> expected_emrfs_properties ) <EOL> def test_using_json_file ( self ) : <EOL> data_path = os . path . join ( <EOL> os . path . dirname ( __file__ ) , '<STR_LIT>' ) <EOL> emrfs_option_value = '<STR_LIT>' % data_path <EOL> expected_emrfs_ba_key_values = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] <EOL> expected_emrfs_properties = { <EOL> '<STR_LIT>' : '<STR_LIT:true>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:3>' , <EOL> '<STR_LIT>' : '<STR_LIT:false>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> self . _assert_bootstrap_actions ( <EOL> emrfs_option_value , expected_emrfs_ba_key_values , <EOL> expected_emrfs_properties ) <EOL> def test_only_one_encryption_type ( self ) : <EOL> self . _assert_error_msg ( <EOL> emrfs_option_value = '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> exception_class_name = '<STR_LIT>' , <EOL> error_msg_kwargs = { '<STR_LIT>' : '<STR_LIT:True>' , '<STR_LIT>' : '<STR_LIT>' } <EOL> ) <EOL> def test_cse_missing_provider_type ( self ) : <EOL> self . _assert_error_msg ( <EOL> emrfs_option_value = '<STR_LIT>' , <EOL> exception_class_name = '<STR_LIT>' , <EOL> error_msg_kwargs = { '<STR_LIT:object_name>' : CSE_OPTION_NAME , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> ) <EOL> def test_cse_kms_missing_key_id ( self ) : <EOL> self . _assert_error_msg ( <EOL> emrfs_option_value = '<STR_LIT>' , <EOL> exception_class_name = '<STR_LIT>' , <EOL> error_msg_kwargs = { '<STR_LIT:object_name>' : CSE_KMS_OPTION_NAME , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> ) <EOL> def test_cse_custom_missing_all ( self ) : <EOL> self . _assert_error_msg ( <EOL> emrfs_option_value = '<STR_LIT>' , <EOL> exception_class_name = '<STR_LIT>' , <EOL> error_msg_kwargs = { '<STR_LIT:object_name>' : CSE_CUSTOM_OPTION_NAME , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> '<STR_LIT>' } <EOL> ) <EOL> def test_cse_custom_missing_provider_class ( self ) : <EOL> self . _assert_error_msg ( <EOL> emrfs_option_value = '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> exception_class_name = '<STR_LIT>' , <EOL> error_msg_kwargs = { '<STR_LIT:object_name>' : CSE_CUSTOM_OPTION_NAME , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> ) <EOL> def test_cse_custom_missing_provider_location ( self ) : <EOL> self . _assert_error_msg ( <EOL> emrfs_option_value = '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> exception_class_name = '<STR_LIT>' , <EOL> error_msg_kwargs = { '<STR_LIT:object_name>' : CSE_CUSTOM_OPTION_NAME , <EOL> '<STR_LIT>' : '<STR_LIT>' } <EOL> ) <EOL> def test_valid_encryption ( self ) : <EOL> self . _assert_error_msg ( <EOL> emrfs_option_value = '<STR_LIT>' , <EOL> exception_class_name = '<STR_LIT>' , <EOL> error_msg_kwargs = { '<STR_LIT>' : '<STR_LIT>' } <EOL> ) <EOL> def test_valid_cse_provider_type ( self ) : <EOL> self . _assert_error_msg ( <EOL> emrfs_option_value = '<STR_LIT>' , <EOL> exception_class_name = '<STR_LIT>' , <EOL> error_msg_kwargs = { '<STR_LIT>' : '<STR_LIT>' } <EOL> ) <EOL> def test_valid_consistent_args ( self ) : <EOL> self . _assert_error_msg ( <EOL> emrfs_option_value = '<STR_LIT>' , <EOL> exception_class_name = '<STR_LIT>' , <EOL> error_msg_kwargs = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : CONSISTENT_OPTION_NAME } <EOL> ) <EOL> def test_valid_cse_kms_args ( self ) : <EOL> self . _assert_error_msg ( <EOL> emrfs_option_value = '<STR_LIT>' , <EOL> exception_class_name = '<STR_LIT>' , <EOL> error_msg_kwargs = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : CSE_KMS_OPTION_NAME } <EOL> ) <EOL> def test_valid_cse_custom_args ( self ) : <EOL> self . _assert_error_msg ( <EOL> emrfs_option_value = '<STR_LIT>' , <EOL> exception_class_name = '<STR_LIT>' , <EOL> error_msg_kwargs = { '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : CSE_CUSTOM_OPTION_NAME } <EOL> ) <EOL> def test_configurations_and_emrfs ( self ) : <EOL> emrfs_option_value = '<STR_LIT>' <EOL> configurations = json . dumps ( DEFAULT_CONFIGURATIONS , <EOL> separators = ( '<STR_LIT:U+002C>' , '<STR_LIT::>' ) ) <EOL> cmd = "<STR_LIT>" % ( DEFAULT_CMD , emrfs_option_value , configurations ) <EOL> expected_emrfs_properties = { '<STR_LIT>' : '<STR_LIT>' } <EOL> emrfs_configuration = copy . deepcopy ( EMPTY_EMRFS_CONFIGURATION ) <EOL> configurations = copy . deepcopy ( DEFAULT_CONFIGURATIONS ) <EOL> result = copy . deepcopy ( DEFAULT_RESULT ) <EOL> emrfs_configuration [ '<STR_LIT>' ] = expected_emrfs_properties <EOL> configurations . append ( emrfs_configuration ) <EOL> result [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> result [ '<STR_LIT>' ] = configurations <EOL> self . assert_params_for_cmd ( cmd , result ) <EOL> def test_duplicate_emrfs_configuration_exception ( self ) : <EOL> emrfs_option_value = '<STR_LIT>' <EOL> expected_emrfs_properties = { '<STR_LIT>' : '<STR_LIT>' } <EOL> emrfs_configuration = copy . deepcopy ( EMPTY_EMRFS_CONFIGURATION ) <EOL> emrfs_configuration [ '<STR_LIT>' ] = expected_emrfs_properties <EOL> configurations = copy . deepcopy ( DEFAULT_CONFIGURATIONS ) <EOL> configurations . append ( EMPTY_EMRFS_CONFIGURATION ) <EOL> configurations_json = json . dumps ( configurations , <EOL> separators = ( '<STR_LIT:U+002C>' , '<STR_LIT::>' ) ) <EOL> cmd = "<STR_LIT>" % ( DEFAULT_CMD , emrfs_option_value , configurations_json ) <EOL> result = copy . deepcopy ( DEFAULT_RESULT ) <EOL> result [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> result [ '<STR_LIT>' ] = configurations <EOL> self . assert_error_msg ( cmd , '<STR_LIT>' ) <EOL> def _assert_error_msg ( self , emrfs_option_value , <EOL> exception_class_name , error_msg_kwargs ) : <EOL> cmd = "<STR_LIT>" % ( DEFAULT_CMD , emrfs_option_value ) <EOL> self . assert_error_msg ( <EOL> cmd , <EOL> exception_class_name = exception_class_name , <EOL> error_msg_kwargs = error_msg_kwargs ) <EOL> cmd = "<STR_LIT>" % ( DEFAULT_CMD , emrfs_option_value ) <EOL> self . assert_error_msg ( <EOL> cmd , <EOL> exception_class_name = exception_class_name , <EOL> error_msg_kwargs = error_msg_kwargs ) <EOL> def _assert_bootstrap_actions ( self , emrfs_option_value , <EOL> expected_emrfs_ba_key_values , <EOL> expected_emrfs_properties , <EOL> provider_location = None ) : <EOL> if expected_emrfs_ba_key_values is not None : <EOL> cmd = "<STR_LIT>" % ( DEFAULT_CMD , emrfs_option_value ) <EOL> result = copy . deepcopy ( DEFAULT_RESULT ) <EOL> result [ '<STR_LIT>' ] = [ self . _create_s3_get_ba_config ( <EOL> provider_location ) ] if provider_location is not None else [ ] <EOL> result [ '<STR_LIT>' ] += [ self . _create_setup_emrfs_ba_config ( <EOL> expected_emrfs_ba_key_values ) ] <EOL> result [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . assert_params_for_cmd ( cmd , result ) <EOL> if expected_emrfs_properties is not None : <EOL> cmd = "<STR_LIT>" % ( DEFAULT_CMD , emrfs_option_value ) <EOL> result = copy . deepcopy ( DEFAULT_RESULT ) <EOL> emrfs_configuration = copy . deepcopy ( EMPTY_EMRFS_CONFIGURATION ) <EOL> emrfs_configuration [ '<STR_LIT>' ] = expected_emrfs_properties <EOL> result [ '<STR_LIT>' ] = [ emrfs_configuration ] <EOL> result [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . assert_params_for_cmd ( cmd , result ) <EOL> def _create_setup_emrfs_ba_config ( self , ba_arg_values ) : <EOL> ba_arg_keys = [ '<STR_LIT>' for x in ba_arg_values ] <EOL> ba_args = [ x for pair in zip ( ba_arg_keys , ba_arg_values ) for x in pair ] <EOL> return { <EOL> '<STR_LIT:Name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : ( '<STR_LIT>' <EOL> '<STR_LIT>' ) , <EOL> '<STR_LIT>' : ba_args <EOL> } <EOL> } <EOL> def _create_s3_get_ba_config ( self , provider_location ) : <EOL> return { <EOL> '<STR_LIT:Name>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : [ <EOL> '<STR_LIT>' , provider_location , <EOL> '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' <EOL> ] <EOL> } <EOL> } </s>
