1032 lines
41 KiB
Python
1032 lines
41 KiB
Python
#!/usr/bin/env
|
|
# Copyright (c) 2012-2013 Mitch Garnaat http://garnaat.org/
|
|
# Copyright 2012-2014 Amazon.com, Inc. or its affiliates. All Rights Reserved.
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License"). You
|
|
# may not use this file except in compliance with the License. A copy of
|
|
# the License is located at
|
|
#
|
|
# http://aws.amazon.com/apache2.0/
|
|
#
|
|
# or in the "license" file accompanying this file. This file is
|
|
# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
|
|
# ANY KIND, either express or implied. See the License for the specific
|
|
# language governing permissions and limitations under the License.
|
|
import logging
|
|
import os
|
|
import shutil
|
|
import tempfile
|
|
|
|
import pytest
|
|
|
|
import botocore.config
|
|
import botocore.exceptions
|
|
import botocore.loaders
|
|
import botocore.session
|
|
from botocore import (
|
|
UNSIGNED,
|
|
client,
|
|
register_initializer,
|
|
unregister_initializer,
|
|
)
|
|
from botocore.configprovider import ConfigChainFactory
|
|
from botocore.hooks import HierarchicalEmitter
|
|
from botocore.model import ServiceModel
|
|
from botocore.paginate import PaginatorModel
|
|
from botocore.waiter import WaiterModel
|
|
from tests import create_session, mock, requires_crt, temporary_file, unittest
|
|
|
|
|
|
class BaseSessionTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.environ = {}
|
|
self.environ_patch = mock.patch('os.environ', self.environ)
|
|
self.environ_patch.start()
|
|
self.environ['FOO_PROFILE'] = 'foo'
|
|
self.environ['FOO_REGION'] = 'us-west-11'
|
|
data_path = os.path.join(os.path.dirname(__file__), 'data')
|
|
self.environ['FOO_DATA_PATH'] = data_path
|
|
config_path = os.path.join(
|
|
os.path.dirname(__file__), 'cfg', 'foo_config'
|
|
)
|
|
self.environ['FOO_CONFIG_FILE'] = config_path
|
|
self.session = create_session()
|
|
config_chain_builder = ConfigChainFactory(
|
|
session=self.session,
|
|
environ=self.environ,
|
|
)
|
|
config_store = self.session.get_component('config_store')
|
|
config_updates = {
|
|
'profile': config_chain_builder.create_config_chain(
|
|
instance_name='profile',
|
|
env_var_names='FOO_PROFILE',
|
|
),
|
|
'region': config_chain_builder.create_config_chain(
|
|
instance_name='region',
|
|
env_var_names='FOO_REGION',
|
|
config_property_names='foo_region',
|
|
),
|
|
'data_path': config_chain_builder.create_config_chain(
|
|
instance_name='data_path',
|
|
env_var_names='FOO_DATA_PATH',
|
|
config_property_names='data_path',
|
|
),
|
|
'config_file': config_chain_builder.create_config_chain(
|
|
instance_name='config_file',
|
|
env_var_names='FOO_CONFIG_FILE',
|
|
),
|
|
'credentials_file': config_chain_builder.create_config_chain(
|
|
instance_name='credentials_file',
|
|
default='/tmp/nowhere',
|
|
),
|
|
'ca_bundle': config_chain_builder.create_config_chain(
|
|
instance_name='ca_bundle',
|
|
env_var_names='FOO_AWS_CA_BUNDLE',
|
|
config_property_names='foo_ca_bundle',
|
|
),
|
|
'api_versions': config_chain_builder.create_config_chain(
|
|
instance_name='api_versions',
|
|
config_property_names='foo_api_versions',
|
|
default={},
|
|
),
|
|
}
|
|
for name, provider in config_updates.items():
|
|
config_store.set_config_provider(name, provider)
|
|
|
|
def update_session_config_mapping(self, logical_name, **kwargs):
|
|
config_chain_builder = ConfigChainFactory(
|
|
session=self.session,
|
|
environ=self.environ,
|
|
)
|
|
self.session.get_component('config_store').set_config_provider(
|
|
logical_name,
|
|
config_chain_builder.create_config_chain(**kwargs),
|
|
)
|
|
|
|
def tearDown(self):
|
|
self.environ_patch.stop()
|
|
|
|
|
|
class SessionTest(BaseSessionTest):
|
|
def close_log_file_handler(self, tempdir, filename):
|
|
logger = logging.getLogger('botocore')
|
|
handlers = logger.handlers
|
|
for handler in handlers[:]:
|
|
if hasattr(handler, 'stream') and handler.stream.name == filename:
|
|
handler.stream.close()
|
|
logger.removeHandler(handler)
|
|
os.remove(filename)
|
|
# logging has an atexit handler that will try to flush/close
|
|
# the file. By setting this flag to False, we'll prevent it
|
|
# from raising an exception, which is fine because we're
|
|
# handling the closing of the file ourself.
|
|
logging.raiseExceptions = False
|
|
shutil.rmtree(tempdir)
|
|
|
|
def test_supports_multiple_env_vars_for_single_logical_name(self):
|
|
self.update_session_config_mapping(
|
|
'profile', env_var_names=['BAR_DEFAULT_PROFILE', 'BAR_PROFILE']
|
|
)
|
|
self.environ['BAR_DEFAULT_PROFILE'] = 'first'
|
|
self.environ['BAR_PROFILE'] = 'second'
|
|
self.assertEqual(self.session.get_config_variable('profile'), 'first')
|
|
|
|
def test_profile_when_set_explicitly(self):
|
|
session = create_session(profile='asdf')
|
|
self.assertEqual(session.profile, 'asdf')
|
|
|
|
def test_profile_when_pulled_from_env(self):
|
|
self.environ['FOO_PROFILE'] = 'bar'
|
|
# Even though we didn't explicitly pass in a profile, the
|
|
# profile property will still look this up for us.
|
|
self.assertEqual(self.session.profile, 'bar')
|
|
|
|
def test_multiple_env_vars_uses_second_var(self):
|
|
self.update_session_config_mapping(
|
|
'profile', env_var_names=['BAR_DEFAULT_PROFILE', 'BAR_PROFILE']
|
|
)
|
|
self.environ.pop('BAR_DEFAULT_PROFILE', None)
|
|
self.environ['BAR_PROFILE'] = 'second'
|
|
self.assertEqual(self.session.get_config_variable('profile'), 'second')
|
|
|
|
def test_profile_does_not_exist_raises_exception(self):
|
|
# Given we have no profile:
|
|
self.environ['FOO_PROFILE'] = 'profile_that_does_not_exist'
|
|
with self.assertRaises(botocore.exceptions.ProfileNotFound):
|
|
self.session.get_scoped_config()
|
|
|
|
def test_variable_does_not_exist(self):
|
|
self.assertIsNone(self.session.get_config_variable('foo/bar'))
|
|
|
|
def test_get_aws_services_in_alphabetical_order(self):
|
|
services = self.session.get_available_services()
|
|
self.assertEqual(sorted(services), services)
|
|
|
|
def test_profile_does_not_exist_with_default_profile(self):
|
|
config = self.session.get_scoped_config()
|
|
# We should have loaded this properly, and we'll check
|
|
# that foo_access_key which is defined in the config
|
|
# file should be present in the loaded config dict.
|
|
self.assertIn('aws_access_key_id', config)
|
|
|
|
def test_type_conversions_occur_when_specified(self):
|
|
# Specify that we can retrieve the var from the
|
|
# FOO_TIMEOUT env var, with a conversion function
|
|
# of int().
|
|
self.update_session_config_mapping(
|
|
'metadata_service_timeout',
|
|
env_var_names='FOO_TIMEOUT',
|
|
conversion_func=int,
|
|
)
|
|
# Environment variables are always strings.
|
|
self.environ['FOO_TIMEOUT'] = '10'
|
|
# But we should type convert this to a string.
|
|
self.assertEqual(
|
|
self.session.get_config_variable('metadata_service_timeout'), 10
|
|
)
|
|
|
|
def test_default_profile_specified_raises_exception(self):
|
|
# If you explicity set the default profile and you don't
|
|
# have that in your config file, an exception is raised.
|
|
config_path = os.path.join(
|
|
os.path.dirname(__file__), 'cfg', 'boto_config_empty'
|
|
)
|
|
self.environ['FOO_CONFIG_FILE'] = config_path
|
|
self.environ['FOO_PROFILE'] = 'default'
|
|
# In this case, even though we specified default, because
|
|
# the boto_config_empty config file does not have a default
|
|
# profile, we should be raising an exception.
|
|
with self.assertRaises(botocore.exceptions.ProfileNotFound):
|
|
self.session.get_scoped_config()
|
|
|
|
def test_file_logger(self):
|
|
tempdir = tempfile.mkdtemp()
|
|
temp_file = os.path.join(tempdir, 'file_logger')
|
|
self.session.set_file_logger(logging.DEBUG, temp_file)
|
|
self.addCleanup(self.close_log_file_handler, tempdir, temp_file)
|
|
self.session.get_credentials()
|
|
self.assertTrue(os.path.isfile(temp_file))
|
|
with open(temp_file) as logfile:
|
|
s = logfile.read()
|
|
self.assertTrue('Looking for credentials' in s)
|
|
|
|
def test_full_config_property(self):
|
|
full_config = self.session.full_config
|
|
self.assertTrue('foo' in full_config['profiles'])
|
|
self.assertTrue('default' in full_config['profiles'])
|
|
|
|
def test_full_config_merges_creds_file_data(self):
|
|
with temporary_file('w') as f:
|
|
self.session.set_config_variable('credentials_file', f.name)
|
|
f.write('[newprofile]\n')
|
|
f.write('aws_access_key_id=FROM_CREDS_FILE_1\n')
|
|
f.write('aws_secret_access_key=FROM_CREDS_FILE_2\n')
|
|
f.flush()
|
|
|
|
full_config = self.session.full_config
|
|
self.assertEqual(
|
|
full_config['profiles']['newprofile'],
|
|
{
|
|
'aws_access_key_id': 'FROM_CREDS_FILE_1',
|
|
'aws_secret_access_key': 'FROM_CREDS_FILE_2',
|
|
},
|
|
)
|
|
|
|
def test_path_not_in_available_profiles(self):
|
|
with temporary_file('w') as f:
|
|
self.session.set_config_variable('credentials_file', f.name)
|
|
f.write('[newprofile]\n')
|
|
f.write('aws_access_key_id=FROM_CREDS_FILE_1\n')
|
|
f.write('aws_secret_access_key=FROM_CREDS_FILE_2\n')
|
|
f.flush()
|
|
|
|
profiles = self.session.available_profiles
|
|
self.assertEqual(set(profiles), {'foo', 'default', 'newprofile'})
|
|
|
|
def test_emit_delegates_to_emitter(self):
|
|
calls = []
|
|
handler = lambda **kwargs: calls.append(kwargs)
|
|
self.session.register('foo', handler)
|
|
self.session.emit('foo')
|
|
self.assertEqual(len(calls), 1)
|
|
self.assertEqual(calls[0]['event_name'], 'foo')
|
|
|
|
def test_emitter_can_be_passed_in(self):
|
|
events = HierarchicalEmitter()
|
|
session = create_session(event_hooks=events)
|
|
calls = []
|
|
handler = lambda **kwargs: calls.append(kwargs)
|
|
events.register('foo', handler)
|
|
|
|
session.emit('foo')
|
|
self.assertEqual(len(calls), 1)
|
|
|
|
def test_emit_first_non_none(self):
|
|
self.session.register('foo', lambda **kwargs: None)
|
|
self.session.register('foo', lambda **kwargs: 'first')
|
|
self.session.register('foo', lambda **kwargs: 'second')
|
|
response = self.session.emit_first_non_none_response('foo')
|
|
self.assertEqual(response, 'first')
|
|
|
|
@mock.patch('logging.getLogger')
|
|
@mock.patch('logging.FileHandler')
|
|
def test_logger_name_can_be_passed_in(self, file_handler, get_logger):
|
|
self.session.set_debug_logger('botocore.hooks')
|
|
get_logger.assert_called_with('botocore.hooks')
|
|
|
|
self.session.set_file_logger('DEBUG', 'debuglog', 'botocore.service')
|
|
get_logger.assert_called_with('botocore.service')
|
|
file_handler.assert_called_with('debuglog')
|
|
|
|
@mock.patch('logging.getLogger')
|
|
@mock.patch('logging.StreamHandler')
|
|
@mock.patch('logging.Formatter')
|
|
def test_general_purpose_logger(self, formatter, file_handler, get_logger):
|
|
self.session.set_stream_logger('foo.bar', 'ERROR', format_string='foo')
|
|
get_logger.assert_called_with('foo.bar')
|
|
get_logger.return_value.setLevel.assert_called_with(logging.DEBUG)
|
|
formatter.assert_called_with('foo')
|
|
|
|
def test_register_with_unique_id(self):
|
|
calls = []
|
|
handler = lambda **kwargs: calls.append(kwargs)
|
|
self.session.register('foo', handler, unique_id='bar')
|
|
self.session.emit('foo')
|
|
self.assertEqual(calls[0]['event_name'], 'foo')
|
|
calls = []
|
|
self.session.unregister('foo', unique_id='bar')
|
|
self.session.emit('foo')
|
|
self.assertEqual(calls, [])
|
|
|
|
|
|
class TestBuiltinEventHandlers(BaseSessionTest):
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.builtin_handlers = [
|
|
('foo', self.on_foo),
|
|
]
|
|
self.foo_called = False
|
|
self.handler_patch = mock.patch(
|
|
'botocore.handlers.BUILTIN_HANDLERS', self.builtin_handlers
|
|
)
|
|
self.handler_patch.start()
|
|
|
|
def on_foo(self, **kwargs):
|
|
self.foo_called = True
|
|
|
|
def tearDown(self):
|
|
super().tearDown()
|
|
self.handler_patch.stop()
|
|
|
|
def test_registered_builtin_handlers(self):
|
|
session = create_session(include_builtin_handlers=True)
|
|
session.emit('foo')
|
|
self.assertTrue(self.foo_called)
|
|
|
|
|
|
class TestSessionConfigurationVars(BaseSessionTest):
|
|
def test_per_session_config_vars(self):
|
|
self.update_session_config_mapping(
|
|
'foobar',
|
|
instance_name='foobar',
|
|
env_var_names='FOOBAR',
|
|
default='default',
|
|
)
|
|
# Default value.
|
|
self.assertEqual(self.session.get_config_variable('foobar'), 'default')
|
|
# Retrieve from os environment variable.
|
|
self.environ['FOOBAR'] = 'fromenv'
|
|
self.assertEqual(self.session.get_config_variable('foobar'), 'fromenv')
|
|
|
|
# Explicit override.
|
|
self.session.set_config_variable('foobar', 'session-instance')
|
|
self.assertEqual(
|
|
self.session.get_config_variable('foobar'), 'session-instance'
|
|
)
|
|
|
|
# Back to default value.
|
|
del self.environ['FOOBAR']
|
|
self.session.set_config_variable('foobar', None)
|
|
self.assertEqual(self.session.get_config_variable('foobar'), 'default')
|
|
|
|
def test_default_value_can_be_overriden(self):
|
|
self.update_session_config_mapping(
|
|
'foobar',
|
|
instance_name='foobar',
|
|
env_var_names='FOOBAR',
|
|
default='default',
|
|
)
|
|
self.assertEqual(self.session.get_config_variable('foobar'), 'default')
|
|
|
|
def test_can_get_with_methods(self):
|
|
self.environ['AWS_DEFAULT_REGION'] = 'env-var'
|
|
self.session.set_config_variable('region', 'instance-var')
|
|
value = self.session.get_config_variable('region')
|
|
self.assertEqual(value, 'instance-var')
|
|
|
|
value = self.session.get_config_variable('region', methods=('env',))
|
|
self.assertEqual(value, 'env-var')
|
|
|
|
|
|
class TestSessionPartitionFiles(BaseSessionTest):
|
|
def test_lists_partitions_on_disk(self):
|
|
mock_resolver = mock.Mock()
|
|
mock_resolver.get_available_partitions.return_value = ['foo']
|
|
self.session._register_internal_component(
|
|
'endpoint_resolver', mock_resolver
|
|
)
|
|
self.assertEqual(['foo'], self.session.get_available_partitions())
|
|
|
|
def test_proxies_list_endpoints_to_resolver(self):
|
|
resolver = mock.Mock()
|
|
resolver.get_available_endpoints.return_value = ['a', 'b']
|
|
self.session._register_internal_component(
|
|
'endpoint_resolver', resolver
|
|
)
|
|
self.session.get_available_regions('foo', 'bar', True)
|
|
|
|
def test_provides_empty_list_for_unknown_service_regions(self):
|
|
regions = self.session.get_available_regions('__foo__')
|
|
self.assertEqual([], regions)
|
|
|
|
def test_provides_correct_partition_for_region(self):
|
|
partition = self.session.get_partition_for_region('us-west-2')
|
|
self.assertEqual(partition, 'aws')
|
|
|
|
def test_provides_correct_partition_for_region_regex(self):
|
|
partition = self.session.get_partition_for_region('af-south-99')
|
|
self.assertEqual(partition, 'aws')
|
|
|
|
def test_provides_correct_partition_for_region_non_default(self):
|
|
partition = self.session.get_partition_for_region('cn-north-1')
|
|
self.assertEqual(partition, 'aws-cn')
|
|
|
|
def test_raises_exception_for_invalid_region(self):
|
|
with self.assertRaises(botocore.exceptions.UnknownRegionError):
|
|
self.session.get_partition_for_region('no-good-1')
|
|
|
|
|
|
class TestSessionUserAgent(BaseSessionTest):
|
|
def test_can_change_user_agent_name(self):
|
|
self.session.user_agent_name = 'something-else'
|
|
self.assertTrue(self.session.user_agent().startswith('something-else'))
|
|
|
|
def test_can_change_user_agent_version(self):
|
|
self.session.user_agent_version = '24.0'
|
|
self.assertTrue(self.session.user_agent().startswith('Botocore/24.0'))
|
|
|
|
def test_can_append_to_user_agent(self):
|
|
self.session.user_agent_extra = 'custom-thing/other'
|
|
self.assertTrue(
|
|
self.session.user_agent().endswith('custom-thing/other')
|
|
)
|
|
|
|
def test_execution_env_not_set(self):
|
|
self.assertFalse(self.session.user_agent().endswith('FooEnv'))
|
|
|
|
def test_execution_env_set(self):
|
|
self.environ['AWS_EXECUTION_ENV'] = 'FooEnv'
|
|
self.assertTrue(self.session.user_agent().endswith(' exec-env/FooEnv'))
|
|
|
|
def test_agent_extra_and_exec_env(self):
|
|
self.session.user_agent_extra = 'custom-thing/other'
|
|
self.environ['AWS_EXECUTION_ENV'] = 'FooEnv'
|
|
user_agent = self.session.user_agent()
|
|
self.assertTrue(user_agent.endswith('custom-thing/other'))
|
|
self.assertIn('exec-env/FooEnv', user_agent)
|
|
|
|
@requires_crt()
|
|
def test_crt_user_agent_appended(self):
|
|
user_agent = self.session.user_agent()
|
|
self.assertIn(' awscrt/', user_agent)
|
|
self.assertNotIn('awscrt/Unknown', user_agent)
|
|
|
|
@requires_crt()
|
|
def test_crt_and_extra_user_agent(self):
|
|
user_agent = self.session.user_agent()
|
|
self.assertIn(' awscrt/', user_agent)
|
|
self.assertNotIn('custom-thing/other', user_agent)
|
|
self.session.user_agent_extra = 'custom-thing/other'
|
|
user_agent_w_extra = self.session.user_agent()
|
|
self.assertIn(' awscrt/', user_agent)
|
|
self.assertTrue(user_agent_w_extra.endswith('custom-thing/other'))
|
|
|
|
|
|
class TestConfigLoaderObject(BaseSessionTest):
|
|
def test_config_loader_delegation(self):
|
|
session = create_session(profile='credfile-profile')
|
|
with temporary_file('w') as f:
|
|
f.write('[credfile-profile]\naws_access_key_id=a\n')
|
|
f.write('aws_secret_access_key=b\n')
|
|
f.flush()
|
|
session.set_config_variable('credentials_file', f.name)
|
|
# Now trying to retrieve the scoped config should pull in
|
|
# values from the shared credentials file.
|
|
self.assertEqual(
|
|
session.get_scoped_config(),
|
|
{'aws_access_key_id': 'a', 'aws_secret_access_key': 'b'},
|
|
)
|
|
|
|
|
|
class TestGetServiceModel(BaseSessionTest):
|
|
def test_get_service_model(self):
|
|
loader = mock.Mock()
|
|
loader.load_service_model.return_value = {
|
|
'metadata': {'serviceId': 'foo'}
|
|
}
|
|
self.session.register_component('data_loader', loader)
|
|
model = self.session.get_service_model('made_up')
|
|
self.assertIsInstance(model, ServiceModel)
|
|
self.assertEqual(model.service_name, 'made_up')
|
|
|
|
|
|
class TestGetPaginatorModel(BaseSessionTest):
|
|
def test_get_paginator_model(self):
|
|
loader = mock.Mock()
|
|
loader.load_service_model.return_value = {"pagination": {}}
|
|
self.session.register_component('data_loader', loader)
|
|
|
|
model = self.session.get_paginator_model('foo')
|
|
|
|
# Verify we get a PaginatorModel back
|
|
self.assertIsInstance(model, PaginatorModel)
|
|
# Verify we called the loader correctly.
|
|
loader.load_service_model.assert_called_with(
|
|
'foo', 'paginators-1', None
|
|
)
|
|
|
|
|
|
class TestGetWaiterModel(BaseSessionTest):
|
|
def test_get_waiter_model(self):
|
|
loader = mock.Mock()
|
|
loader.load_service_model.return_value = {"version": 2, "waiters": {}}
|
|
self.session.register_component('data_loader', loader)
|
|
|
|
model = self.session.get_waiter_model('foo')
|
|
|
|
# Verify we (1) get the expected return data,
|
|
self.assertIsInstance(model, WaiterModel)
|
|
self.assertEqual(model.waiter_names, [])
|
|
# and (2) call the loader correctly.
|
|
loader.load_service_model.assert_called_with('foo', 'waiters-2', None)
|
|
|
|
|
|
class TestCreateClient(BaseSessionTest):
|
|
def test_can_create_client(self):
|
|
sts_client = self.session.create_client('sts', 'us-west-2')
|
|
self.assertIsInstance(sts_client, client.BaseClient)
|
|
|
|
def test_credential_provider_not_called_when_creds_provided(self):
|
|
cred_provider = mock.Mock()
|
|
self.session.register_component('credential_provider', cred_provider)
|
|
self.session.create_client(
|
|
'sts',
|
|
'us-west-2',
|
|
aws_access_key_id='foo',
|
|
aws_secret_access_key='bar',
|
|
aws_session_token='baz',
|
|
)
|
|
self.assertFalse(
|
|
cred_provider.load_credentials.called,
|
|
"Credential provider was called even though "
|
|
"explicit credentials were provided to the "
|
|
"create_client call.",
|
|
)
|
|
|
|
def test_cred_provider_called_when_partial_creds_provided(self):
|
|
with self.assertRaises(botocore.exceptions.PartialCredentialsError):
|
|
self.session.create_client(
|
|
'sts',
|
|
'us-west-2',
|
|
aws_access_key_id='foo',
|
|
aws_secret_access_key=None,
|
|
)
|
|
with self.assertRaises(botocore.exceptions.PartialCredentialsError):
|
|
self.session.create_client(
|
|
'sts',
|
|
'us-west-2',
|
|
aws_access_key_id=None,
|
|
aws_secret_access_key='foo',
|
|
)
|
|
|
|
def test_cred_provider_not_called_on_unsigned_client(self):
|
|
cred_provider = mock.Mock()
|
|
self.session.register_component('credential_provider', cred_provider)
|
|
config = botocore.config.Config(signature_version=UNSIGNED)
|
|
self.session.create_client('sts', 'us-west-2', config=config)
|
|
self.assertFalse(cred_provider.load_credentials.called)
|
|
|
|
@mock.patch('botocore.client.ClientCreator')
|
|
def test_config_passed_to_client_creator(self, client_creator):
|
|
# Make sure there is no default set
|
|
self.assertEqual(self.session.get_default_client_config(), None)
|
|
|
|
# The config passed to the client should be the one that is used
|
|
# in creating the client.
|
|
config = botocore.config.Config(region_name='us-west-2')
|
|
self.session.create_client('sts', config=config)
|
|
client_creator.return_value.create_client.assert_called_with(
|
|
service_name=mock.ANY,
|
|
region_name=mock.ANY,
|
|
is_secure=mock.ANY,
|
|
endpoint_url=mock.ANY,
|
|
verify=mock.ANY,
|
|
credentials=mock.ANY,
|
|
scoped_config=mock.ANY,
|
|
client_config=config,
|
|
api_version=mock.ANY,
|
|
auth_token=mock.ANY,
|
|
)
|
|
|
|
@mock.patch('botocore.client.ClientCreator')
|
|
def test_create_client_with_default_client_config(self, client_creator):
|
|
config = botocore.config.Config()
|
|
self.session.set_default_client_config(config)
|
|
self.session.create_client('sts')
|
|
|
|
client_creator.return_value.create_client.assert_called_with(
|
|
service_name=mock.ANY,
|
|
region_name=mock.ANY,
|
|
is_secure=mock.ANY,
|
|
endpoint_url=mock.ANY,
|
|
verify=mock.ANY,
|
|
credentials=mock.ANY,
|
|
scoped_config=mock.ANY,
|
|
client_config=config,
|
|
api_version=mock.ANY,
|
|
auth_token=mock.ANY,
|
|
)
|
|
|
|
@mock.patch('botocore.client.ClientCreator')
|
|
def test_create_client_with_merging_client_configs(self, client_creator):
|
|
config = botocore.config.Config(region_name='us-west-2')
|
|
other_config = botocore.config.Config(region_name='us-east-1')
|
|
self.session.set_default_client_config(config)
|
|
self.session.create_client('sts', config=other_config)
|
|
|
|
# Grab the client config used in creating the client
|
|
used_client_config = (
|
|
client_creator.return_value.create_client.call_args[1][
|
|
'client_config'
|
|
]
|
|
)
|
|
# Check that the client configs were merged
|
|
self.assertEqual(used_client_config.region_name, 'us-east-1')
|
|
# Make sure that the client config used is not the default client
|
|
# config or the one passed in. It should be a new config.
|
|
self.assertIsNot(used_client_config, config)
|
|
self.assertIsNot(used_client_config, other_config)
|
|
|
|
def test_create_client_with_region(self):
|
|
ec2_client = self.session.create_client('ec2', 'us-west-2')
|
|
self.assertEqual(ec2_client.meta.region_name, 'us-west-2')
|
|
|
|
def test_create_client_with_region_and_client_config(self):
|
|
config = botocore.config.Config()
|
|
# Use a client config with no region configured.
|
|
ec2_client = self.session.create_client(
|
|
'ec2', region_name='us-west-2', config=config
|
|
)
|
|
self.assertEqual(ec2_client.meta.region_name, 'us-west-2')
|
|
|
|
# If the region name is changed, it should not change the
|
|
# region of the client
|
|
config.region_name = 'us-east-1'
|
|
self.assertEqual(ec2_client.meta.region_name, 'us-west-2')
|
|
|
|
# Now make a new client with the updated client config.
|
|
ec2_client = self.session.create_client('ec2', config=config)
|
|
self.assertEqual(ec2_client.meta.region_name, 'us-east-1')
|
|
|
|
def test_create_client_no_region_and_no_client_config(self):
|
|
ec2_client = self.session.create_client('ec2')
|
|
self.assertEqual(ec2_client.meta.region_name, 'us-west-11')
|
|
|
|
@mock.patch('botocore.client.ClientCreator')
|
|
def test_create_client_with_ca_bundle_from_config(self, client_creator):
|
|
with temporary_file('w') as f:
|
|
del self.environ['FOO_PROFILE']
|
|
self.environ['FOO_CONFIG_FILE'] = f.name
|
|
f.write('[default]\n')
|
|
f.write('foo_ca_bundle=config-certs.pem\n')
|
|
f.flush()
|
|
|
|
self.session.create_client('ec2', 'us-west-2')
|
|
call_kwargs = client_creator.return_value.create_client.call_args[
|
|
1
|
|
]
|
|
self.assertEqual(call_kwargs['verify'], 'config-certs.pem')
|
|
|
|
@mock.patch('botocore.client.ClientCreator')
|
|
def test_create_client_with_ca_bundle_from_env_var(self, client_creator):
|
|
self.environ['FOO_AWS_CA_BUNDLE'] = 'env-certs.pem'
|
|
self.session.create_client('ec2', 'us-west-2')
|
|
call_kwargs = client_creator.return_value.create_client.call_args[1]
|
|
self.assertEqual(call_kwargs['verify'], 'env-certs.pem')
|
|
|
|
@mock.patch('botocore.client.ClientCreator')
|
|
def test_create_client_with_verify_param(self, client_creator):
|
|
self.session.create_client(
|
|
'ec2', 'us-west-2', verify='verify-certs.pem'
|
|
)
|
|
call_kwargs = client_creator.return_value.create_client.call_args[1]
|
|
self.assertEqual(call_kwargs['verify'], 'verify-certs.pem')
|
|
|
|
@mock.patch('botocore.client.ClientCreator')
|
|
def test_create_client_verify_param_overrides_all(self, client_creator):
|
|
with temporary_file('w') as f:
|
|
# Set the ca cert using the config file
|
|
del self.environ['FOO_PROFILE']
|
|
self.environ['FOO_CONFIG_FILE'] = f.name
|
|
f.write('[default]\n')
|
|
f.write('foo_ca_bundle=config-certs.pem\n')
|
|
f.flush()
|
|
|
|
# Set the ca cert with an environment variable
|
|
self.environ['FOO_AWS_CA_BUNDLE'] = 'env-certs.pem'
|
|
|
|
# Set the ca cert using the verify parameter
|
|
self.session.create_client(
|
|
'ec2', 'us-west-2', verify='verify-certs.pem'
|
|
)
|
|
call_kwargs = client_creator.return_value.create_client.call_args[
|
|
1
|
|
]
|
|
# The verify parameter should override all the other
|
|
# configurations
|
|
self.assertEqual(call_kwargs['verify'], 'verify-certs.pem')
|
|
|
|
@mock.patch('botocore.client.ClientCreator')
|
|
def test_create_client_use_no_api_version_by_default(self, client_creator):
|
|
self.session.create_client('myservice', 'us-west-2')
|
|
call_kwargs = client_creator.return_value.create_client.call_args[1]
|
|
self.assertEqual(call_kwargs['api_version'], None)
|
|
|
|
@mock.patch('botocore.client.ClientCreator')
|
|
def test_create_client_uses_api_version_from_config(self, client_creator):
|
|
config_api_version = '2012-01-01'
|
|
with temporary_file('w') as f:
|
|
del self.environ['FOO_PROFILE']
|
|
self.environ['FOO_CONFIG_FILE'] = f.name
|
|
f.write('[default]\n')
|
|
f.write(
|
|
'foo_api_versions =\n'
|
|
' myservice = %s\n' % config_api_version
|
|
)
|
|
f.flush()
|
|
|
|
self.session.create_client('myservice', 'us-west-2')
|
|
call_kwargs = client_creator.return_value.create_client.call_args[
|
|
1
|
|
]
|
|
self.assertEqual(call_kwargs['api_version'], config_api_version)
|
|
|
|
@mock.patch('botocore.client.ClientCreator')
|
|
def test_can_specify_multiple_versions_from_config(self, client_creator):
|
|
config_api_version = '2012-01-01'
|
|
second_config_api_version = '2013-01-01'
|
|
with temporary_file('w') as f:
|
|
del self.environ['FOO_PROFILE']
|
|
self.environ['FOO_CONFIG_FILE'] = f.name
|
|
f.write('[default]\n')
|
|
f.write(
|
|
f'foo_api_versions =\n'
|
|
f' myservice = {config_api_version}\n'
|
|
f' myservice2 = {second_config_api_version}\n'
|
|
)
|
|
f.flush()
|
|
|
|
self.session.create_client('myservice', 'us-west-2')
|
|
call_kwargs = client_creator.return_value.create_client.call_args[
|
|
1
|
|
]
|
|
self.assertEqual(call_kwargs['api_version'], config_api_version)
|
|
|
|
self.session.create_client('myservice2', 'us-west-2')
|
|
call_kwargs = client_creator.return_value.create_client.call_args[
|
|
1
|
|
]
|
|
self.assertEqual(
|
|
call_kwargs['api_version'], second_config_api_version
|
|
)
|
|
|
|
@mock.patch('botocore.client.ClientCreator')
|
|
def test_param_api_version_overrides_config_value(self, client_creator):
|
|
config_api_version = '2012-01-01'
|
|
override_api_version = '2014-01-01'
|
|
with temporary_file('w') as f:
|
|
del self.environ['FOO_PROFILE']
|
|
self.environ['FOO_CONFIG_FILE'] = f.name
|
|
f.write('[default]\n')
|
|
f.write(
|
|
'foo_api_versions =\n'
|
|
' myservice = %s\n' % config_api_version
|
|
)
|
|
f.flush()
|
|
|
|
self.session.create_client(
|
|
'myservice', 'us-west-2', api_version=override_api_version
|
|
)
|
|
call_kwargs = client_creator.return_value.create_client.call_args[
|
|
1
|
|
]
|
|
self.assertEqual(call_kwargs['api_version'], override_api_version)
|
|
|
|
@mock.patch('botocore.client.ClientCreator')
|
|
def test_defaults_mode_resolved_from_config_store(self, client_creator):
|
|
config_store = self.session.get_component('config_store')
|
|
config_store.set_config_variable('defaults_mode', 'standard')
|
|
self.session.create_client('sts', 'us-west-2')
|
|
self.assertIsNot(client_creator.call_args[0][-1], config_store)
|
|
|
|
@mock.patch('botocore.client.ClientCreator')
|
|
def test_defaults_mode_resolved_from_client_config(self, client_creator):
|
|
config_store = self.session.get_component('config_store')
|
|
config = botocore.config.Config(defaults_mode='standard')
|
|
self.session.create_client('sts', 'us-west-2', config=config)
|
|
self.assertIsNot(client_creator.call_args[0][-1], config_store)
|
|
|
|
@mock.patch('botocore.client.ClientCreator')
|
|
def test_defaults_mode_resolved_invalid_mode_exception(
|
|
self, client_creator
|
|
):
|
|
with self.assertRaises(botocore.exceptions.InvalidDefaultsMode):
|
|
config = botocore.config.Config(defaults_mode='foo')
|
|
self.session.create_client('sts', 'us-west-2', config=config)
|
|
|
|
@mock.patch('botocore.client.ClientCreator')
|
|
def test_defaults_mode_resolved_legacy(self, client_creator):
|
|
config_store = self.session.get_component('config_store')
|
|
self.session.create_client('sts', 'us-west-2')
|
|
self.assertIs(client_creator.call_args[0][-1], config_store)
|
|
|
|
|
|
class TestSessionComponent(BaseSessionTest):
|
|
def test_internal_component(self):
|
|
component = object()
|
|
self.session._register_internal_component('internal', component)
|
|
self.assertIs(
|
|
self.session._get_internal_component('internal'), component
|
|
)
|
|
with self.assertRaises(ValueError):
|
|
# get_component has been deprecated to the public
|
|
with pytest.warns(DeprecationWarning):
|
|
self.session.get_component('internal')
|
|
|
|
def test_internal_endpoint_resolver_is_same_as_deprecated_public(self):
|
|
endpoint_resolver = self.session._get_internal_component(
|
|
'endpoint_resolver'
|
|
)
|
|
# get_component has been deprecated to the public
|
|
with pytest.warns(DeprecationWarning):
|
|
self.assertIs(
|
|
self.session.get_component('endpoint_resolver'),
|
|
endpoint_resolver,
|
|
)
|
|
|
|
def test_internal_exceptions_factory_is_same_as_deprecated_public(self):
|
|
exceptions_factory = self.session._get_internal_component(
|
|
'exceptions_factory'
|
|
)
|
|
# get_component has been deprecated to the public
|
|
with pytest.warns(DeprecationWarning):
|
|
self.assertIs(
|
|
self.session.get_component('exceptions_factory'),
|
|
exceptions_factory,
|
|
)
|
|
|
|
|
|
class TestClientMonitoring(BaseSessionTest):
|
|
def assert_created_client_is_monitored(self, session):
|
|
with mock.patch(
|
|
'botocore.monitoring.Monitor', spec=True
|
|
) as mock_monitor:
|
|
client = session.create_client('ec2', 'us-west-2')
|
|
mock_monitor.return_value.register.assert_called_with(
|
|
client.meta.events
|
|
)
|
|
|
|
def assert_monitoring_host_and_port(self, session, host, port):
|
|
with mock.patch(
|
|
'botocore.monitoring.SocketPublisher', spec=True
|
|
) as mock_publisher:
|
|
session.create_client('ec2', 'us-west-2')
|
|
self.assertEqual(mock_publisher.call_count, 1)
|
|
_, args, kwargs = mock_publisher.mock_calls[0]
|
|
self.assertEqual(kwargs.get('host'), host)
|
|
self.assertEqual(kwargs.get('port'), port)
|
|
|
|
def assert_created_client_is_not_monitored(self, session):
|
|
with mock.patch(
|
|
'botocore.session.monitoring.Monitor', spec=True
|
|
) as mock_monitor:
|
|
session.create_client('ec2', 'us-west-2')
|
|
mock_monitor.return_value.register.assert_not_called()
|
|
|
|
def test_with_csm_enabled_from_config(self):
|
|
with temporary_file('w') as f:
|
|
del self.environ['FOO_PROFILE']
|
|
self.environ['FOO_CONFIG_FILE'] = f.name
|
|
f.write('[default]\n')
|
|
f.write('csm_enabled=true\n')
|
|
f.flush()
|
|
self.assert_created_client_is_monitored(self.session)
|
|
|
|
def test_with_csm_enabled_from_env(self):
|
|
self.environ['AWS_CSM_ENABLED'] = 'true'
|
|
self.assert_created_client_is_monitored(self.session)
|
|
|
|
def test_with_csm_host(self):
|
|
custom_host = '10.13.37.1'
|
|
self.environ['AWS_CSM_ENABLED'] = 'true'
|
|
self.environ['AWS_CSM_HOST'] = custom_host
|
|
self.assert_monitoring_host_and_port(self.session, custom_host, 31000)
|
|
|
|
def test_with_csm_port(self):
|
|
custom_port = '1234'
|
|
self.environ['AWS_CSM_ENABLED'] = 'true'
|
|
self.environ['AWS_CSM_PORT'] = custom_port
|
|
self.assert_monitoring_host_and_port(
|
|
self.session,
|
|
'127.0.0.1',
|
|
int(custom_port),
|
|
)
|
|
|
|
def test_with_csm_disabled_from_config(self):
|
|
with temporary_file('w') as f:
|
|
del self.environ['FOO_PROFILE']
|
|
self.environ['FOO_CONFIG_FILE'] = f.name
|
|
f.write('[default]\n')
|
|
f.write('csm_enabled=false\n')
|
|
f.flush()
|
|
self.assert_created_client_is_not_monitored(self.session)
|
|
|
|
def test_with_csm_disabled_from_env(self):
|
|
self.environ['AWS_CSM_ENABLED'] = 'false'
|
|
self.assert_created_client_is_not_monitored(self.session)
|
|
|
|
def test_csm_not_configured(self):
|
|
self.assert_created_client_is_not_monitored(self.session)
|
|
|
|
|
|
class TestComponentLocator(unittest.TestCase):
|
|
def setUp(self):
|
|
self.components = botocore.session.ComponentLocator()
|
|
|
|
def test_unknown_component_raises_exception(self):
|
|
with self.assertRaises(ValueError):
|
|
self.components.get_component('unknown-component')
|
|
|
|
def test_can_register_and_retrieve_component(self):
|
|
component = object()
|
|
self.components.register_component('foo', component)
|
|
self.assertIs(self.components.get_component('foo'), component)
|
|
|
|
def test_last_registration_wins(self):
|
|
first = object()
|
|
second = object()
|
|
self.components.register_component('foo', first)
|
|
self.components.register_component('foo', second)
|
|
self.assertIs(self.components.get_component('foo'), second)
|
|
|
|
def test_can_lazy_register_a_component(self):
|
|
component = object()
|
|
lazy = lambda: component
|
|
self.components.lazy_register_component('foo', lazy)
|
|
self.assertIs(self.components.get_component('foo'), component)
|
|
|
|
def test_latest_registration_wins_even_if_lazy(self):
|
|
first = object()
|
|
second = object()
|
|
lazy_second = lambda: second
|
|
self.components.register_component('foo', first)
|
|
self.components.lazy_register_component('foo', lazy_second)
|
|
self.assertIs(self.components.get_component('foo'), second)
|
|
|
|
def test_latest_registration_overrides_lazy(self):
|
|
first = object()
|
|
second = object()
|
|
lazy_first = lambda: first
|
|
self.components.lazy_register_component('foo', lazy_first)
|
|
self.components.register_component('foo', second)
|
|
self.assertIs(self.components.get_component('foo'), second)
|
|
|
|
def test_lazy_registration_factory_does_not_remove_from_list_on_error(
|
|
self,
|
|
):
|
|
class ArbitraryError(Exception):
|
|
pass
|
|
|
|
def bad_factory():
|
|
raise ArbitraryError("Factory raises an exception.")
|
|
|
|
self.components.lazy_register_component('foo', bad_factory)
|
|
|
|
with self.assertRaises(ArbitraryError):
|
|
self.components.get_component('foo')
|
|
|
|
# Trying again should raise the same exception,
|
|
# not an ValueError("Unknown component")
|
|
with self.assertRaises(ArbitraryError):
|
|
self.components.get_component('foo')
|
|
|
|
|
|
class TestDefaultClientConfig(BaseSessionTest):
|
|
def test_new_session_has_no_default_client_config(self):
|
|
self.assertEqual(self.session.get_default_client_config(), None)
|
|
|
|
def test_set_and_get_client_config(self):
|
|
client_config = botocore.config.Config()
|
|
self.session.set_default_client_config(client_config)
|
|
self.assertIs(self.session.get_default_client_config(), client_config)
|
|
|
|
|
|
class TestSessionRegionSetup(BaseSessionTest):
|
|
def test_new_session_with_valid_region(self):
|
|
s3_client = self.session.create_client('s3', 'us-west-2')
|
|
self.assertIsInstance(s3_client, client.BaseClient)
|
|
self.assertEqual(s3_client.meta.region_name, 'us-west-2')
|
|
|
|
def test_new_session_with_unknown_region(self):
|
|
s3_client = self.session.create_client('s3', 'MyCustomRegion1')
|
|
self.assertIsInstance(s3_client, client.BaseClient)
|
|
self.assertEqual(s3_client.meta.region_name, 'MyCustomRegion1')
|
|
|
|
def test_new_session_with_invalid_region(self):
|
|
with self.assertRaises(botocore.exceptions.InvalidRegionError):
|
|
self.session.create_client('s3', 'not.a.real#region')
|
|
|
|
def test_new_session_with_none_region(self):
|
|
s3_client = self.session.create_client('s3', region_name=None)
|
|
self.assertIsInstance(s3_client, client.BaseClient)
|
|
self.assertTrue(s3_client.meta.region_name is not None)
|
|
|
|
|
|
class TestInitializationHooks(BaseSessionTest):
|
|
def test_can_register_init_hook(self):
|
|
call_args = []
|
|
|
|
def init_hook(session):
|
|
call_args.append(session)
|
|
|
|
register_initializer(init_hook)
|
|
self.addCleanup(unregister_initializer, init_hook)
|
|
session = create_session()
|
|
self.assertEqual(call_args, [session])
|
|
|
|
def test_can_unregister_hook(self):
|
|
call_args = []
|
|
|
|
def init_hook(session):
|
|
call_args.append(session)
|
|
|
|
register_initializer(init_hook)
|
|
unregister_initializer(init_hook)
|
|
create_session()
|
|
self.assertEqual(call_args, [])
|
|
|
|
def test_unregister_hook_raises_value_error(self):
|
|
not_registered = lambda session: None
|
|
with self.assertRaises(ValueError):
|
|
self.assertRaises(unregister_initializer(not_registered))
|