from tests import unittest, temporary_file
import mock
import botocore.session
from botocore.exceptions import ProfileNotFound
class TestSession(unittest.TestCase):
def setUp(self):
self.environ = {}
self.env_patch = mock.patch('os.environ', self.environ)
self.env_patch.start()
self.session = botocore.session.get_session()
def tearDown(self):
self.env_patch.stop()
def test_profile_precedence(self):
self.environ['AWS_PROFILE'] = 'from_env_var'
self.session.set_config_variable('profile', 'from_session_instance')
self.assertEqual(self.session.profile, 'from_session_instance')
def test_credentials_with_profile_precedence(self):
self.environ['AWS_PROFILE'] = 'from_env_var'
self.session.set_config_variable('profile', 'from_session_instance')
try:
creds = self.session.get_credentials()
except ProfileNotFound as e:
self.assertNotIn('from_env_var', str(e))
self.assertIn('from_session_instance', str(e))
def test_session_profile_overrides_env_vars(self):
with temporary_file('w') as f:
self.environ['AWS_ACCESS_KEY_ID'] = 'env_var_akid'
self.environ['AWS_SECRET_ACCESS_KEY'] = 'env_var_sak'
self.environ['AWS_SHARED_CREDENTIALS_FILE'] = f.name
f.write(
'[from_session_instance]\n'
'aws_access_key_id=shared_creds_akid\n'
'aws_secret_access_key=shared_creds_sak\n'
)
f.flush()
self.session.set_config_variable('profile', 'from_session_instance')
creds = self.session.get_credentials()
self.assertEqual(creds.access_key, 'shared_creds_akid')
self.assertEqual(creds.secret_key, 'shared_creds_sak')
def test_profile_does_not_win_if_all_from_env_vars(self):
with temporary_file('w') as f:
self.environ['AWS_SHARED_CREDENTIALS_FILE'] = f.name
self.environ['AWS_PROFILE'] = 'myprofile'
f.write(
'[myprofile]\n'
)
f.flush()
self.environ['AWS_ACCESS_KEY_ID'] = 'env_var_akid'
self.environ['AWS_SECRET_ACCESS_KEY'] = 'env_var_sak'
creds = self.session.get_credentials()
self.assertEqual(creds.access_key, 'env_var_akid')
self.assertEqual(creds.secret_key, 'env_var_sak')
def test_provides_available_regions_for_same_endpoint_prefix(self):
regions = self.session.get_available_regions('s3')
self.assertTrue(regions)
def test_provides_available_regions_for_different_endpoint_prefix(self):
regions = self.session.get_available_regions('elb')
self.assertTrue(regions)
def test_does_not_provide_regions_for_mismatch_service_name(self):
regions = self.session.get_available_regions('elasticloadbalancing')
self.assertEqual(regions, [])