832 lines
29 KiB
Python
832 lines
29 KiB
Python
# Copyright 2018 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 copy
|
|
import os
|
|
|
|
import pytest
|
|
|
|
import botocore.configprovider
|
|
import botocore.session as session
|
|
from botocore.configprovider import (
|
|
BaseProvider,
|
|
ChainProvider,
|
|
ConfigChainFactory,
|
|
ConfigValueStore,
|
|
ConstantProvider,
|
|
DefaultConfigResolver,
|
|
EnvironmentProvider,
|
|
InstanceVarProvider,
|
|
ScopedConfigProvider,
|
|
SectionConfigProvider,
|
|
SmartDefaultsConfigStoreFactory,
|
|
)
|
|
from botocore.exceptions import ConnectTimeoutError
|
|
from botocore.utils import IMDSRegionProvider
|
|
from tests import mock, unittest
|
|
|
|
|
|
class TestConfigChainFactory(unittest.TestCase):
|
|
def assert_chain_does_provide(
|
|
self,
|
|
instance_map,
|
|
environ_map,
|
|
scoped_config_map,
|
|
create_config_chain_args,
|
|
expected_value,
|
|
):
|
|
fake_session = mock.Mock(spec=session.Session)
|
|
fake_session.get_scoped_config.return_value = scoped_config_map
|
|
fake_session.instance_variables.return_value = instance_map
|
|
builder = ConfigChainFactory(fake_session, environ=environ_map)
|
|
chain = builder.create_config_chain(**create_config_chain_args)
|
|
value = chain.provide()
|
|
self.assertEqual(value, expected_value)
|
|
|
|
def test_chain_builder_can_provide_instance(self):
|
|
self.assert_chain_does_provide(
|
|
instance_map={'instance_var': 'from-instance'},
|
|
environ_map={},
|
|
scoped_config_map={},
|
|
create_config_chain_args={
|
|
'instance_name': 'instance_var',
|
|
},
|
|
expected_value='from-instance',
|
|
)
|
|
|
|
def test_chain_builder_can_skip_instance(self):
|
|
self.assert_chain_does_provide(
|
|
instance_map={'wrong_instance_var': 'instance'},
|
|
environ_map={'ENV_VAR': 'env'},
|
|
scoped_config_map={},
|
|
create_config_chain_args={
|
|
'instance_name': 'instance_var',
|
|
'env_var_names': 'ENV_VAR',
|
|
},
|
|
expected_value='env',
|
|
)
|
|
|
|
def test_chain_builder_can_provide_env_var(self):
|
|
self.assert_chain_does_provide(
|
|
instance_map={},
|
|
environ_map={'ENV_VAR': 'from-env'},
|
|
scoped_config_map={},
|
|
create_config_chain_args={
|
|
'env_var_names': 'ENV_VAR',
|
|
},
|
|
expected_value='from-env',
|
|
)
|
|
|
|
def test_does_provide_none_if_no_variable_exists_in_env_var_list(self):
|
|
self.assert_chain_does_provide(
|
|
instance_map={},
|
|
environ_map={},
|
|
scoped_config_map={},
|
|
create_config_chain_args={
|
|
'env_var_names': ['FOO'],
|
|
},
|
|
expected_value=None,
|
|
)
|
|
|
|
def test_does_provide_value_if_variable_exists_in_env_var_list(self):
|
|
self.assert_chain_does_provide(
|
|
instance_map={},
|
|
environ_map={'FOO': 'bar'},
|
|
scoped_config_map={},
|
|
create_config_chain_args={
|
|
'env_var_names': ['FOO'],
|
|
},
|
|
expected_value='bar',
|
|
)
|
|
|
|
def test_does_provide_first_non_none_value_first_in_env_var_list(self):
|
|
self.assert_chain_does_provide(
|
|
instance_map={},
|
|
environ_map={'FOO': 'baz'},
|
|
scoped_config_map={},
|
|
create_config_chain_args={
|
|
'env_var_names': ['FOO', 'BAR'],
|
|
},
|
|
expected_value='baz',
|
|
)
|
|
|
|
def test_does_provide_first_non_none_value_second_in_env_var_list(self):
|
|
self.assert_chain_does_provide(
|
|
instance_map={},
|
|
environ_map={'BAR': 'baz'},
|
|
scoped_config_map={},
|
|
create_config_chain_args={
|
|
'env_var_names': ['FOO', 'BAR'],
|
|
},
|
|
expected_value='baz',
|
|
)
|
|
|
|
def test_does_provide_none_if_all_list_env_vars_are_none(self):
|
|
self.assert_chain_does_provide(
|
|
instance_map={},
|
|
environ_map={},
|
|
scoped_config_map={},
|
|
create_config_chain_args={
|
|
'env_var_names': ['FOO', 'BAR'],
|
|
},
|
|
expected_value=None,
|
|
)
|
|
|
|
def test_does_provide_first_value_when_both_env_vars_exist(self):
|
|
self.assert_chain_does_provide(
|
|
instance_map={},
|
|
environ_map={'FOO': 'baz', 'BAR': 'buz'},
|
|
scoped_config_map={},
|
|
create_config_chain_args={
|
|
'env_var_names': ['FOO', 'BAR'],
|
|
},
|
|
expected_value='baz',
|
|
)
|
|
|
|
def test_chain_builder_can_provide_config_var(self):
|
|
self.assert_chain_does_provide(
|
|
instance_map={},
|
|
environ_map={},
|
|
scoped_config_map={'config_var': 'from-config'},
|
|
create_config_chain_args={
|
|
'config_property_names': 'config_var',
|
|
},
|
|
expected_value='from-config',
|
|
)
|
|
|
|
def test_chain_builder_can_provide_nested_config_var(self):
|
|
self.assert_chain_does_provide(
|
|
instance_map={},
|
|
environ_map={},
|
|
scoped_config_map={'config_var': {'nested-key': 'nested-val'}},
|
|
create_config_chain_args={
|
|
'config_property_names': ('config_var', 'nested-key'),
|
|
},
|
|
expected_value='nested-val',
|
|
)
|
|
|
|
def test_provide_value_from_config_list(self):
|
|
self.assert_chain_does_provide(
|
|
instance_map={},
|
|
environ_map={},
|
|
scoped_config_map={'var': 'val'},
|
|
create_config_chain_args={
|
|
'config_property_names': ['var'],
|
|
},
|
|
expected_value='val',
|
|
)
|
|
|
|
def test_provide_value_from_config_list_looks_for_non_none_vals(self):
|
|
self.assert_chain_does_provide(
|
|
instance_map={},
|
|
environ_map={},
|
|
scoped_config_map={'non_none_var': 'non_none_val'},
|
|
create_config_chain_args={
|
|
'config_property_names': ['none_var', 'non_none_var'],
|
|
},
|
|
expected_value='non_none_val',
|
|
)
|
|
|
|
def test_provide_value_from_config_list_retrieves_first_non_none_val(self):
|
|
self.assert_chain_does_provide(
|
|
instance_map={},
|
|
environ_map={},
|
|
scoped_config_map={'first': 'first_val', 'second': 'second_val'},
|
|
create_config_chain_args={
|
|
'config_property_names': ['first', 'second'],
|
|
},
|
|
expected_value='first_val',
|
|
)
|
|
|
|
def test_provide_value_from_config_list_if_all_vars_are_none(self):
|
|
self.assert_chain_does_provide(
|
|
instance_map={},
|
|
environ_map={},
|
|
scoped_config_map={},
|
|
create_config_chain_args={
|
|
'config_property_names': ['config1', 'config2'],
|
|
},
|
|
expected_value=None,
|
|
)
|
|
|
|
def test_provide_value_from_list_with_nested_var(self):
|
|
self.assert_chain_does_provide(
|
|
instance_map={},
|
|
environ_map={},
|
|
scoped_config_map={'section': {'nested_var': 'nested_val'}},
|
|
create_config_chain_args={
|
|
'config_property_names': [('section', 'nested_var')],
|
|
},
|
|
expected_value='nested_val',
|
|
)
|
|
|
|
def test_chain_builder_can_provide_default(self):
|
|
self.assert_chain_does_provide(
|
|
instance_map={},
|
|
environ_map={},
|
|
scoped_config_map={},
|
|
create_config_chain_args={'default': 'from-default'},
|
|
expected_value='from-default',
|
|
)
|
|
|
|
def test_chain_provider_does_follow_priority_instance_var(self):
|
|
self.assert_chain_does_provide(
|
|
instance_map={'instance_var': 'from-instance'},
|
|
environ_map={'ENV_VAR': 'from-env'},
|
|
scoped_config_map={'config_var': 'from-config'},
|
|
create_config_chain_args={
|
|
'instance_name': 'instance_var',
|
|
'env_var_names': 'ENV_VAR',
|
|
'config_property_names': 'config_var',
|
|
'default': 'from-default',
|
|
},
|
|
expected_value='from-instance',
|
|
)
|
|
|
|
def test_chain_provider_does_follow_priority_env_var(self):
|
|
self.assert_chain_does_provide(
|
|
instance_map={'wrong_instance_var': 'from-instance'},
|
|
environ_map={'ENV_VAR': 'from-env'},
|
|
scoped_config_map={'config_var': 'from-confi'},
|
|
create_config_chain_args={
|
|
'instance_name': 'instance_var',
|
|
'env_var_names': 'ENV_VAR',
|
|
'config_property_names': 'config_var',
|
|
'default': 'from-default',
|
|
},
|
|
expected_value='from-env',
|
|
)
|
|
|
|
def test_chain_provider_does_follow_priority_config(self):
|
|
self.assert_chain_does_provide(
|
|
instance_map={'wrong_instance_var': 'from-instance'},
|
|
environ_map={'WRONG_ENV_VAR': 'from-env'},
|
|
scoped_config_map={'config_var': 'from-config'},
|
|
create_config_chain_args={
|
|
'instance_name': 'instance_var',
|
|
'env_var_names': 'ENV_VAR',
|
|
'config_property_names': 'config_var',
|
|
'default': 'from-default',
|
|
},
|
|
expected_value='from-config',
|
|
)
|
|
|
|
def test_chain_provider_does_follow_priority_default(self):
|
|
self.assert_chain_does_provide(
|
|
instance_map={'wrong_instance_var': 'from-instance'},
|
|
environ_map={'WRONG_ENV_VAR': 'from-env'},
|
|
scoped_config_map={'wrong_config_var': 'from-config'},
|
|
create_config_chain_args={
|
|
'instance_name': 'instance_var',
|
|
'env_var_names': 'ENV_VAR',
|
|
'config_property_names': 'config_var',
|
|
'default': 'from-default',
|
|
},
|
|
expected_value='from-default',
|
|
)
|
|
|
|
|
|
class TestConfigValueStore(unittest.TestCase):
|
|
def test_does_provide_none_if_no_variable_exists(self):
|
|
provider = ConfigValueStore()
|
|
value = provider.get_config_variable('fake_variable')
|
|
self.assertIsNone(value)
|
|
|
|
def test_does_provide_value_if_variable_exists(self):
|
|
mock_value_provider = mock.Mock(spec=BaseProvider)
|
|
mock_value_provider.provide.return_value = 'foo'
|
|
provider = ConfigValueStore(
|
|
mapping={
|
|
'fake_variable': mock_value_provider,
|
|
}
|
|
)
|
|
value = provider.get_config_variable('fake_variable')
|
|
self.assertEqual(value, 'foo')
|
|
|
|
def test_can_set_variable(self):
|
|
provider = ConfigValueStore()
|
|
provider.set_config_variable('fake_variable', 'foo')
|
|
value = provider.get_config_variable('fake_variable')
|
|
self.assertEqual(value, 'foo')
|
|
|
|
def test_can_set_config_provider(self):
|
|
foo_value_provider = mock.Mock(spec=BaseProvider)
|
|
foo_value_provider.provide.return_value = 'foo'
|
|
provider = ConfigValueStore(
|
|
mapping={
|
|
'fake_variable': foo_value_provider,
|
|
}
|
|
)
|
|
|
|
value = provider.get_config_variable('fake_variable')
|
|
self.assertEqual(value, 'foo')
|
|
|
|
bar_value_provider = mock.Mock(spec=BaseProvider)
|
|
bar_value_provider.provide.return_value = 'bar'
|
|
provider.set_config_provider('fake_variable', bar_value_provider)
|
|
|
|
value = provider.get_config_variable('fake_variable')
|
|
self.assertEqual(value, 'bar')
|
|
|
|
def test_can_get_config_provider(self):
|
|
chain_provider = ChainProvider(
|
|
providers=[ConstantProvider(value='bar')]
|
|
)
|
|
config_value_store = ConfigValueStore(
|
|
mapping={
|
|
'fake_variable': chain_provider,
|
|
}
|
|
)
|
|
provider = config_value_store.get_config_provider('fake_variable')
|
|
value = config_value_store.get_config_variable('fake_variable')
|
|
self.assertIsInstance(provider, ChainProvider)
|
|
self.assertEqual(value, 'bar')
|
|
|
|
def test_can_get_config_provider_non_chain_provider(self):
|
|
constant_provider = ConstantProvider(value='bar')
|
|
config_value_store = ConfigValueStore(
|
|
mapping={
|
|
'fake_variable': constant_provider,
|
|
}
|
|
)
|
|
provider = config_value_store.get_config_provider('fake_variable')
|
|
value = config_value_store.get_config_variable('fake_variable')
|
|
self.assertIsInstance(provider, ConstantProvider)
|
|
self.assertEqual(value, 'bar')
|
|
|
|
|
|
class TestInstanceVarProvider(unittest.TestCase):
|
|
def assert_provides_value(self, name, instance_map, expected_value):
|
|
fake_session = mock.Mock(spec=session.Session)
|
|
fake_session.instance_variables.return_value = instance_map
|
|
|
|
provider = InstanceVarProvider(
|
|
instance_var=name,
|
|
session=fake_session,
|
|
)
|
|
value = provider.provide()
|
|
self.assertEqual(value, expected_value)
|
|
|
|
def test_can_provide_value(self):
|
|
self.assert_provides_value(
|
|
name='foo',
|
|
instance_map={'foo': 'bar'},
|
|
expected_value='bar',
|
|
)
|
|
|
|
def test_does_provide_none_if_value_not_in_dict(self):
|
|
self.assert_provides_value(
|
|
name='foo',
|
|
instance_map={},
|
|
expected_value=None,
|
|
)
|
|
|
|
|
|
class TestEnvironmentProvider(unittest.TestCase):
|
|
def assert_does_provide(self, env, name, expected_value):
|
|
provider = EnvironmentProvider(name=name, env=env)
|
|
value = provider.provide()
|
|
self.assertEqual(value, expected_value)
|
|
|
|
def test_does_provide_none_if_no_variable_exists(self):
|
|
self.assert_does_provide(
|
|
name='FOO',
|
|
env={},
|
|
expected_value=None,
|
|
)
|
|
|
|
def test_does_provide_value_if_variable_exists(self):
|
|
self.assert_does_provide(
|
|
name='FOO',
|
|
env={
|
|
'FOO': 'bar',
|
|
},
|
|
expected_value='bar',
|
|
)
|
|
|
|
|
|
class TestScopedConfigProvider(unittest.TestCase):
|
|
def assert_provides_value(
|
|
self, config_file_values, config_var_name, expected_value
|
|
):
|
|
fake_session = mock.Mock(spec=session.Session)
|
|
fake_session.get_scoped_config.return_value = config_file_values
|
|
property_provider = ScopedConfigProvider(
|
|
config_var_name=config_var_name,
|
|
session=fake_session,
|
|
)
|
|
value = property_provider.provide()
|
|
self.assertEqual(value, expected_value)
|
|
|
|
def test_can_provide_value(self):
|
|
self.assert_provides_value(
|
|
config_file_values={'foo': 'bar'},
|
|
config_var_name='foo',
|
|
expected_value='bar',
|
|
)
|
|
|
|
def test_does_provide_none_if_var_not_in_config(self):
|
|
self.assert_provides_value(
|
|
config_file_values={'foo': 'bar'},
|
|
config_var_name='no_such_var',
|
|
expected_value=None,
|
|
)
|
|
|
|
def test_provide_nested_value(self):
|
|
self.assert_provides_value(
|
|
config_file_values={'section': {'nested_var': 'nested_val'}},
|
|
config_var_name=('section', 'nested_var'),
|
|
expected_value='nested_val',
|
|
)
|
|
|
|
def test_provide_nested_value_but_not_section(self):
|
|
self.assert_provides_value(
|
|
config_file_values={'section': 'not-nested'},
|
|
config_var_name=('section', 'nested_var'),
|
|
expected_value=None,
|
|
)
|
|
|
|
|
|
def _make_provider_that_returns(return_value):
|
|
provider = mock.Mock(spec=BaseProvider)
|
|
provider.provide.return_value = return_value
|
|
return provider
|
|
|
|
|
|
def _make_providers_that_return(return_values):
|
|
mocks = []
|
|
for return_value in return_values:
|
|
provider = _make_provider_that_returns(return_value)
|
|
mocks.append(provider)
|
|
return mocks
|
|
|
|
|
|
def assert_chain_does_provide(providers, expected_value):
|
|
provider = ChainProvider(
|
|
providers=providers,
|
|
)
|
|
assert provider.provide() == expected_value
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
'case',
|
|
(
|
|
(None, []),
|
|
(None, [None]),
|
|
('foo', ['foo']),
|
|
('foo', ['foo', 'bar']),
|
|
('bar', [None, 'bar']),
|
|
('foo', ['foo', None]),
|
|
('baz', [None, None, 'baz']),
|
|
('bar', [None, 'bar', None]),
|
|
('foo', ['foo', 'bar', None]),
|
|
('foo', ['foo', 'bar', 'baz']),
|
|
),
|
|
)
|
|
def test_chain_provider(case):
|
|
# Each case is a tuple with the first element being the expected return
|
|
# value from the ChainProvider. The second value being a list of return
|
|
# values from the individual providers that are in the chain.
|
|
assert_chain_does_provide(_make_providers_that_return(case[1]), case[0])
|
|
|
|
|
|
class TestChainProvider(unittest.TestCase):
|
|
def test_can_convert_provided_value(self):
|
|
chain_provider = ChainProvider(
|
|
providers=_make_providers_that_return(['1']),
|
|
conversion_func=int,
|
|
)
|
|
value = chain_provider.provide()
|
|
self.assertIsInstance(value, int)
|
|
self.assertEqual(value, 1)
|
|
|
|
|
|
class TestConstantProvider(unittest.TestCase):
|
|
def test_can_provide_value(self):
|
|
provider = ConstantProvider(value='foo')
|
|
value = provider.provide()
|
|
self.assertEqual(value, 'foo')
|
|
|
|
|
|
class TestSectionConfigProvider(unittest.TestCase):
|
|
def assert_provides_value(
|
|
self,
|
|
config_file_values,
|
|
section_name,
|
|
expected_value,
|
|
override_providers=None,
|
|
):
|
|
fake_session = mock.Mock(spec=session.Session)
|
|
fake_session.get_scoped_config.return_value = config_file_values
|
|
provider = SectionConfigProvider(
|
|
section_name=section_name,
|
|
session=fake_session,
|
|
override_providers=override_providers,
|
|
)
|
|
value = provider.provide()
|
|
self.assertEqual(value, expected_value)
|
|
|
|
def test_provide_section_config(self):
|
|
self.assert_provides_value(
|
|
config_file_values={'mysection': {'section_var': 'section_val'}},
|
|
section_name='mysection',
|
|
expected_value={'section_var': 'section_val'},
|
|
)
|
|
|
|
def test_provide_service_config_missing_service(self):
|
|
self.assert_provides_value(
|
|
config_file_values={},
|
|
section_name='mysection',
|
|
expected_value=None,
|
|
)
|
|
|
|
def test_provide_service_config_not_a_section(self):
|
|
self.assert_provides_value(
|
|
config_file_values={'myservice': 'not-a-section'},
|
|
section_name='mysection',
|
|
expected_value=None,
|
|
)
|
|
|
|
def test_provide_section_config_with_overrides(self):
|
|
self.assert_provides_value(
|
|
config_file_values={
|
|
'mysection': {
|
|
'override_var': 'from_config_file',
|
|
'no_override_var': 'from_config_file',
|
|
}
|
|
},
|
|
section_name='mysection',
|
|
override_providers={'override_var': ConstantProvider('override')},
|
|
expected_value={
|
|
'override_var': 'override',
|
|
'no_override_var': 'from_config_file',
|
|
},
|
|
)
|
|
|
|
def test_provide_section_config_with_only_overrides(self):
|
|
self.assert_provides_value(
|
|
config_file_values={},
|
|
section_name='mysection',
|
|
override_providers={'override_var': ConstantProvider('override')},
|
|
expected_value={
|
|
'override_var': 'override',
|
|
},
|
|
)
|
|
|
|
|
|
class TestSmartDefaults:
|
|
def _template(self):
|
|
return {
|
|
"base": {
|
|
"retryMode": "standard",
|
|
"stsRegionalEndpoints": "regional",
|
|
"s3UsEast1RegionalEndpoints": "regional",
|
|
"connectTimeoutInMillis": 1000,
|
|
"tlsNegotiationTimeoutInMillis": 1000,
|
|
},
|
|
"modes": {
|
|
"standard": {
|
|
"connectTimeoutInMillis": {"multiply": 2},
|
|
"tlsNegotiationTimeoutInMillis": {"multiply": 2},
|
|
},
|
|
"in-region": {
|
|
"connectTimeoutInMillis": {"multiply": 1},
|
|
"tlsNegotiationTimeoutInMillis": {"multiply": 1},
|
|
},
|
|
"cross-region": {
|
|
"connectTimeoutInMillis": {"multiply": 2.8},
|
|
"tlsNegotiationTimeoutInMillis": {"multiply": 2.8},
|
|
},
|
|
"mobile": {
|
|
"connectTimeoutInMillis": {"override": 10000},
|
|
"tlsNegotiationTimeoutInMillis": {"add": 10000},
|
|
"retryMode": {"override": "adaptive"},
|
|
},
|
|
},
|
|
}
|
|
|
|
def _create_default_config_resolver(self):
|
|
return DefaultConfigResolver(self._template())
|
|
|
|
@pytest.fixture
|
|
def smart_defaults_factory(self):
|
|
fake_session = mock.Mock(spec=session.Session)
|
|
fake_session.get_scoped_config.return_value = {}
|
|
default_config_resolver = self._create_default_config_resolver()
|
|
return SmartDefaultsConfigStoreFactory(
|
|
default_config_resolver, imds_region_provider=mock.Mock()
|
|
)
|
|
|
|
@pytest.fixture
|
|
def fake_session(self):
|
|
fake_session = mock.Mock(spec=session.Session)
|
|
fake_session.get_scoped_config.return_value = {}
|
|
return fake_session
|
|
|
|
def _create_config_value_store(self, s3_mapping={}, **override_kwargs):
|
|
provider_foo = ConstantProvider(value='foo')
|
|
environment_provider_foo = EnvironmentProvider(
|
|
name='AWS_RETRY_MODE', env={'AWS_RETRY_MODE': None}
|
|
)
|
|
fake_session = mock.Mock(spec=session.Session)
|
|
fake_session.get_scoped_config.return_value = {}
|
|
# Testing with three different providers to validate
|
|
# SmartDefaultsConfigStoreFactory._get_new_chain_provider
|
|
mapping = {
|
|
'sts_regional_endpoints': ChainProvider(providers=[provider_foo]),
|
|
'retry_mode': ChainProvider(providers=[environment_provider_foo]),
|
|
's3': SectionConfigProvider('s3', fake_session, s3_mapping),
|
|
}
|
|
mapping.update(**override_kwargs)
|
|
config_store = ConfigValueStore(mapping=mapping)
|
|
return config_store
|
|
|
|
def _create_os_environ_patcher(self):
|
|
return mock.patch.object(
|
|
botocore.configprovider.os, 'environ', mock.Mock(wraps=os.environ)
|
|
)
|
|
|
|
def test_config_store_deepcopy(self):
|
|
config_store = ConfigValueStore()
|
|
config_store.set_config_provider('foo', ConstantProvider('bar'))
|
|
config_store_copy = copy.deepcopy(config_store)
|
|
config_store_copy.set_config_provider('fizz', ConstantProvider('buzz'))
|
|
assert config_store.get_config_variable('fizz') is None
|
|
assert config_store_copy.get_config_variable('foo') == 'bar'
|
|
|
|
@pytest.mark.parametrize(
|
|
'defaults_mode, retry_mode, sts_regional_endpoints,'
|
|
' us_east_1_regional_endpoint, connect_timeout',
|
|
[
|
|
('standard', 'standard', 'regional', 'regional', 2000),
|
|
('in-region', 'standard', 'regional', 'regional', 1000),
|
|
('cross-region', 'standard', 'regional', 'regional', 2800),
|
|
('mobile', 'adaptive', 'regional', 'regional', 10000),
|
|
],
|
|
)
|
|
def test_get_defualt_config_values(
|
|
self,
|
|
defaults_mode,
|
|
retry_mode,
|
|
sts_regional_endpoints,
|
|
us_east_1_regional_endpoint,
|
|
connect_timeout,
|
|
):
|
|
default_config_resolver = self._create_default_config_resolver()
|
|
default_values = default_config_resolver.get_default_config_values(
|
|
defaults_mode
|
|
)
|
|
assert default_values['retryMode'] == retry_mode
|
|
assert default_values['stsRegionalEndpoints'] == sts_regional_endpoints
|
|
assert (
|
|
default_values['s3UsEast1RegionalEndpoints']
|
|
== us_east_1_regional_endpoint
|
|
)
|
|
assert default_values['connectTimeoutInMillis'] == connect_timeout
|
|
|
|
def test_resolve_default_values_on_config(
|
|
self, smart_defaults_factory, fake_session
|
|
):
|
|
config_store = self._create_config_value_store()
|
|
smart_defaults_factory.merge_smart_defaults(
|
|
config_store, 'standard', 'foo'
|
|
)
|
|
s3_config = config_store.get_config_variable('s3')
|
|
assert s3_config['us_east_1_regional_endpoint'] == 'regional'
|
|
assert config_store.get_config_variable('retry_mode') == 'standard'
|
|
assert (
|
|
config_store.get_config_variable('sts_regional_endpoints')
|
|
== 'regional'
|
|
)
|
|
assert config_store.get_config_variable('connect_timeout') == 2
|
|
|
|
def test_no_resolve_default_s3_values_on_config(
|
|
self, smart_defaults_factory, fake_session
|
|
):
|
|
environment_provider = EnvironmentProvider(
|
|
name='AWS_S3_US_EAST_1_REGIONAL_ENDPOINT',
|
|
env={'AWS_S3_US_EAST_1_REGIONAL_ENDPOINT': 'legacy'},
|
|
)
|
|
s3_mapping = {
|
|
'us_east_1_regional_endpoint': ChainProvider(
|
|
providers=[environment_provider]
|
|
)
|
|
}
|
|
config_store = self._create_config_value_store(s3_mapping=s3_mapping)
|
|
smart_defaults_factory.merge_smart_defaults(
|
|
config_store, 'standard', 'foo'
|
|
)
|
|
s3_config = config_store.get_config_variable('s3')
|
|
assert s3_config['us_east_1_regional_endpoint'] == 'legacy'
|
|
assert config_store.get_config_variable('retry_mode') == 'standard'
|
|
assert (
|
|
config_store.get_config_variable('sts_regional_endpoints')
|
|
== 'regional'
|
|
)
|
|
assert config_store.get_config_variable('connect_timeout') == 2
|
|
|
|
def test_resolve_default_s3_values_on_config(
|
|
self, smart_defaults_factory, fake_session
|
|
):
|
|
s3_mapping = {
|
|
'use_arn_region': ChainProvider(
|
|
providers=[ConstantProvider(value=False)]
|
|
)
|
|
}
|
|
config_store = self._create_config_value_store(s3_mapping=s3_mapping)
|
|
smart_defaults_factory.merge_smart_defaults(
|
|
config_store, 'standard', 'foo'
|
|
)
|
|
s3_config = config_store.get_config_variable('s3')
|
|
assert s3_config['us_east_1_regional_endpoint'] == 'regional'
|
|
assert config_store.get_config_variable('retry_mode') == 'standard'
|
|
assert (
|
|
config_store.get_config_variable('sts_regional_endpoints')
|
|
== 'regional'
|
|
)
|
|
assert config_store.get_config_variable('connect_timeout') == 2
|
|
|
|
@pytest.mark.parametrize(
|
|
'execution_env_var, region_env_var, default_region_env_var, '
|
|
'imds_region, client_region, resolved_mode',
|
|
[
|
|
(
|
|
'AWS_Lambda_python3.6',
|
|
'us-east-1',
|
|
None,
|
|
None,
|
|
'us-east-1',
|
|
'in-region',
|
|
),
|
|
(
|
|
'AWS_Lambda_python3.6',
|
|
'us-west-2',
|
|
'us-west-2',
|
|
None,
|
|
'us-east-1',
|
|
'cross-region',
|
|
),
|
|
(
|
|
'AWS_Lambda_python3.6',
|
|
None,
|
|
None,
|
|
'us-west-2',
|
|
'us-east-1',
|
|
'cross-region',
|
|
),
|
|
(None, None, 'us-east-1', 'us-east-1', 'us-east-1', 'in-region'),
|
|
(None, None, None, 'us-west-2', 'us-east-1', 'cross-region'),
|
|
(None, None, None, None, 'us-west-2', 'standard'),
|
|
],
|
|
)
|
|
def test_resolve_auto_mode(
|
|
self,
|
|
execution_env_var,
|
|
region_env_var,
|
|
default_region_env_var,
|
|
imds_region,
|
|
client_region,
|
|
resolved_mode,
|
|
):
|
|
imds_region_provider = mock.Mock(spec=IMDSRegionProvider)
|
|
imds_region_provider.provide.return_value = imds_region
|
|
default_config_resolver = mock.Mock()
|
|
with mock.patch.object(
|
|
botocore.configprovider.os, 'environ', mock.Mock(wraps=os.environ)
|
|
) as os_environ_patcher:
|
|
os_environ_patcher.get.side_effect = [
|
|
execution_env_var,
|
|
default_region_env_var,
|
|
region_env_var,
|
|
]
|
|
smart_defaults_factory = SmartDefaultsConfigStoreFactory(
|
|
default_config_resolver, imds_region_provider
|
|
)
|
|
mode = smart_defaults_factory.resolve_auto_mode(client_region)
|
|
assert mode == resolved_mode
|
|
|
|
def test_resolve_auto_mode_imds_region_provider_connect_timeout(self):
|
|
imds_region_provider = mock.Mock(spec=IMDSRegionProvider)
|
|
imds_region_provider.provide.side_effect = ConnectTimeoutError(
|
|
endpoint_url='foo'
|
|
)
|
|
default_config_resolver = mock.Mock()
|
|
with mock.patch.object(
|
|
botocore.configprovider.os, 'environ', mock.Mock(wraps=os.environ)
|
|
) as os_environ_patcher:
|
|
os_environ_patcher.get.side_effect = [None] * 3
|
|
smart_defaults_factory = SmartDefaultsConfigStoreFactory(
|
|
default_config_resolver, imds_region_provider
|
|
)
|
|
mode = smart_defaults_factory.resolve_auto_mode('us-west-2')
|
|
assert mode == 'standard'
|