1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
|
# Copyright 2017 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 botocore.session
from botocore.model import OperationNotFoundError
from botocore.utils import parse_timestamp
def test_lint_shared_example_configs():
session = botocore.session.Session()
loader = session.get_component('data_loader')
services = loader.list_available_services('examples-1')
for service in services:
service_model = session.get_service_model(service)
example_config = loader.load_service_model(
service, 'examples-1', service_model.api_version
)
examples = example_config.get("examples", {})
for operation, operation_examples in examples.items():
for example in operation_examples:
yield _lint_single_example, operation, example, service_model
def _lint_single_example(operation_name, example_config, service_model):
# The operation should actually exist
assert_operation_exists(service_model, operation_name)
operation_model = service_model.operation_model(operation_name)
assert_valid_values(service_model.service_name, operation_model,
example_config)
def assert_valid_values(service_name, operation_model, example_config):
example_input = example_config.get('input')
input_shape = operation_model.input_shape
example_id = example_config['id']
if input_shape is None and example_input:
raise AssertionError(
"Input found in example for %s from %s with id %s, but no input "
"shape is defined." % (
operation_model.name, service_name, example_id
))
example_output = example_config.get('output')
output_shape = operation_model.output_shape
if output_shape is None and example_output:
raise AssertionError(
"Output found in example for %s from %s with id %s, but no output "
"shape is defined." % (
operation_model.name, service_name, example_id
))
try:
if example_input is not None and input_shape is not None:
_assert_valid_values(
input_shape, example_input, [input_shape.name])
if example_output is not None and output_shape is not None:
_assert_valid_values(
output_shape, example_output, [output_shape.name])
except AssertionError as e:
raise AssertionError(
"Invalid value in example for %s from %s with id %s: %s" % (
operation_model.name, service_name, example_id, e
))
def _assert_valid_values(shape, example_value, path):
if shape.type_name == 'timestamp':
_assert_valid_timestamp(example_value, path)
elif shape.type_name == 'structure':
_assert_valid_structure_values(shape, example_value, path)
elif shape.type_name == 'list':
_assert_valid_list_values(shape, example_value, path)
elif shape.type_name == 'map':
_assert_valid_map_values(shape, example_value, path)
def _assert_valid_structure_values(shape, example_dict, path):
invalid_members = [k for k in example_dict.keys()
if k not in shape.members]
if invalid_members:
dotted_path = '.'.join(path)
raise AssertionError(
"Invalid members found for %s: %s" % (dotted_path, invalid_members)
)
for member_name, example_value in example_dict.items():
member = shape.members[member_name]
_assert_valid_values(member, example_value, path + [member_name])
def _assert_valid_list_values(shape, example_values, path):
member = shape.member
for i, value in enumerate(example_values):
name = "%s[%s]" % (path[-1], i)
_assert_valid_values(member, value, path[:-1] + [name])
def _assert_valid_map_values(shape, example_value, path):
for key, value in example_value.items():
name = '%s["%s"]' % (path[-1], key)
_assert_valid_values(shape.value, value, path[:-1] + [name])
def _assert_valid_timestamp(timestamp, path):
try:
parse_timestamp(timestamp).timetuple()
except Exception as e:
dotted_path = '.'.join(path)
raise AssertionError('Failed to parse timestamp %s for %s: %s' % (
timestamp, dotted_path, e))
def assert_operation_exists(service_model, operation_name):
try:
service_model.operation_model(operation_name)
except OperationNotFoundError:
raise AssertionError(
"Examples found in %s for operation %s that does not exist." % (
service_model.service_name, operation_name))
|
