1011 lines
38 KiB
Python
1011 lines
38 KiB
Python
import pytest
|
|
|
|
from tests import unittest
|
|
|
|
from botocore import model
|
|
from botocore.compat import OrderedDict
|
|
from botocore.exceptions import MissingServiceIdError
|
|
|
|
|
|
@pytest.mark.parametrize("property_name", ['api_version', 'protocol'])
|
|
def test_missing_model_attribute_raises_exception(property_name):
|
|
service_model = model.ServiceModel({'metadata': {'endpointPrefix': 'foo'}})
|
|
with pytest.raises(model.UndefinedModelAttributeError):
|
|
getattr(service_model, property_name)
|
|
|
|
|
|
class TestServiceId(unittest.TestCase):
|
|
def test_hypenize_replaces_spaces(self):
|
|
self.assertEqual(
|
|
model.ServiceId('my service').hyphenize(), 'my-service'
|
|
)
|
|
|
|
def test_hyphenize_lower_cases(self):
|
|
self.assertEqual(model.ServiceId('MyService').hyphenize(), 'myservice')
|
|
|
|
|
|
class TestServiceModel(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.model = {
|
|
'metadata': {'protocol': 'query',
|
|
'endpointPrefix': 'endpoint-prefix',
|
|
'serviceId': 'MyService'},
|
|
'documentation': 'Documentation value',
|
|
'operations': {},
|
|
'shapes': {
|
|
'StringShape': {'type': 'string'}
|
|
}
|
|
}
|
|
self.error_shapes = {
|
|
'ExceptionOne': {
|
|
'exception': True,
|
|
'type': 'structure',
|
|
'members': {},
|
|
},
|
|
'ExceptionTwo': {
|
|
'exception': True,
|
|
'type': 'structure',
|
|
'members': {},
|
|
'error': {
|
|
'code': 'FooCode'
|
|
}
|
|
},
|
|
}
|
|
self.service_model = model.ServiceModel(self.model)
|
|
|
|
def test_metadata_available(self):
|
|
# You should be able to access the metadata in a service description
|
|
# through the service model object.
|
|
self.assertEqual(self.service_model.metadata.get('protocol'), 'query')
|
|
|
|
def test_service_name_can_be_overriden(self):
|
|
service_model = model.ServiceModel(self.model,
|
|
service_name='myservice')
|
|
self.assertEqual(service_model.service_name, 'myservice')
|
|
|
|
def test_service_name_defaults_to_endpoint_prefix(self):
|
|
self.assertEqual(self.service_model.service_name, 'endpoint-prefix')
|
|
|
|
def test_service_id(self):
|
|
self.assertEqual(self.service_model.service_id, 'MyService')
|
|
|
|
def test_hyphenize_service_id(self):
|
|
self.assertEqual(
|
|
self.service_model.service_id.hyphenize(), 'myservice')
|
|
|
|
def test_service_id_does_not_exist(self):
|
|
service_model = {
|
|
'metadata': {
|
|
'protocol': 'query',
|
|
'endpointPrefix': 'endpoint-prefix',
|
|
},
|
|
'documentation': 'Documentation value',
|
|
'operations': {},
|
|
'shapes': {
|
|
'StringShape': {'type': 'string'}
|
|
}
|
|
}
|
|
service_name = 'myservice'
|
|
service_model = model.ServiceModel(service_model, service_name)
|
|
with self.assertRaisesRegex(model.UndefinedModelAttributeError,
|
|
service_name):
|
|
service_model.service_id
|
|
|
|
def test_operation_does_not_exist(self):
|
|
with self.assertRaises(model.OperationNotFoundError):
|
|
self.service_model.operation_model('NoExistOperation')
|
|
|
|
def test_signing_name_defaults_to_endpoint_prefix(self):
|
|
self.assertEqual(self.service_model.signing_name, 'endpoint-prefix')
|
|
|
|
def test_documentation_exposed_as_property(self):
|
|
self.assertEqual(self.service_model.documentation,
|
|
'Documentation value')
|
|
|
|
def test_shape_names(self):
|
|
self.assertEqual(self.service_model.shape_names, ['StringShape'])
|
|
|
|
def test_repr_has_service_name(self):
|
|
self.assertEqual(repr(self.service_model),
|
|
'ServiceModel(endpoint-prefix)')
|
|
|
|
def test_shape_for_error_code(self):
|
|
self.model['shapes'].update(self.error_shapes)
|
|
self.service_model = model.ServiceModel(self.model)
|
|
shape = self.service_model.shape_for_error_code('ExceptionOne')
|
|
self.assertEqual(shape.name, 'ExceptionOne')
|
|
shape = self.service_model.shape_for_error_code('FooCode')
|
|
self.assertEqual(shape.name, 'ExceptionTwo')
|
|
|
|
def test_error_shapes(self):
|
|
self.model['shapes'].update(self.error_shapes)
|
|
self.service_model = model.ServiceModel(self.model)
|
|
error_shapes = self.service_model.error_shapes
|
|
error_shape_names = [shape.name for shape in error_shapes]
|
|
self.assertEqual(len(error_shape_names), 2)
|
|
self.assertIn('ExceptionOne', error_shape_names)
|
|
self.assertIn('ExceptionTwo', error_shape_names)
|
|
|
|
|
|
class TestOperationModelFromService(unittest.TestCase):
|
|
def setUp(self):
|
|
self.model = {
|
|
'metadata': {'protocol': 'query', 'endpointPrefix': 'foo'},
|
|
'documentation': '',
|
|
'operations': {
|
|
'OperationName': {
|
|
'http': {
|
|
'method': 'POST',
|
|
'requestUri': '/',
|
|
},
|
|
'name': 'OperationName',
|
|
'input': {
|
|
'shape': 'OperationNameRequest'
|
|
},
|
|
'output': {
|
|
'shape': 'OperationNameResponse',
|
|
},
|
|
'errors': [{'shape': 'NoSuchResourceException'}],
|
|
'documentation': 'Docs for OperationName',
|
|
'authtype': 'v4'
|
|
},
|
|
'OperationTwo': {
|
|
'http': {
|
|
'method': 'POST',
|
|
'requestUri': '/',
|
|
},
|
|
'name': 'OperationTwo',
|
|
'input': {
|
|
'shape': 'OperationNameRequest'
|
|
},
|
|
'output': {
|
|
'shape': 'OperationNameResponse',
|
|
},
|
|
'errors': [{'shape': 'NoSuchResourceException'}],
|
|
'documentation': 'Docs for OperationTwo',
|
|
}
|
|
},
|
|
'shapes': {
|
|
'OperationNameRequest': {
|
|
'type': 'structure',
|
|
'members': {
|
|
'Arg1': {'shape': 'stringType'},
|
|
'Arg2': {'shape': 'stringType'},
|
|
}
|
|
},
|
|
'OperationNameResponse': {
|
|
'type': 'structure',
|
|
'members': {
|
|
'String': {
|
|
'shape': 'stringType',
|
|
}
|
|
}
|
|
},
|
|
'NoSuchResourceException': {
|
|
'type': 'structure',
|
|
'members': {}
|
|
},
|
|
'stringType': {
|
|
'type': 'string',
|
|
}
|
|
}
|
|
}
|
|
self.service_model = model.ServiceModel(self.model)
|
|
|
|
def test_wire_name_always_matches_model(self):
|
|
service_model = model.ServiceModel(self.model)
|
|
operation = model.OperationModel(
|
|
self.model['operations']['OperationName'], service_model, 'Foo')
|
|
self.assertEqual(operation.name, 'Foo')
|
|
self.assertEqual(operation.wire_name, 'OperationName')
|
|
|
|
def test_operation_name_in_repr(self):
|
|
service_model = model.ServiceModel(self.model)
|
|
operation = service_model.operation_model('OperationName')
|
|
self.assertIn('OperationName', repr(operation))
|
|
|
|
def test_name_and_wire_name_defaults_to_same_value(self):
|
|
service_model = model.ServiceModel(self.model)
|
|
operation = model.OperationModel(
|
|
self.model['operations']['OperationName'], service_model)
|
|
self.assertEqual(operation.name, 'OperationName')
|
|
self.assertEqual(operation.wire_name, 'OperationName')
|
|
|
|
def test_name_from_service(self):
|
|
service_model = model.ServiceModel(self.model)
|
|
operation = service_model.operation_model('OperationName')
|
|
self.assertEqual(operation.name, 'OperationName')
|
|
|
|
def test_name_from_service_model_when_differs_from_name(self):
|
|
self.model['operations']['Foo'] = \
|
|
self.model['operations']['OperationName']
|
|
service_model = model.ServiceModel(self.model)
|
|
operation = service_model.operation_model('Foo')
|
|
self.assertEqual(operation.name, 'Foo')
|
|
|
|
def test_operation_input_model(self):
|
|
service_model = model.ServiceModel(self.model)
|
|
operation = service_model.operation_model('OperationName')
|
|
self.assertEqual(operation.name, 'OperationName')
|
|
# Operations should also have a reference to the top level metadata.
|
|
self.assertEqual(operation.metadata['protocol'], 'query')
|
|
self.assertEqual(operation.http['method'], 'POST')
|
|
self.assertEqual(operation.http['requestUri'], '/')
|
|
shape = operation.input_shape
|
|
self.assertEqual(shape.name, 'OperationNameRequest')
|
|
self.assertEqual(list(sorted(shape.members)), ['Arg1', 'Arg2'])
|
|
|
|
def test_has_documentation_property(self):
|
|
service_model = model.ServiceModel(self.model)
|
|
operation = service_model.operation_model('OperationName')
|
|
self.assertEqual(operation.documentation, 'Docs for OperationName')
|
|
|
|
def test_service_model_available_from_operation_model(self):
|
|
service_model = model.ServiceModel(self.model)
|
|
operation = service_model.operation_model('OperationName')
|
|
# This is an identity comparison because we don't implement
|
|
# __eq__, so we may need to change this in the future.
|
|
self.assertEqual(
|
|
operation.service_model, service_model)
|
|
|
|
def test_operation_output_model(self):
|
|
service_model = model.ServiceModel(self.model)
|
|
operation = service_model.operation_model('OperationName')
|
|
output = operation.output_shape
|
|
self.assertEqual(list(output.members), ['String'])
|
|
self.assertFalse(operation.has_streaming_output)
|
|
|
|
def test_operation_shape_not_required(self):
|
|
# It's ok if there's no output shape. We'll just get a return value of
|
|
# None.
|
|
del self.model['operations']['OperationName']['output']
|
|
service_model = model.ServiceModel(self.model)
|
|
operation = service_model.operation_model('OperationName')
|
|
output_shape = operation.output_shape
|
|
self.assertIsNone(output_shape)
|
|
|
|
def test_error_shapes(self):
|
|
service_model = model.ServiceModel(self.model)
|
|
operation = service_model.operation_model('OperationName')
|
|
# OperationName only has a NoSuchResourceException
|
|
self.assertEqual(len(operation.error_shapes), 1)
|
|
self.assertEqual(
|
|
operation.error_shapes[0].name, 'NoSuchResourceException')
|
|
|
|
def test_has_auth_type(self):
|
|
operation = self.service_model.operation_model('OperationName')
|
|
self.assertEqual(operation.auth_type, 'v4')
|
|
|
|
def test_auth_type_not_set(self):
|
|
operation = self.service_model.operation_model('OperationTwo')
|
|
self.assertIsNone(operation.auth_type)
|
|
|
|
def test_deprecated_present(self):
|
|
self.model['operations']['OperationName']['deprecated'] = True
|
|
service_model = model.ServiceModel(self.model)
|
|
operation_name = service_model.operation_model('OperationName')
|
|
self.assertTrue(operation_name.deprecated)
|
|
|
|
def test_deprecated_present_false(self):
|
|
self.model['operations']['OperationName']['deprecated'] = False
|
|
service_model = model.ServiceModel(self.model)
|
|
operation_name = service_model.operation_model('OperationName')
|
|
self.assertFalse(operation_name.deprecated)
|
|
|
|
def test_deprecated_absent(self):
|
|
service_model = model.ServiceModel(self.model)
|
|
operation_two = service_model.operation_model('OperationTwo')
|
|
self.assertFalse(operation_two.deprecated)
|
|
|
|
def test_endpoint_operation_present(self):
|
|
self.model['operations']['OperationName']['endpointoperation'] = True
|
|
service_model = model.ServiceModel(self.model)
|
|
operation_name = service_model.operation_model('OperationName')
|
|
self.assertTrue(operation_name.is_endpoint_discovery_operation)
|
|
|
|
def test_endpoint_operation_present_false(self):
|
|
self.model['operations']['OperationName']['endpointoperation'] = False
|
|
service_model = model.ServiceModel(self.model)
|
|
operation_name = service_model.operation_model('OperationName')
|
|
self.assertFalse(operation_name.is_endpoint_discovery_operation)
|
|
|
|
def test_endpoint_operation_absent(self):
|
|
operation_two = self.service_model.operation_model('OperationName')
|
|
self.assertFalse(operation_two.is_endpoint_discovery_operation)
|
|
|
|
def test_endpoint_discovery_required(self):
|
|
operation = self.model['operations']['OperationName']
|
|
operation['endpointdiscovery'] = {'required': True}
|
|
service_model = model.ServiceModel(self.model)
|
|
self.assertTrue(service_model.endpoint_discovery_required)
|
|
|
|
def test_endpoint_discovery_required_false(self):
|
|
self.model['operations']['OperationName']['endpointdiscovery'] = {}
|
|
service_model = model.ServiceModel(self.model)
|
|
self.assertFalse(service_model.endpoint_discovery_required)
|
|
|
|
def test_endpoint_discovery_required_no_value(self):
|
|
operation = self.model['operations']['OperationName']
|
|
self.assertTrue(operation.get('endpointdiscovery') is None)
|
|
service_model = model.ServiceModel(self.model)
|
|
self.assertFalse(service_model.endpoint_discovery_required)
|
|
|
|
def test_endpoint_discovery_present(self):
|
|
operation = self.model['operations']['OperationName']
|
|
operation['endpointdiscovery'] = {'required': True}
|
|
service_model = model.ServiceModel(self.model)
|
|
operation_name = service_model.operation_model('OperationName')
|
|
self.assertTrue(operation_name.endpoint_discovery.get('required'))
|
|
|
|
def test_endpoint_discovery_absent(self):
|
|
operation_name = self.service_model.operation_model('OperationName')
|
|
self.assertIsNone(operation_name.endpoint_discovery)
|
|
|
|
|
|
class TestOperationModelEventStreamTypes(unittest.TestCase):
|
|
def setUp(self):
|
|
super(TestOperationModelEventStreamTypes, self).setUp()
|
|
self.model = {
|
|
'metadata': {'protocol': 'rest-xml', 'endpointPrefix': 'foo'},
|
|
'documentation': '',
|
|
'operations': {
|
|
'OperationName': {
|
|
'http': {
|
|
'method': 'POST',
|
|
'requestUri': '/',
|
|
},
|
|
'name': 'OperationName',
|
|
'input': {'shape': 'OperationRequest'},
|
|
'output': {'shape': 'OperationResponse'},
|
|
}
|
|
},
|
|
'shapes': {
|
|
'NormalStructure': {
|
|
'type': 'structure',
|
|
'members': {
|
|
'Input': {'shape': 'StringType'}
|
|
}
|
|
},
|
|
'OperationRequest': {
|
|
'type': 'structure',
|
|
'members': {
|
|
'String': {'shape': 'StringType'},
|
|
"Body": {'shape': 'EventStreamStructure'}
|
|
},
|
|
'payload': 'Body'
|
|
},
|
|
'OperationResponse': {
|
|
'type': 'structure',
|
|
'members': {
|
|
'String': {'shape': 'StringType'},
|
|
"Body": {'shape': 'EventStreamStructure'}
|
|
},
|
|
'payload': 'Body'
|
|
},
|
|
'StringType': {'type': 'string'},
|
|
'BlobType': {'type': 'blob'},
|
|
'EventStreamStructure': {
|
|
'eventstream': True,
|
|
'type': 'structure',
|
|
'members': {
|
|
'EventA': {'shape': 'EventAStructure'},
|
|
'EventB': {'shape': 'EventBStructure'}
|
|
}
|
|
},
|
|
'EventAStructure': {
|
|
'event': True,
|
|
'type': 'structure',
|
|
'members': {
|
|
'Payload': {
|
|
'shape': 'BlobType',
|
|
'eventpayload': True
|
|
},
|
|
'Header': {
|
|
'shape': 'StringType',
|
|
'eventheader': True
|
|
}
|
|
}
|
|
},
|
|
'EventBStructure': {
|
|
'event': True,
|
|
'type': 'structure',
|
|
'members': {
|
|
'Records': {'shape': 'StringType'}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
def update_operation(self, **kwargs):
|
|
operation = self.model['operations']['OperationName']
|
|
operation.update(kwargs)
|
|
|
|
def test_event_stream_input_for_operation(self):
|
|
service_model = model.ServiceModel(self.model)
|
|
operation = service_model.operation_model('OperationName')
|
|
self.assertTrue(operation.has_event_stream_input)
|
|
event_stream_input = operation.get_event_stream_input()
|
|
self.assertEqual(event_stream_input.name, 'EventStreamStructure')
|
|
|
|
def test_no_event_stream_input_for_operation(self):
|
|
self.update_operation(input={'shape': 'NormalStructure'})
|
|
service_model = model.ServiceModel(self.model)
|
|
operation = service_model.operation_model('OperationName')
|
|
self.assertFalse(operation.has_event_stream_input)
|
|
self.assertEqual(operation.get_event_stream_input(), None)
|
|
|
|
def test_event_stream_output_for_operation(self):
|
|
service_model = model.ServiceModel(self.model)
|
|
operation = service_model.operation_model('OperationName')
|
|
self.assertTrue(operation.has_event_stream_output)
|
|
output = operation.get_event_stream_output()
|
|
self.assertEqual(output.name, 'EventStreamStructure')
|
|
|
|
def test_no_event_stream_output_for_operation(self):
|
|
self.update_operation(output={'shape': 'NormalStructure'})
|
|
service_model = model.ServiceModel(self.model)
|
|
operation = service_model.operation_model('OperationName')
|
|
self.assertFalse(operation.has_event_stream_output)
|
|
self.assertEqual(operation.get_event_stream_output(), None)
|
|
|
|
def test_no_output_shape(self):
|
|
self.update_operation(output=None)
|
|
del self.model['operations']['OperationName']['output']
|
|
service_model = model.ServiceModel(self.model)
|
|
operation = service_model.operation_model('OperationName')
|
|
self.assertFalse(operation.has_event_stream_output)
|
|
self.assertEqual(operation.get_event_stream_output(), None)
|
|
|
|
|
|
class TestOperationModelStreamingTypes(unittest.TestCase):
|
|
def setUp(self):
|
|
super(TestOperationModelStreamingTypes, self).setUp()
|
|
self.model = {
|
|
'metadata': {'protocol': 'query', 'endpointPrefix': 'foo'},
|
|
'documentation': '',
|
|
'operations': {
|
|
'OperationName': {
|
|
'name': 'OperationName',
|
|
'input': {
|
|
'shape': 'OperationRequest',
|
|
},
|
|
'output': {
|
|
'shape': 'OperationResponse',
|
|
},
|
|
}
|
|
},
|
|
'shapes': {
|
|
'OperationRequest': {
|
|
'type': 'structure',
|
|
'members': {
|
|
'String': {
|
|
'shape': 'stringType',
|
|
},
|
|
"Body": {
|
|
'shape': 'blobType',
|
|
}
|
|
},
|
|
'payload': 'Body'
|
|
},
|
|
'OperationResponse': {
|
|
'type': 'structure',
|
|
'members': {
|
|
'String': {
|
|
'shape': 'stringType',
|
|
},
|
|
"Body": {
|
|
'shape': 'blobType',
|
|
}
|
|
},
|
|
'payload': 'Body'
|
|
},
|
|
'stringType': {
|
|
'type': 'string',
|
|
},
|
|
'blobType': {
|
|
'type': 'blob'
|
|
}
|
|
}
|
|
}
|
|
|
|
def remove_payload(self, type):
|
|
self.model['shapes']['Operation' + type].pop('payload')
|
|
|
|
def test_streaming_input_for_operation(self):
|
|
service_model = model.ServiceModel(self.model)
|
|
operation = service_model.operation_model('OperationName')
|
|
self.assertTrue(operation.has_streaming_input)
|
|
self.assertEqual(operation.get_streaming_input().name, 'blobType')
|
|
|
|
def test_not_streaming_input_for_operation(self):
|
|
self.remove_payload('Request')
|
|
service_model = model.ServiceModel(self.model)
|
|
operation = service_model.operation_model('OperationName')
|
|
self.assertFalse(operation.has_streaming_input)
|
|
self.assertEqual(operation.get_streaming_input(), None)
|
|
|
|
def test_streaming_output_for_operation(self):
|
|
service_model = model.ServiceModel(self.model)
|
|
operation = service_model.operation_model('OperationName')
|
|
self.assertTrue(operation.has_streaming_output)
|
|
self.assertEqual(operation.get_streaming_output().name, 'blobType')
|
|
|
|
def test_not_streaming_output_for_operation(self):
|
|
self.remove_payload('Response')
|
|
service_model = model.ServiceModel(self.model)
|
|
operation = service_model.operation_model('OperationName')
|
|
self.assertFalse(operation.has_streaming_output)
|
|
self.assertEqual(operation.get_streaming_output(), None)
|
|
|
|
|
|
class TestDeepMerge(unittest.TestCase):
|
|
def setUp(self):
|
|
self.shapes = {
|
|
'SetQueueAttributes': {
|
|
'type': 'structure',
|
|
'members': {
|
|
'MapExample': {'shape': 'StrToStrMap',
|
|
'locationName': 'Attribute'},
|
|
}
|
|
},
|
|
'SetQueueAttributes2': {
|
|
'type': 'structure',
|
|
'members': {
|
|
'MapExample': {'shape': 'StrToStrMap',
|
|
'locationName': 'Attribute2'},
|
|
}
|
|
},
|
|
'StrToStrMap': {
|
|
'type': 'map',
|
|
'key': {'shape': 'StringType', 'locationName': 'Name'},
|
|
'value': {'shape': 'StringType', 'locationName': 'Value'},
|
|
'flattened': True,
|
|
'name': 'NotAttribute',
|
|
},
|
|
'StringType': {'type': 'string'}
|
|
}
|
|
self.shape_resolver = model.ShapeResolver(self.shapes)
|
|
|
|
def test_deep_merge(self):
|
|
shape = self.shape_resolver.get_shape_by_name('SetQueueAttributes')
|
|
map_merged = shape.members['MapExample']
|
|
# map_merged has a serialization as a member trait as well as
|
|
# in the StrToStrMap.
|
|
# The member trait should have precedence.
|
|
self.assertEqual(map_merged.serialization,
|
|
# member beats the definition.
|
|
{'name': 'Attribute',
|
|
# From the definition.
|
|
'flattened': True,})
|
|
# Ensure we don't merge/mutate the original dicts.
|
|
self.assertEqual(map_merged.key.serialization['name'], 'Name')
|
|
self.assertEqual(map_merged.value.serialization['name'], 'Value')
|
|
self.assertEqual(map_merged.key.serialization['name'], 'Name')
|
|
|
|
def test_merges_copy_dict(self):
|
|
shape = self.shape_resolver.get_shape_by_name('SetQueueAttributes')
|
|
map_merged = shape.members['MapExample']
|
|
self.assertEqual(map_merged.serialization.get('name'), 'Attribute')
|
|
|
|
shape2 = self.shape_resolver.get_shape_by_name('SetQueueAttributes2')
|
|
map_merged2 = shape2.members['MapExample']
|
|
self.assertEqual(map_merged2.serialization.get('name'), 'Attribute2')
|
|
|
|
|
|
class TestShapeResolver(unittest.TestCase):
|
|
def test_get_shape_by_name(self):
|
|
shape_map = {
|
|
'Foo': {
|
|
'type': 'structure',
|
|
'members': {
|
|
'Bar': {'shape': 'StringType'},
|
|
'Baz': {'shape': 'StringType'},
|
|
}
|
|
},
|
|
"StringType": {
|
|
"type": "string"
|
|
}
|
|
}
|
|
resolver = model.ShapeResolver(shape_map)
|
|
shape = resolver.get_shape_by_name('Foo')
|
|
self.assertEqual(shape.name, 'Foo')
|
|
self.assertEqual(shape.type_name, 'structure')
|
|
|
|
def test_resolve_shape_reference(self):
|
|
shape_map = {
|
|
'Foo': {
|
|
'type': 'structure',
|
|
'members': {
|
|
'Bar': {'shape': 'StringType'},
|
|
'Baz': {'shape': 'StringType'},
|
|
}
|
|
},
|
|
"StringType": {
|
|
"type": "string"
|
|
}
|
|
}
|
|
resolver = model.ShapeResolver(shape_map)
|
|
shape = resolver.resolve_shape_ref({'shape': 'StringType'})
|
|
self.assertEqual(shape.name, 'StringType')
|
|
self.assertEqual(shape.type_name, 'string')
|
|
|
|
def test_resolve_shape_references_with_member_traits(self):
|
|
shape_map = {
|
|
'Foo': {
|
|
'type': 'structure',
|
|
'members': {
|
|
'Bar': {'shape': 'StringType'},
|
|
'Baz': {'shape': 'StringType', 'locationName': 'other'},
|
|
}
|
|
},
|
|
"StringType": {
|
|
"type": "string"
|
|
}
|
|
}
|
|
resolver = model.ShapeResolver(shape_map)
|
|
shape = resolver.resolve_shape_ref({'shape': 'StringType',
|
|
'locationName': 'other'})
|
|
self.assertEqual(shape.serialization['name'], 'other')
|
|
self.assertEqual(shape.name, 'StringType')
|
|
|
|
def test_serialization_cache(self):
|
|
shape_map = {
|
|
'Foo': {
|
|
'type': 'structure',
|
|
'members': {
|
|
'Baz': {'shape': 'StringType', 'locationName': 'other'},
|
|
}
|
|
},
|
|
"StringType": {
|
|
"type": "string"
|
|
}
|
|
}
|
|
resolver = model.ShapeResolver(shape_map)
|
|
shape = resolver.resolve_shape_ref({'shape': 'StringType',
|
|
'locationName': 'other'})
|
|
self.assertEqual(shape.serialization['name'], 'other')
|
|
# serialization is computed on demand, and a cache is kept.
|
|
# This is just verifying that trying to access serialization again
|
|
# gives the same result. We don't actually care that it's cached,
|
|
# we just care that the cache doesn't mess with correctness.
|
|
self.assertEqual(shape.serialization['name'], 'other')
|
|
|
|
def test_shape_overrides(self):
|
|
shape_map = {
|
|
"StringType": {
|
|
"type": "string",
|
|
"documentation": "Original documentation"
|
|
}
|
|
}
|
|
resolver = model.ShapeResolver(shape_map)
|
|
shape = resolver.get_shape_by_name('StringType')
|
|
self.assertEqual(shape.documentation, 'Original documentation')
|
|
|
|
shape = resolver.resolve_shape_ref({'shape': 'StringType',
|
|
'documentation': 'override'})
|
|
self.assertEqual(shape.documentation, 'override')
|
|
|
|
def test_shape_type_structure(self):
|
|
shapes = {
|
|
'ChangePasswordRequest': {
|
|
'type': 'structure',
|
|
'members': {
|
|
'OldPassword': {'shape': 'passwordType'},
|
|
'NewPassword': {'shape': 'passwordType'},
|
|
}
|
|
},
|
|
'passwordType': {
|
|
"type":"string",
|
|
}
|
|
}
|
|
resolver = model.ShapeResolver(shapes)
|
|
shape = resolver.get_shape_by_name('ChangePasswordRequest')
|
|
self.assertEqual(shape.type_name, 'structure')
|
|
self.assertEqual(shape.name, 'ChangePasswordRequest')
|
|
self.assertEqual(list(sorted(shape.members)),
|
|
['NewPassword', 'OldPassword'])
|
|
self.assertEqual(shape.members['OldPassword'].name, 'passwordType')
|
|
self.assertEqual(shape.members['OldPassword'].type_name, 'string')
|
|
self.assertEqual(shape.error_code, None)
|
|
|
|
def test_exception_error_code(self):
|
|
shapes = {
|
|
'FooException': {
|
|
'exception': True,
|
|
'type': 'structure',
|
|
'members': {}
|
|
}
|
|
}
|
|
# Test without explicit error code
|
|
resolver = model.ShapeResolver(shapes)
|
|
shape = resolver.get_shape_by_name('FooException')
|
|
self.assertTrue(shape.metadata['exception'])
|
|
self.assertEqual(shape.error_code, 'FooException')
|
|
# Test with explicit error code
|
|
shapes['FooException']['error'] = {'code': 'ExceptionCode'}
|
|
resolver = model.ShapeResolver(shapes)
|
|
shape = resolver.get_shape_by_name('FooException')
|
|
self.assertTrue(shape.metadata['exception'])
|
|
self.assertEqual(shape.error_code, 'ExceptionCode')
|
|
|
|
def test_shape_metadata(self):
|
|
shapes = {
|
|
'ChangePasswordRequest': {
|
|
'type': 'structure',
|
|
'required': ['OldPassword', 'NewPassword'],
|
|
'members': {
|
|
'OldPassword': {'shape': 'passwordType'},
|
|
'NewPassword': {'shape': 'passwordType'},
|
|
}
|
|
},
|
|
'passwordType': {
|
|
"type":"string",
|
|
"min":1,
|
|
"max":128,
|
|
"sensitive":True
|
|
}
|
|
}
|
|
resolver = model.ShapeResolver(shapes)
|
|
shape = resolver.get_shape_by_name('ChangePasswordRequest')
|
|
self.assertEqual(shape.metadata['required'],
|
|
['OldPassword', 'NewPassword'])
|
|
member = shape.members['OldPassword']
|
|
self.assertEqual(member.metadata['min'], 1)
|
|
self.assertEqual(member.metadata['max'], 128)
|
|
self.assertEqual(member.metadata['sensitive'], True)
|
|
|
|
def test_error_shape_metadata(self):
|
|
shapes = {
|
|
'ResourceNotFoundException': {
|
|
'type': 'structure',
|
|
'members': {
|
|
'message': {
|
|
'shape': 'ErrorMessage',
|
|
}
|
|
},
|
|
'exception': True,
|
|
'retryable': {'throttling': True}
|
|
}
|
|
}
|
|
resolver = model.ShapeResolver(shapes)
|
|
shape = resolver.get_shape_by_name('ResourceNotFoundException')
|
|
self.assertEqual(
|
|
shape.metadata,
|
|
{'exception': True, 'retryable': {'throttling': True}}
|
|
)
|
|
|
|
def test_shape_list(self):
|
|
shapes = {
|
|
'mfaDeviceListType': {
|
|
"type":"list",
|
|
"member": {"shape": "MFADevice"},
|
|
},
|
|
'MFADevice': {
|
|
'type': 'structure',
|
|
'members': {
|
|
'UserName': {'shape': 'userNameType'}
|
|
}
|
|
},
|
|
'userNameType': {
|
|
'type': 'string'
|
|
}
|
|
}
|
|
resolver = model.ShapeResolver(shapes)
|
|
shape = resolver.get_shape_by_name('mfaDeviceListType')
|
|
self.assertEqual(shape.member.type_name, 'structure')
|
|
self.assertEqual(shape.member.name, 'MFADevice')
|
|
self.assertEqual(list(shape.member.members), ['UserName'])
|
|
|
|
def test_shape_does_not_exist(self):
|
|
resolver = model.ShapeResolver({})
|
|
with self.assertRaises(model.NoShapeFoundError):
|
|
resolver.get_shape_by_name('NoExistShape')
|
|
|
|
def test_missing_type_key(self):
|
|
shapes = {
|
|
'UnknownType': {
|
|
'NotTheTypeKey': 'someUnknownType'
|
|
}
|
|
}
|
|
resolver = model.ShapeResolver(shapes)
|
|
with self.assertRaises(model.InvalidShapeError):
|
|
resolver.get_shape_by_name('UnknownType')
|
|
|
|
def test_bad_shape_ref(self):
|
|
# This is an example of a denormalized model,
|
|
# which should raise an exception.
|
|
shapes = {
|
|
'Struct': {
|
|
'type': 'structure',
|
|
'members': {
|
|
'A': {'type': 'string'},
|
|
'B': {'type': 'string'},
|
|
}
|
|
}
|
|
}
|
|
resolver = model.ShapeResolver(shapes)
|
|
with self.assertRaises(model.InvalidShapeReferenceError):
|
|
struct = resolver.get_shape_by_name('Struct')
|
|
# Resolving the members will fail because
|
|
# the 'A' and 'B' members are not shape refs.
|
|
struct.members
|
|
|
|
def test_shape_name_in_repr(self):
|
|
shapes = {
|
|
'StringType': {
|
|
'type': 'string',
|
|
}
|
|
}
|
|
resolver = model.ShapeResolver(shapes)
|
|
self.assertIn('StringType',
|
|
repr(resolver.get_shape_by_name('StringType')))
|
|
|
|
|
|
class TestBuilders(unittest.TestCase):
|
|
|
|
def test_structure_shape_builder_with_scalar_types(self):
|
|
b = model.DenormalizedStructureBuilder()
|
|
shape = b.with_members({
|
|
'A': {'type': 'string'},
|
|
'B': {'type': 'integer'},
|
|
}).build_model()
|
|
self.assertIsInstance(shape, model.StructureShape)
|
|
self.assertEqual(sorted(list(shape.members)), ['A', 'B'])
|
|
self.assertEqual(shape.members['A'].type_name, 'string')
|
|
self.assertEqual(shape.members['B'].type_name, 'integer')
|
|
|
|
def test_structure_shape_with_structure_type(self):
|
|
b = model.DenormalizedStructureBuilder()
|
|
shape = b.with_members({
|
|
'A': {
|
|
'type': 'structure',
|
|
'members': {
|
|
'A-1': {'type': 'string'},
|
|
}
|
|
},
|
|
}).build_model()
|
|
self.assertIsInstance(shape, model.StructureShape)
|
|
self.assertEqual(list(shape.members), ['A'])
|
|
self.assertEqual(shape.members['A'].type_name, 'structure')
|
|
self.assertEqual(list(shape.members['A'].members), ['A-1'])
|
|
|
|
def test_structure_shape_with_list(self):
|
|
b = model.DenormalizedStructureBuilder()
|
|
shape = b.with_members({
|
|
'A': {
|
|
'type': 'list',
|
|
'member': {
|
|
'type': 'string'
|
|
}
|
|
},
|
|
}).build_model()
|
|
self.assertIsInstance(shape.members['A'], model.ListShape)
|
|
self.assertEqual(shape.members['A'].member.type_name, 'string')
|
|
|
|
def test_structure_shape_with_map_type(self):
|
|
b = model.DenormalizedStructureBuilder()
|
|
shape = b.with_members({
|
|
'A': {
|
|
'type': 'map',
|
|
'key': {'type': 'string'},
|
|
'value': {'type': 'string'},
|
|
}
|
|
}).build_model()
|
|
self.assertIsInstance(shape.members['A'], model.MapShape)
|
|
map_shape = shape.members['A']
|
|
self.assertEqual(map_shape.key.type_name, 'string')
|
|
self.assertEqual(map_shape.value.type_name, 'string')
|
|
|
|
def test_nested_structure(self):
|
|
b = model.DenormalizedStructureBuilder()
|
|
shape = b.with_members({
|
|
'A': {
|
|
'type': 'structure',
|
|
'members': {
|
|
'B': {
|
|
'type': 'structure',
|
|
'members': {
|
|
'C': {
|
|
'type': 'string',
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}).build_model()
|
|
self.assertEqual(
|
|
shape.members['A'].members['B'].members['C'].type_name, 'string')
|
|
|
|
def test_enum_values_on_string_used(self):
|
|
b = model.DenormalizedStructureBuilder()
|
|
enum_values = ['foo', 'bar', 'baz']
|
|
shape = b.with_members({
|
|
'A': {
|
|
'type': 'string',
|
|
'enum': enum_values,
|
|
},
|
|
}).build_model()
|
|
self.assertIsInstance(shape, model.StructureShape)
|
|
string_shape = shape.members['A']
|
|
self.assertIsInstance(string_shape, model.StringShape)
|
|
self.assertEqual(string_shape.metadata['enum'], enum_values)
|
|
self.assertEqual(string_shape.enum, enum_values)
|
|
|
|
def test_documentation_on_shape_used(self):
|
|
b = model.DenormalizedStructureBuilder()
|
|
shape = b.with_members({
|
|
'A': {
|
|
'type': 'string',
|
|
'documentation': 'MyDocs',
|
|
},
|
|
}).build_model()
|
|
self.assertEqual(shape.members['A'].documentation,
|
|
'MyDocs')
|
|
|
|
def test_min_max_used_in_metadata(self):
|
|
b = model.DenormalizedStructureBuilder()
|
|
shape = b.with_members({
|
|
'A': {
|
|
'type': 'string',
|
|
'documentation': 'MyDocs',
|
|
'min': 2,
|
|
'max': 3,
|
|
},
|
|
}).build_model()
|
|
metadata = shape.members['A'].metadata
|
|
self.assertEqual(metadata.get('min'), 2)
|
|
self.assertEqual(metadata.get('max'), 3)
|
|
|
|
def test_use_shape_name_when_provided(self):
|
|
b = model.DenormalizedStructureBuilder()
|
|
shape = b.with_members({
|
|
'A': {
|
|
'type': 'string',
|
|
'shape_name': 'MyStringShape',
|
|
},
|
|
}).build_model()
|
|
self.assertEqual(shape.members['A'].name, 'MyStringShape')
|
|
|
|
def test_unknown_shape_type(self):
|
|
b = model.DenormalizedStructureBuilder()
|
|
with self.assertRaises(model.InvalidShapeError):
|
|
b.with_members({
|
|
'A': {
|
|
'type': 'brand-new-shape-type',
|
|
},
|
|
}).build_model()
|
|
|
|
def test_ordered_shape_builder(self):
|
|
b = model.DenormalizedStructureBuilder()
|
|
shape = b.with_members(OrderedDict(
|
|
[
|
|
('A', {
|
|
'type': 'string'
|
|
}),
|
|
('B', {
|
|
'type': 'structure',
|
|
'members': OrderedDict(
|
|
[
|
|
('C', {
|
|
'type': 'string'
|
|
}),
|
|
('D', {
|
|
'type': 'string'
|
|
})
|
|
]
|
|
)
|
|
})
|
|
]
|
|
)).build_model()
|
|
|
|
# Members should be in order
|
|
self.assertEqual(['A', 'B'], list(shape.members.keys()))
|
|
|
|
# Nested structure members should *also* stay ordered
|
|
self.assertEqual(['C', 'D'], list(shape.members['B'].members.keys()))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|