<s> from datetime import datetime , timedelta <EOL> from tests . compat import mock , unittest <EOL> import os <EOL> from boto import provider <EOL> from boto . compat import expanduser <EOL> from boto . exception import InvalidInstanceMetadataError <EOL> INSTANCE_CONFIG = { <EOL> '<STR_LIT>' : { <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' <EOL> } <EOL> } <EOL> class TestProvider ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> self . environ = { } <EOL> self . config = { } <EOL> self . shared_config = { } <EOL> self . metadata_patch = mock . patch ( '<STR_LIT>' ) <EOL> self . config_patch = mock . patch ( '<STR_LIT>' , <EOL> self . get_config ) <EOL> self . has_config_patch = mock . patch ( '<STR_LIT>' , <EOL> self . has_config ) <EOL> self . config_object_patch = mock . patch . object ( <EOL> provider . Config , '<STR_LIT>' , self . get_shared_config ) <EOL> self . has_config_object_patch = mock . patch . object ( <EOL> provider . Config , '<STR_LIT>' , self . has_shared_config ) <EOL> self . environ_patch = mock . patch ( '<STR_LIT>' , self . environ ) <EOL> self . get_instance_metadata = self . metadata_patch . start ( ) <EOL> self . get_instance_metadata . return_value = None <EOL> self . config_patch . start ( ) <EOL> self . has_config_patch . start ( ) <EOL> self . config_object_patch . start ( ) <EOL> self . has_config_object_patch . start ( ) <EOL> self . environ_patch . start ( ) <EOL> def tearDown ( self ) : <EOL> self . metadata_patch . stop ( ) <EOL> self . config_patch . stop ( ) <EOL> self . has_config_patch . stop ( ) <EOL> self . config_object_patch . stop ( ) <EOL> self . has_config_object_patch . stop ( ) <EOL> self . environ_patch . stop ( ) <EOL> def has_config ( self , section_name , key ) : <EOL> try : <EOL> self . config [ section_name ] [ key ] <EOL> return True <EOL> except KeyError : <EOL> return False <EOL> def get_config ( self , section_name , key ) : <EOL> try : <EOL> return self . config [ section_name ] [ key ] <EOL> except KeyError : <EOL> return None <EOL> def has_shared_config ( self , section_name , key ) : <EOL> try : <EOL> self . shared_config [ section_name ] [ key ] <EOL> return True <EOL> except KeyError : <EOL> return False <EOL> def get_shared_config ( self , section_name , key ) : <EOL> try : <EOL> return self . shared_config [ section_name ] [ key ] <EOL> except KeyError : <EOL> return None <EOL> def test_passed_in_values_are_used ( self ) : <EOL> p = provider . Provider ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( p . access_key , '<STR_LIT>' ) <EOL> self . assertEqual ( p . secret_key , '<STR_LIT>' ) <EOL> self . assertEqual ( p . security_token , '<STR_LIT>' ) <EOL> def test_environment_variables_are_used ( self ) : <EOL> self . environ [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . environ [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> p = provider . Provider ( '<STR_LIT>' ) <EOL> self . assertEqual ( p . access_key , '<STR_LIT>' ) <EOL> self . assertEqual ( p . secret_key , '<STR_LIT>' ) <EOL> self . assertIsNone ( p . security_token ) <EOL> def test_environment_variable_aws_security_token ( self ) : <EOL> self . environ [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . environ [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . environ [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> p = provider . Provider ( '<STR_LIT>' ) <EOL> self . assertEqual ( p . access_key , '<STR_LIT>' ) <EOL> self . assertEqual ( p . secret_key , '<STR_LIT>' ) <EOL> self . assertEqual ( p . security_token , '<STR_LIT>' ) <EOL> def test_no_credentials_provided ( self ) : <EOL> p = provider . Provider ( <EOL> '<STR_LIT>' , <EOL> provider . NO_CREDENTIALS_PROVIDED , <EOL> provider . NO_CREDENTIALS_PROVIDED , <EOL> provider . NO_CREDENTIALS_PROVIDED <EOL> ) <EOL> self . assertEqual ( p . access_key , provider . NO_CREDENTIALS_PROVIDED ) <EOL> self . assertEqual ( p . secret_key , provider . NO_CREDENTIALS_PROVIDED ) <EOL> self . assertEqual ( p . security_token , provider . NO_CREDENTIALS_PROVIDED ) <EOL> def test_config_profile_values_are_used ( self ) : <EOL> self . config = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } , '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } , '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } , '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> } <EOL> p = provider . Provider ( '<STR_LIT>' , profile_name = '<STR_LIT>' ) <EOL> self . assertEqual ( p . access_key , '<STR_LIT>' ) <EOL> self . assertEqual ( p . secret_key , '<STR_LIT>' ) <EOL> p = provider . Provider ( '<STR_LIT>' , profile_name = '<STR_LIT>' ) <EOL> self . assertEqual ( p . access_key , '<STR_LIT>' ) <EOL> self . assertEqual ( p . secret_key , '<STR_LIT>' ) <EOL> self . assertEqual ( p . security_token , '<STR_LIT>' ) <EOL> q = provider . Provider ( '<STR_LIT>' , profile_name = '<STR_LIT>' ) <EOL> self . assertEqual ( q . access_key , '<STR_LIT>' ) <EOL> self . assertEqual ( q . secret_key , '<STR_LIT>' ) <EOL> def test_config_missing_profile ( self ) : <EOL> self . shared_config = { <EOL> '<STR_LIT:default>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> } <EOL> self . config = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> } <EOL> with self . assertRaises ( provider . ProfileNotFoundError ) : <EOL> provider . Provider ( '<STR_LIT>' , profile_name = '<STR_LIT>' ) <EOL> def test_config_values_are_used ( self ) : <EOL> self . config = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> } <EOL> p = provider . Provider ( '<STR_LIT>' ) <EOL> self . assertEqual ( p . access_key , '<STR_LIT>' ) <EOL> self . assertEqual ( p . secret_key , '<STR_LIT>' ) <EOL> self . assertIsNone ( p . security_token ) <EOL> def test_config_value_security_token_is_used ( self ) : <EOL> self . config = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> } <EOL> p = provider . Provider ( '<STR_LIT>' ) <EOL> self . assertEqual ( p . access_key , '<STR_LIT>' ) <EOL> self . assertEqual ( p . secret_key , '<STR_LIT>' ) <EOL> self . assertEqual ( p . security_token , '<STR_LIT>' ) <EOL> def test_keyring_is_used ( self ) : <EOL> self . config = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:test>' , <EOL> } <EOL> } <EOL> import sys <EOL> try : <EOL> import keyring <EOL> imported = True <EOL> except ImportError : <EOL> sys . modules [ '<STR_LIT>' ] = keyring = type ( mock ) ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> imported = False <EOL> try : <EOL> with mock . patch ( '<STR_LIT>' , create = True ) : <EOL> keyring . get_password . side_effect = ( <EOL> lambda kr , login : kr + login + '<STR_LIT>' ) <EOL> p = provider . Provider ( '<STR_LIT>' ) <EOL> self . assertEqual ( p . access_key , '<STR_LIT>' ) <EOL> self . assertEqual ( p . secret_key , '<STR_LIT>' ) <EOL> self . assertIsNone ( p . security_token ) <EOL> finally : <EOL> if not imported : <EOL> del sys . modules [ '<STR_LIT>' ] <EOL> def test_passed_in_values_beat_env_vars ( self ) : <EOL> self . environ [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . environ [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . environ [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> p = provider . Provider ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( p . access_key , '<STR_LIT>' ) <EOL> self . assertEqual ( p . secret_key , '<STR_LIT>' ) <EOL> self . assertEqual ( p . security_token , None ) <EOL> def test_env_vars_beat_shared_creds_values ( self ) : <EOL> self . environ [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . environ [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . shared_config = { <EOL> '<STR_LIT:default>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> } <EOL> p = provider . Provider ( '<STR_LIT>' ) <EOL> self . assertEqual ( p . access_key , '<STR_LIT>' ) <EOL> self . assertEqual ( p . secret_key , '<STR_LIT>' ) <EOL> self . assertIsNone ( p . security_token ) <EOL> def test_shared_creds_beat_config_values ( self ) : <EOL> self . shared_config = { <EOL> '<STR_LIT:default>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> } <EOL> self . config = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> } <EOL> p = provider . Provider ( '<STR_LIT>' ) <EOL> self . assertEqual ( p . access_key , '<STR_LIT>' ) <EOL> self . assertEqual ( p . secret_key , '<STR_LIT>' ) <EOL> self . assertIsNone ( p . security_token ) <EOL> def test_shared_creds_profile_beats_defaults ( self ) : <EOL> self . shared_config = { <EOL> '<STR_LIT:default>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } , <EOL> '<STR_LIT:foo>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> } <EOL> p = provider . Provider ( '<STR_LIT>' , profile_name = '<STR_LIT:foo>' ) <EOL> self . assertEqual ( p . access_key , '<STR_LIT>' ) <EOL> self . assertEqual ( p . secret_key , '<STR_LIT>' ) <EOL> self . assertIsNone ( p . security_token ) <EOL> def test_env_profile_loads_profile ( self ) : <EOL> self . environ [ '<STR_LIT>' ] = '<STR_LIT:foo>' <EOL> self . shared_config = { <EOL> '<STR_LIT:default>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } , <EOL> '<STR_LIT:foo>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> } <EOL> self . config = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } , <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> } <EOL> p = provider . Provider ( '<STR_LIT>' ) <EOL> self . assertEqual ( p . access_key , '<STR_LIT>' ) <EOL> self . assertEqual ( p . secret_key , '<STR_LIT>' ) <EOL> self . assertIsNone ( p . security_token ) <EOL> self . shared_config = { } <EOL> p = provider . Provider ( '<STR_LIT>' ) <EOL> self . assertEqual ( p . access_key , '<STR_LIT>' ) <EOL> self . assertEqual ( p . secret_key , '<STR_LIT>' ) <EOL> self . assertIsNone ( p . security_token ) <EOL> def test_env_vars_security_token_beats_config_values ( self ) : <EOL> self . environ [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . environ [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . environ [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . shared_config = { <EOL> '<STR_LIT:default>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> } <EOL> self . config = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> } <EOL> p = provider . Provider ( '<STR_LIT>' ) <EOL> self . assertEqual ( p . access_key , '<STR_LIT>' ) <EOL> self . assertEqual ( p . secret_key , '<STR_LIT>' ) <EOL> self . assertEqual ( p . security_token , '<STR_LIT>' ) <EOL> self . environ . clear ( ) <EOL> p = provider . Provider ( '<STR_LIT>' ) <EOL> self . assertEqual ( p . security_token , '<STR_LIT>' ) <EOL> self . shared_config . clear ( ) <EOL> p = provider . Provider ( '<STR_LIT>' ) <EOL> self . assertEqual ( p . security_token , '<STR_LIT>' ) <EOL> def test_metadata_server_credentials ( self ) : <EOL> self . get_instance_metadata . return_value = INSTANCE_CONFIG <EOL> p = provider . Provider ( '<STR_LIT>' ) <EOL> self . assertEqual ( p . access_key , '<STR_LIT>' ) <EOL> self . assertEqual ( p . secret_key , '<STR_LIT>' ) <EOL> self . assertEqual ( p . security_token , '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> self . get_instance_metadata . call_args [ <NUM_LIT:1> ] [ '<STR_LIT:data>' ] , <EOL> '<STR_LIT>' ) <EOL> def test_metadata_server_returns_bad_type ( self ) : <EOL> self . get_instance_metadata . return_value = { <EOL> '<STR_LIT>' : [ ] , <EOL> } <EOL> with self . assertRaises ( InvalidInstanceMetadataError ) : <EOL> p = provider . Provider ( '<STR_LIT>' ) <EOL> def test_metadata_server_returns_empty_string ( self ) : <EOL> self . get_instance_metadata . return_value = { <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> with self . assertRaises ( InvalidInstanceMetadataError ) : <EOL> p = provider . Provider ( '<STR_LIT>' ) <EOL> def test_metadata_server_returns_missing_keys ( self ) : <EOL> self . get_instance_metadata . return_value = { <EOL> '<STR_LIT>' : { <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> } <EOL> } <EOL> with self . assertRaises ( InvalidInstanceMetadataError ) : <EOL> p = provider . Provider ( '<STR_LIT>' ) <EOL> def test_refresh_credentials ( self ) : <EOL> now = datetime . utcnow ( ) <EOL> first_expiration = ( now + timedelta ( seconds = <NUM_LIT:10> ) ) . strftime ( <EOL> "<STR_LIT>" ) <EOL> credentials = { <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : first_expiration , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' , <EOL> u'<STR_LIT>' : u'<STR_LIT>' <EOL> } <EOL> instance_config = { '<STR_LIT>' : credentials } <EOL> self . get_instance_metadata . return_value = instance_config <EOL> p = provider . Provider ( '<STR_LIT>' ) <EOL> self . assertEqual ( p . access_key , '<STR_LIT>' ) <EOL> self . assertEqual ( p . secret_key , '<STR_LIT>' ) <EOL> self . assertEqual ( p . security_token , '<STR_LIT>' ) <EOL> self . assertIsNotNone ( p . _credential_expiry_time ) <EOL> expired = now - timedelta ( seconds = <NUM_LIT:20> ) <EOL> p . _credential_expiry_time = expired <EOL> credentials [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> credentials [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> credentials [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . get_instance_metadata . return_value = instance_config <EOL> self . assertEqual ( p . access_key , '<STR_LIT>' ) <EOL> self . assertEqual ( p . secret_key , '<STR_LIT>' ) <EOL> self . assertEqual ( p . security_token , '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' ) <EOL> def test_metadata_config_params ( self , config_float , config_int ) : <EOL> config_int . return_value = <NUM_LIT:10> <EOL> config_float . return_value = <NUM_LIT> <EOL> self . get_instance_metadata . return_value = INSTANCE_CONFIG <EOL> p = provider . Provider ( '<STR_LIT>' ) <EOL> self . assertEqual ( p . access_key , '<STR_LIT>' ) <EOL> self . assertEqual ( p . secret_key , '<STR_LIT>' ) <EOL> self . assertEqual ( p . security_token , '<STR_LIT>' ) <EOL> self . get_instance_metadata . assert_called_with ( <EOL> timeout = <NUM_LIT> , num_retries = <NUM_LIT:10> , <EOL> data = '<STR_LIT>' ) <EOL> def test_provider_google ( self ) : <EOL> self . environ [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . environ [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . shared_config = { <EOL> '<STR_LIT:default>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> } <EOL> self . config = { <EOL> '<STR_LIT>' : { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> } <EOL> p = provider . Provider ( '<STR_LIT>' ) <EOL> self . assertEqual ( p . access_key , '<STR_LIT>' ) <EOL> self . assertEqual ( p . secret_key , '<STR_LIT>' ) <EOL> self . environ . clear ( ) <EOL> p = provider . Provider ( '<STR_LIT>' ) <EOL> self . assertEqual ( p . access_key , '<STR_LIT>' ) <EOL> self . assertEqual ( p . secret_key , '<STR_LIT>' ) <EOL> self . shared_config . clear ( ) <EOL> p = provider . Provider ( '<STR_LIT>' ) <EOL> self . assertEqual ( p . access_key , '<STR_LIT>' ) <EOL> self . assertEqual ( p . secret_key , '<STR_LIT>' ) <EOL> @ mock . patch ( '<STR_LIT>' , return_value = True ) <EOL> @ mock . patch . object ( provider . Config , '<STR_LIT>' ) <EOL> def test_shared_config_loading ( self , load_from_path , exists ) : <EOL> provider . Provider ( '<STR_LIT>' ) <EOL> path = os . path . join ( expanduser ( '<STR_LIT>' ) , '<STR_LIT>' , '<STR_LIT>' ) <EOL> exists . assert_called_once_with ( path ) <EOL> load_from_path . assert_called_once_with ( path ) <EOL> exists . reset_mock ( ) <EOL> load_from_path . reset_mock ( ) <EOL> provider . Provider ( '<STR_LIT>' ) <EOL> path = os . path . join ( expanduser ( '<STR_LIT>' ) , '<STR_LIT>' , '<STR_LIT>' ) <EOL> exists . assert_called_once_with ( path ) <EOL> load_from_path . assert_called_once_with ( path ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> unittest . main ( ) </s>
