367 lines
16 KiB
Python
367 lines
16 KiB
Python
# Copyright 2020 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.
|
|
from botocore.compat import json
|
|
from botocore.config import Config
|
|
from botocore.discovery import EndpointDiscoveryRequired
|
|
from botocore.exceptions import (
|
|
ClientError,
|
|
InvalidEndpointDiscoveryConfigurationError,
|
|
)
|
|
from tests import ClientHTTPStubber, temporary_file
|
|
from tests.functional import FunctionalSessionTest
|
|
|
|
|
|
class TestEndpointDiscovery(FunctionalSessionTest):
|
|
def setUp(self):
|
|
super().setUp()
|
|
self.region = 'us-west-2'
|
|
|
|
def create_client(
|
|
self,
|
|
service_name='dynamodb',
|
|
region=None,
|
|
config=None,
|
|
endpoint_url=None,
|
|
):
|
|
if region is None:
|
|
region = self.region
|
|
client = self.session.create_client(
|
|
service_name, region, config=config, endpoint_url=endpoint_url
|
|
)
|
|
http_stubber = ClientHTTPStubber(client)
|
|
|
|
return client, http_stubber
|
|
|
|
def add_describe_endpoints_response(self, stubber, discovered_endpoint):
|
|
response = {
|
|
'Endpoints': [
|
|
{
|
|
'Address': discovered_endpoint,
|
|
'CachePeriodInMinutes': 1,
|
|
}
|
|
]
|
|
}
|
|
response_body = json.dumps(response).encode()
|
|
stubber.add_response(status=200, body=response_body)
|
|
stubber.add_response(status=200, body=b'{}')
|
|
|
|
def set_endpoint_discovery_config_file(self, fileobj, config_val):
|
|
fileobj.write(
|
|
'[default]\n' 'endpoint_discovery_enabled=%s\n' % config_val
|
|
)
|
|
fileobj.flush()
|
|
self.environ['AWS_CONFIG_FILE'] = fileobj.name
|
|
|
|
def assert_endpoint_discovery_used(self, stubber, discovered_endpoint):
|
|
self.assertEqual(len(stubber.requests), 2)
|
|
discover_request = stubber.requests[1]
|
|
self.assertEqual(discover_request.url, discovered_endpoint)
|
|
|
|
def assert_discovery_skipped(self, stubber, operation):
|
|
self.assertEqual(len(stubber.requests), 1)
|
|
self.assertEqual(
|
|
stubber.requests[0].headers.get('X-Amz-Target'), operation
|
|
)
|
|
|
|
def assert_endpoint_used(self, actual_url, expected_url):
|
|
self.assertEqual(actual_url, expected_url)
|
|
|
|
def test_endpoint_discovery_enabled(self):
|
|
discovered_endpoint = 'https://discovered.domain'
|
|
config = Config(endpoint_discovery_enabled=True)
|
|
client, http_stubber = self.create_client(config=config)
|
|
with http_stubber as stubber:
|
|
self.add_describe_endpoints_response(stubber, discovered_endpoint)
|
|
client.describe_table(TableName='sometable')
|
|
self.assert_endpoint_discovery_used(stubber, discovered_endpoint)
|
|
|
|
def test_endpoint_discovery_with_invalid_endpoint(self):
|
|
response = {
|
|
'Error': {
|
|
'Code': 'InvalidEndpointException',
|
|
'Message': 'Test Error',
|
|
}
|
|
}
|
|
response_body = json.dumps(response).encode()
|
|
|
|
config = Config(endpoint_discovery_enabled=True)
|
|
client, http_stubber = self.create_client(config=config)
|
|
with http_stubber as stubber:
|
|
stubber.add_response(status=421, body=response_body)
|
|
with self.assertRaises(ClientError):
|
|
client.describe_table(TableName='sometable')
|
|
|
|
def test_endpoint_discovery_disabled(self):
|
|
config = Config(endpoint_discovery_enabled=False)
|
|
client, http_stubber = self.create_client(config=config)
|
|
with http_stubber as stubber:
|
|
stubber.add_response(status=200, body=b'{}')
|
|
client.describe_table(TableName='sometable')
|
|
self.assertEqual(len(stubber.requests), 1)
|
|
|
|
def test_endpoint_discovery_no_config_default(self):
|
|
client, http_stubber = self.create_client()
|
|
with http_stubber as stubber:
|
|
stubber.add_response(status=200, body=b'{}')
|
|
client.describe_table(TableName='sometable')
|
|
self.assertEqual(len(stubber.requests), 1)
|
|
|
|
def test_endpoint_discovery_default_required_endpoint(self):
|
|
discovered_endpoint = "https://discovered.domain"
|
|
client, http_stubber = self.create_client(
|
|
service_name="test-discovery-endpoint"
|
|
)
|
|
with http_stubber as stubber:
|
|
self.add_describe_endpoints_response(stubber, discovered_endpoint)
|
|
client.test_discovery_required(Foo="bar")
|
|
self.assert_endpoint_discovery_used(stubber, discovered_endpoint)
|
|
|
|
def test_endpoint_discovery_required_with_discovery_enabled(self):
|
|
discovered_endpoint = "https://discovered.domain"
|
|
config = Config(endpoint_discovery_enabled=True)
|
|
client, http_stubber = self.create_client(
|
|
service_name="test-discovery-endpoint", config=config
|
|
)
|
|
with http_stubber as stubber:
|
|
self.add_describe_endpoints_response(stubber, discovered_endpoint)
|
|
client.test_discovery_required(Foo="bar")
|
|
self.assert_endpoint_discovery_used(stubber, discovered_endpoint)
|
|
|
|
def test_endpoint_discovery_required_with_discovery_disabled(self):
|
|
discovered_endpoint = "https://discovered.domain"
|
|
config = Config(endpoint_discovery_enabled=False)
|
|
client, http_stubber = self.create_client(
|
|
service_name="test-discovery-endpoint", config=config
|
|
)
|
|
self.add_describe_endpoints_response(http_stubber, discovered_endpoint)
|
|
with self.assertRaises(EndpointDiscoveryRequired):
|
|
client.test_discovery_required(Foo="bar")
|
|
|
|
def test_endpoint_discovery_required_with_custom_endpoint(self):
|
|
endpoint = "https://custom.domain/"
|
|
client, http_stubber = self.create_client(
|
|
service_name="test-discovery-endpoint", endpoint_url=endpoint
|
|
)
|
|
with http_stubber as stubber:
|
|
stubber.add_response(status=200, body=b'{}')
|
|
client.test_discovery_required(Foo="bar")
|
|
self.assert_discovery_skipped(
|
|
stubber, b"test-discovery-endpoint.TestDiscoveryRequired"
|
|
)
|
|
self.assert_endpoint_used(stubber.requests[0].url, endpoint)
|
|
|
|
def test_endpoint_discovery_disabled_with_custom_endpoint(self):
|
|
endpoint = "https://custom.domain/"
|
|
config = Config(endpoint_discovery_enabled=False)
|
|
client, http_stubber = self.create_client(
|
|
service_name="test-discovery-endpoint",
|
|
config=config,
|
|
endpoint_url=endpoint,
|
|
)
|
|
with http_stubber as stubber:
|
|
stubber.add_response(status=200, body=b'{}')
|
|
client.test_discovery_required(Foo="bar")
|
|
self.assert_discovery_skipped(
|
|
stubber, b"test-discovery-endpoint.TestDiscoveryRequired"
|
|
)
|
|
self.assert_endpoint_used(stubber.requests[0].url, endpoint)
|
|
|
|
def test_endpoint_discovery_enabled_with_custom_endpoint(self):
|
|
endpoint = "https://custom.domain/"
|
|
config = Config(endpoint_discovery_enabled=True)
|
|
client, http_stubber = self.create_client(
|
|
service_name="test-discovery-endpoint",
|
|
config=config,
|
|
endpoint_url=endpoint,
|
|
)
|
|
with http_stubber as stubber:
|
|
stubber.add_response(status=200, body=b'{}')
|
|
client.test_discovery_required(Foo="bar")
|
|
self.assert_discovery_skipped(
|
|
stubber, b"test-discovery-endpoint.TestDiscoveryRequired"
|
|
)
|
|
self.assert_endpoint_used(stubber.requests[0].url, endpoint)
|
|
|
|
def test_endpoint_discovery_optional_with_custom_endpoint(self):
|
|
endpoint = "https://custom.domain/"
|
|
client, http_stubber = self.create_client(
|
|
service_name="test-discovery-endpoint", endpoint_url=endpoint
|
|
)
|
|
with http_stubber as stubber:
|
|
stubber.add_response(status=200, body=b'{}')
|
|
client.test_discovery_optional(Foo="bar")
|
|
self.assert_discovery_skipped(
|
|
stubber, b"test-discovery-endpoint.TestDiscoveryOptional"
|
|
)
|
|
self.assert_endpoint_used(stubber.requests[0].url, endpoint)
|
|
|
|
def test_endpoint_discovery_optional_disabled_with_custom_endpoint(self):
|
|
endpoint = "https://custom.domain/"
|
|
config = Config(endpoint_discovery_enabled=False)
|
|
client, http_stubber = self.create_client(
|
|
service_name="test-discovery-endpoint",
|
|
config=config,
|
|
endpoint_url=endpoint,
|
|
)
|
|
with http_stubber as stubber:
|
|
stubber.add_response(status=200, body=b'{}')
|
|
client.test_discovery_optional(Foo="bar")
|
|
self.assert_discovery_skipped(
|
|
stubber,
|
|
b"test-discovery-endpoint.TestDiscoveryOptional",
|
|
)
|
|
self.assert_endpoint_used(stubber.requests[0].url, endpoint)
|
|
|
|
def test_endpoint_discovery_default_optional_endpoint(self):
|
|
client, http_stubber = self.create_client(
|
|
service_name="test-discovery-endpoint"
|
|
)
|
|
with http_stubber as stubber:
|
|
stubber.add_response(status=200, body=b'{}')
|
|
client.test_discovery_optional(Foo="bar")
|
|
self.assertEqual(len(stubber.requests), 1)
|
|
|
|
def test_endpoint_discovery_enabled_optional_endpoint(self):
|
|
discovered_endpoint = 'https://discovered.domain'
|
|
config = Config(endpoint_discovery_enabled=True)
|
|
client, http_stubber = self.create_client(
|
|
service_name="test-discovery-endpoint", config=config
|
|
)
|
|
with http_stubber as stubber:
|
|
self.add_describe_endpoints_response(stubber, discovered_endpoint)
|
|
client.test_discovery_optional(Foo="bar")
|
|
self.assert_endpoint_discovery_used(stubber, discovered_endpoint)
|
|
|
|
def test_endpoint_discovery_manual_auto_on_required_endpoint(self):
|
|
discovered_endpoint = 'https://discovered.domain'
|
|
config = Config(endpoint_discovery_enabled=" aUto \n")
|
|
client, http_stubber = self.create_client(
|
|
service_name="test-discovery-endpoint", config=config
|
|
)
|
|
with http_stubber as stubber:
|
|
self.add_describe_endpoints_response(stubber, discovered_endpoint)
|
|
client.test_discovery_required(Foo="bar")
|
|
self.assert_endpoint_discovery_used(stubber, discovered_endpoint)
|
|
|
|
def test_endpoint_discovery_enabled_with_random_string(self):
|
|
config = Config(endpoint_discovery_enabled="bad value")
|
|
with self.assertRaises(InvalidEndpointDiscoveryConfigurationError):
|
|
client, http_stubber = self.create_client(
|
|
service_name="test-discovery-endpoint", config=config
|
|
)
|
|
|
|
def test_endpoint_discovery_required_with_env_var_default(self):
|
|
self.environ['AWS_ENDPOINT_DISCOVERY_ENABLED'] = 'auto'
|
|
discovered_endpoint = 'https://discovered.domain'
|
|
client, http_stubber = self.create_client(
|
|
service_name="test-discovery-endpoint", config=None
|
|
)
|
|
with http_stubber as stubber:
|
|
self.add_describe_endpoints_response(stubber, discovered_endpoint)
|
|
client.test_discovery_required(Foo="bar")
|
|
self.assert_endpoint_discovery_used(stubber, discovered_endpoint)
|
|
|
|
def test_endpoint_discovery_optional_with_env_var_default(self):
|
|
self.environ['AWS_ENDPOINT_DISCOVERY_ENABLED'] = 'auto'
|
|
client, http_stubber = self.create_client(
|
|
service_name="test-discovery-endpoint", config=None
|
|
)
|
|
with http_stubber as stubber:
|
|
stubber.add_response(status=200, body=b'{}')
|
|
client.test_discovery_optional(Foo="bar")
|
|
self.assert_discovery_skipped(
|
|
stubber, b"test-discovery-endpoint.TestDiscoveryOptional"
|
|
)
|
|
|
|
def test_endpoint_discovery_optional_with_env_var_enabled(self):
|
|
self.environ['AWS_ENDPOINT_DISCOVERY_ENABLED'] = "True"
|
|
discovered_endpoint = 'https://discovered.domain'
|
|
client, http_stubber = self.create_client(
|
|
service_name="test-discovery-endpoint"
|
|
)
|
|
with http_stubber as stubber:
|
|
self.add_describe_endpoints_response(stubber, discovered_endpoint)
|
|
client.test_discovery_optional(Foo="bar")
|
|
self.assert_endpoint_discovery_used(stubber, discovered_endpoint)
|
|
|
|
def test_endpoint_discovery_required_with_env_var_disabled(self):
|
|
self.environ['AWS_ENDPOINT_DISCOVERY_ENABLED'] = "False"
|
|
discovered_endpoint = 'https://discovered.domain'
|
|
client, http_stubber = self.create_client(
|
|
service_name="test-discovery-endpoint"
|
|
)
|
|
self.add_describe_endpoints_response(http_stubber, discovered_endpoint)
|
|
with self.assertRaises(EndpointDiscoveryRequired):
|
|
client.test_discovery_required(Foo="bar")
|
|
|
|
def test_endpoint_discovery_with_config_file_enabled(self):
|
|
with temporary_file('w') as f:
|
|
self.set_endpoint_discovery_config_file(f, "True")
|
|
discovered_endpoint = 'https://discovered.domain'
|
|
client, http_stubber = self.create_client(
|
|
service_name="test-discovery-endpoint"
|
|
)
|
|
with http_stubber as stubber:
|
|
self.add_describe_endpoints_response(
|
|
stubber, discovered_endpoint
|
|
)
|
|
client.test_discovery_required(Foo="bar")
|
|
self.assert_endpoint_discovery_used(
|
|
stubber, discovered_endpoint
|
|
)
|
|
|
|
def test_endpoint_discovery_with_config_file_enabled_lowercase(self):
|
|
with temporary_file('w') as f:
|
|
self.set_endpoint_discovery_config_file(f, "true")
|
|
discovered_endpoint = 'https://discovered.domain'
|
|
client, http_stubber = self.create_client(
|
|
service_name="test-discovery-endpoint"
|
|
)
|
|
with http_stubber as stubber:
|
|
self.add_describe_endpoints_response(
|
|
stubber, discovered_endpoint
|
|
)
|
|
client.test_discovery_required(Foo="bar")
|
|
self.assert_endpoint_discovery_used(
|
|
stubber, discovered_endpoint
|
|
)
|
|
|
|
def test_endpoint_discovery_with_config_file_disabled(self):
|
|
with temporary_file('w') as f:
|
|
self.set_endpoint_discovery_config_file(f, "false")
|
|
discovered_endpoint = 'https://discovered.domain'
|
|
client, http_stubber = self.create_client(
|
|
service_name="test-discovery-endpoint"
|
|
)
|
|
self.add_describe_endpoints_response(
|
|
http_stubber, discovered_endpoint
|
|
)
|
|
with self.assertRaises(EndpointDiscoveryRequired):
|
|
client.test_discovery_required(Foo="bar")
|
|
|
|
def test_endpoint_discovery_with_config_file_auto(self):
|
|
with temporary_file('w') as f:
|
|
self.set_endpoint_discovery_config_file(f, "AUTO")
|
|
discovered_endpoint = 'https://discovered.domain'
|
|
client, http_stubber = self.create_client(
|
|
service_name="test-discovery-endpoint"
|
|
)
|
|
with http_stubber as stubber:
|
|
self.add_describe_endpoints_response(
|
|
stubber, discovered_endpoint
|
|
)
|
|
client.test_discovery_required(Foo="bar")
|
|
self.assert_endpoint_discovery_used(
|
|
stubber, discovered_endpoint
|
|
)
|