Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def create_argument_table(self, arguments):
if not arguments:
return OrderedDict()
argument_table = OrderedDict()
for arg in arguments:
if isinstance(arg, BaseCLIArgument):
argument_table[arg.name] = arg
else:
argument_table[arg] = BaseCLIArgument(arg)
return argument_table
def test_does_decode_template_body_in_order(self):
expected_ordering = OrderedDict([
('TemplateVersion', 1.0),
('APropertyOfSomeKind', 'a value'),
('list', [1, 2, 3]),
('nested', OrderedDict([('key', 'value'),
('foo', 'bar')]))
])
template_string = json.dumps(expected_ordering)
parsed_response = {'TemplateBody': template_string}
handlers.json_decode_template_body(parsed=parsed_response)
result = parsed_response['TemplateBody']
self.assertTrue(isinstance(result, OrderedDict))
for element, expected_element in zip(result, expected_ordering):
self.assertEqual(element, expected_element)
"input_token": "NextResult",
"output_token": "NextResult",
"limit_key": "MaxResults",
"result_key": "Biz"
}
}
}
self.example_json_model = {
"version": 1,
"examples": {
"SampleOperation": [{
"id": "sample-id",
"title": "sample-title",
"description": "Sample Description.",
"input": OrderedDict([
("Biz", "foo"),
]),
"comments": {
"input": {
"Biz": "bar"
},
def test_update_existing(self):
initial = OrderedDict([
("apiVersion", "v1"),
("clusters", [
OrderedDict([
("cluster", OrderedDict([
("server", "endpoint")
])),
("name", "clustername")
])
]),
("contexts", []),
("current-context", None),
("kind", "Config"),
("preferences", OrderedDict()),
("users", [])
])
cluster = OrderedDict([
("cluster", OrderedDict([
("certificate-authority-data", "data"),
("server", "endpoint")
])),
("name", "clustername")
])
correct = OrderedDict([
("apiVersion", "v1"),
("clusters", [
OrderedDict([
("cluster", OrderedDict([
("certificate-authority-data", "data"),
("server", "endpoint")
def create_command_table(self, commands, command_creator):
if not commands:
return OrderedDict()
command_table = OrderedDict()
for name, command in commands.items():
if isinstance(command, CLICommand):
# Already a valid command, no need to fake one
command_table[name] = command
else:
command_table[name] = command_creator(name, command)
return command_table
def test_update_existing(self):
initial = OrderedDict([
("apiVersion", "v1"),
("clusters", [
OrderedDict([
("cluster", OrderedDict([
("server", "endpoint")
])),
("name", "clustername")
])
]),
("contexts", []),
("current-context", None),
("kind", "Config"),
("preferences", OrderedDict()),
("users", [])
])
cluster = OrderedDict([
def get_cluster_entry(self):
"""
Return a cluster entry generated using
the previously obtained description.
"""
cert_data = self._get_cluster_description().get("certificateAuthority",
{"data": ""})["data"]
endpoint = self._get_cluster_description().get("endpoint")
arn = self._get_cluster_description().get("arn")
return OrderedDict([
("cluster", OrderedDict([
("certificate-authority-data", cert_data),
("server", endpoint)
])),
("name", arn)
])
:param service_model: The service model
:param paginator_config: The paginator config associated to a particular
paginator.
:param include_signature: Whether or not to include the signature.
It is useful for generating docstrings.
"""
# Retrieve the operation model of the underlying operation.
operation_model = service_model.operation_model(
paginator_name)
# Add representations of the request and response parameters
# we want to include in the description of the paginate method.
# These are parameters we expose via the botocore interface.
pagination_config_members = OrderedDict()
pagination_config_members['MaxItems'] = DocumentedShape(
name='MaxItems', type_name='integer',
documentation=(
'<p>The total number of items to return. If the total '
'number of items available is more than the value '
'specified in max-items then a <code>NextToken</code> '
'will be provided in the output that you can use to '
'resume pagination.</p>'))
if paginator_config.get('limit_key', None):
pagination_config_members['PageSize'] = DocumentedShape(
name='PageSize', type_name='integer',
documentation='<p>The size of each page.</p><p>')
pagination_config_members['StartingToken'] = DocumentedShape(</p>
def _unpack_json_cli_arg(argument_model, value, cli_name):
try:
return json.loads(value, object_pairs_hook=OrderedDict)
except ValueError as e:
raise ParamError(
cli_name, "Invalid JSON: %s\nJSON received: %s"
% (e, value))
:param service_model: The service model
:param paginator_config: The paginator config associated to a particular
paginator.
:param include_signature: Whether or not to include the signature.
It is useful for generating docstrings.
"""
# Retrieve the operation model of the underlying operation.
operation_model = service_model.operation_model(
paginator_name)
# Add representations of the request and response parameters
# we want to include in the description of the paginate method.
# These are parameters we expose via the botocore interface.
pagination_config_members = OrderedDict()
pagination_config_members['MaxItems'] = DocumentedShape(
name='MaxItems', type_name='integer',
documentation=(
'<p>The total number of items to return. If the total '
'number of items available is more than the value '
'specified in max-items then a <code>NextToken</code> '
'will be provided in the output that you can use to '
'resume pagination.</p>'))
if paginator_config.get('limit_key', None):
pagination_config_members['PageSize'] = DocumentedShape(
name='PageSize', type_name='integer',
documentation='<p>The size of each page.</p><p>')
pagination_config_members['StartingToken'] = DocumentedShape(</p>