Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #34 from spulec/master

Browse Source 
Merge upstream
This commit is contained in:
Bert Blommers 2020-03-17 15:08:45 +00:00 committed by GitHub
commit 60c98ca836
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
29 changed files with 1154 additions and 193 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
IMPLEMENTATION_COVERAGE.md
View File

@ -2237,7 +2237,7 @@
- [ ] verify_trust
## dynamodb
17% implemented
24% implemented
- [ ] batch_get_item
- [ ] batch_write_item
- [ ] create_backup
@ -2268,7 +2268,7 @@
- [ ] restore_table_to_point_in_time
- [X] scan
- [ ] tag_resource
- [ ] transact_get_items
- [X] transact_get_items
- [ ] transact_write_items
- [ ] untag_resource
- [ ] update_continuous_backups

11
moto/batch/models.py
View File

@ -301,7 +301,7 @@ class Job(threading.Thread, BaseModel):
self.job_name = name
self.job_id = str(uuid.uuid4())
self.job_definition = job_def
self.container_overrides = container_overrides
self.container_overrides = container_overrides or {}
self.job_queue = job_queue
self.job_state = "SUBMITTED" # One of SUBMITTED | PENDING | RUNNABLE | STARTING | RUNNING | SUCCEEDED | FAILED
self.job_queue.jobs.append(self)
@ -317,6 +317,7 @@ class Job(threading.Thread, BaseModel):
self.docker_client = docker.from_env()
self._log_backend = log_backend
self.log_stream_name = None
# Unfortunately mocking replaces this method w/o fallback enabled, so we
# need to replace it if we detect it's been mocked
@ -338,10 +339,11 @@ class Job(threading.Thread, BaseModel):
"jobId": self.job_id,
"jobName": self.job_name,
"jobQueue": self.job_queue.arn,
"startedAt": datetime2int(self.job_started_at),
"status": self.job_state,
"dependsOn": [],
}
if result["status"] not in ["SUBMITTED", "PENDING", "RUNNABLE", "STARTING"]:
result["startedAt"] = datetime2int(self.job_started_at)
if self.job_stopped:
result["stoppedAt"] = datetime2int(self.job_stopped_at)
result["container"] = {}
@ -503,7 +505,10 @@ class Job(threading.Thread, BaseModel):
for line in logs_stdout + logs_stderr:
date, line = line.split(" ", 1)
date = dateutil.parser.parse(date)
date = int(date.timestamp())
# TODO: Replace with int(date.timestamp()) once we yeet Python2 out of the window
date = int(
(time.mktime(date.timetuple()) + date.microsecond / 1000000.0)
)
logs.append({"timestamp": date, "message": line.strip()})
# Send to cloudwatch

8
moto/cloudformation/parsing.py
View File

@ -196,13 +196,13 @@ def clean_json(resource_json, resources_map):
)
else:
fn_sub_value = clean_json(resource_json["Fn::Sub"], resources_map)
to_sub = re.findall('(?=\${)[^!^"]*?}', fn_sub_value)
literals = re.findall('(?=\${!)[^"]*?}', fn_sub_value)
to_sub = re.findall(r'(?=\${)[^!^"]*?}', fn_sub_value)
literals = re.findall(r'(?=\${!)[^"]*?}', fn_sub_value)
for sub in to_sub:
if "." in sub:
cleaned_ref = clean_json(
{
"Fn::GetAtt": re.findall('(?<=\${)[^"]*?(?=})', sub)[
"Fn::GetAtt": re.findall(r'(?<=\${)[^"]*?(?=})', sub)[
0
].split(".")
},
@ -210,7 +210,7 @@ def clean_json(resource_json, resources_map):
)
else:
cleaned_ref = clean_json(
{"Ref": re.findall('(?<=\${)[^"]*?(?=})', sub)[0]},
{"Ref": re.findall(r'(?<=\${)[^"]*?(?=})', sub)[0]},
resources_map,
)
fn_sub_value = fn_sub_value.replace(sub, cleaned_ref)

12
moto/cloudwatch/models.py
View File

@ -2,13 +2,14 @@ import json
from boto3 import Session
from moto.core.utils import iso_8601_datetime_with_milliseconds
from moto.core.utils import iso_8601_datetime_without_milliseconds
from moto.core import BaseBackend, BaseModel
from moto.core.exceptions import RESTError
from datetime import datetime, timedelta
from dateutil.tz import tzutc
from uuid import uuid4
from .utils import make_arn_for_dashboard
from dateutil import parser
from moto.core import ACCOUNT_ID as DEFAULT_ACCOUNT_ID
@ -66,6 +67,7 @@ class FakeAlarm(BaseModel):
ok_actions,
insufficient_data_actions,
unit,
actions_enabled,
):
self.name = name
self.namespace = namespace
@ -79,6 +81,7 @@ class FakeAlarm(BaseModel):
self.dimensions = [
Dimension(dimension["name"], dimension["value"]) for dimension in dimensions
]
self.actions_enabled = actions_enabled
self.alarm_actions = alarm_actions
self.ok_actions = ok_actions
self.insufficient_data_actions = insufficient_data_actions
@ -146,7 +149,7 @@ class Dashboard(BaseModel):
class Statistics:
def __init__(self, stats, dt):
self.timestamp = iso_8601_datetime_with_milliseconds(dt)
self.timestamp = iso_8601_datetime_without_milliseconds(dt)
self.values = []
self.stats = stats
@ -214,6 +217,7 @@ class CloudWatchBackend(BaseBackend):
ok_actions,
insufficient_data_actions,
unit,
actions_enabled,
):
alarm = FakeAlarm(
name,
@ -230,6 +234,7 @@ class CloudWatchBackend(BaseBackend):
ok_actions,
insufficient_data_actions,
unit,
actions_enabled,
)
self.alarms[name] = alarm
return alarm
@ -278,8 +283,7 @@ class CloudWatchBackend(BaseBackend):
# Preserve "datetime" for get_metric_statistics comparisons
timestamp = metric_member.get("Timestamp")
if timestamp is not None and type(timestamp) != datetime:
timestamp = datetime.strptime(timestamp, "%Y-%m-%dT%H:%M:%S.%fZ")
timestamp = timestamp.replace(tzinfo=tzutc())
timestamp = parser.parse(timestamp)
self.metric_data.append(
MetricDatum(
namespace,

2
moto/cloudwatch/responses.py
View File

@ -28,6 +28,7 @@ class CloudWatchResponse(BaseResponse):
dimensions = self._get_list_prefix("Dimensions.member")
alarm_actions = self._get_multi_param("AlarmActions.member")
ok_actions = self._get_multi_param("OKActions.member")
actions_enabled = self._get_param("ActionsEnabled")
insufficient_data_actions = self._get_multi_param(
"InsufficientDataActions.member"
)
@ -47,6 +48,7 @@ class CloudWatchResponse(BaseResponse):
ok_actions,
insufficient_data_actions,
unit,
actions_enabled,
)
template = self.response_template(PUT_METRIC_ALARM_TEMPLATE)
return template.render(alarm=alarm)

4
moto/core/models.py
View File

@ -347,7 +347,7 @@ class BotocoreEventMockAWS(BaseMockAWS):
responses_mock.add(
CallbackResponse(
method=method,
url=re.compile("https?://.+.amazonaws.com/.*"),
url=re.compile(r"https?://.+.amazonaws.com/.*"),
callback=not_implemented_callback,
stream=True,
match_querystring=False,
@ -356,7 +356,7 @@ class BotocoreEventMockAWS(BaseMockAWS):
botocore_mock.add(
CallbackResponse(
method=method,
url=re.compile("https?://.+.amazonaws.com/.*"),
url=re.compile(r"https?://.+.amazonaws.com/.*"),
callback=not_implemented_callback,
stream=True,
match_querystring=False,

2
moto/core/utils.py
View File

@ -95,7 +95,7 @@ def convert_regex_to_flask_path(url_path):
match_name, match_pattern = reg.groups()
return '<regex("{0}"):{1}>'.format(match_pattern, match_name)
url_path = re.sub("\(\?P<(.*?)>(.*?)\)", caller, url_path)
url_path = re.sub(r"\(\?P<(.*?)>(.*?)\)", caller, url_path)
if url_path.endswith("/?"):
# Flask does own handling of trailing slashes

20
moto/dynamodb2/comparisons.py
View File

@ -251,9 +251,9 @@ class ConditionExpressionParser:
def _lex_one_node(self, remaining_expression):
# TODO: Handle indexing like [1]
attribute_regex = "(:|#)?[A-z0-9\-_]+"
attribute_regex = r"(:|#)?[A-z0-9\-_]+"
patterns = [
(self.Nonterminal.WHITESPACE, re.compile("^ +")),
(self.Nonterminal.WHITESPACE, re.compile(r"^ +")),
(
self.Nonterminal.COMPARATOR,
re.compile(
@ -270,12 +270,14 @@ class ConditionExpressionParser:
(
self.Nonterminal.OPERAND,
re.compile(
"^" + attribute_regex + "(\." + attribute_regex + "|\[[0-9]\])*"
r"^{attribute_regex}(\.{attribute_regex}|\[[0-9]\])*".format(
attribute_regex=attribute_regex
)
),
),
(self.Nonterminal.COMMA, re.compile("^,")),
(self.Nonterminal.LEFT_PAREN, re.compile("^\(")),
(self.Nonterminal.RIGHT_PAREN, re.compile("^\)")),
(self.Nonterminal.COMMA, re.compile(r"^,")),
(self.Nonterminal.LEFT_PAREN, re.compile(r"^\(")),
(self.Nonterminal.RIGHT_PAREN, re.compile(r"^\)")),
]
for nonterminal, pattern in patterns:
@ -285,7 +287,7 @@ class ConditionExpressionParser:
break
else: # pragma: no cover
raise ValueError(
"Cannot parse condition starting at: " + remaining_expression
"Cannot parse condition starting at:{}".format(remaining_expression)
)
node = self.Node(
@ -318,7 +320,7 @@ class ConditionExpressionParser:
for child in children:
self._assert(
child.nonterminal == self.Nonterminal.IDENTIFIER,
"Cannot use %s in path" % child.text,
"Cannot use {} in path".format(child.text),
[node],
)
output.append(
@ -392,7 +394,7 @@ class ConditionExpressionParser:
elif name.startswith("["):
# e.g. [123]
if not name.endswith("]"): # pragma: no cover
raise ValueError("Bad path element %s" % name)
raise ValueError("Bad path element {}".format(name))
return self.Node(
nonterminal=self.Nonterminal.IDENTIFIER,
kind=self.Kind.LITERAL,

7
moto/dynamodb2/models.py
View File

@ -792,6 +792,12 @@ class Table(BaseModel):
expression_attribute_values=None,
overwrite=False,
):
if self.hash_key_attr not in item_attrs.keys():
raise ValueError(
"One or more parameter values were invalid: Missing the key "
+ self.hash_key_attr
+ " in the item"
)
hash_value = DynamoType(item_attrs.get(self.hash_key_attr))
if self.has_range_key:
range_value = DynamoType(item_attrs.get(self.range_key_attr))
@ -808,7 +814,6 @@ class Table(BaseModel):
else:
lookup_range_value = DynamoType(expected_range_value)
current = self.get_item(hash_value, lookup_range_value)
item = Item(
hash_value, self.hash_key_type, range_value, self.range_key_type, item_attrs
)

73
moto/dynamodb2/responses.py
View File

@ -10,6 +10,9 @@ from .exceptions import InvalidIndexNameError, InvalidUpdateExpression, ItemSize
from .models import dynamodb_backends, dynamo_json_dump
TRANSACTION_MAX_ITEMS = 25
def has_empty_keys_or_values(_dict):
if _dict == "":
return True
@ -293,11 +296,9 @@ class DynamoHandler(BaseResponse):
except ItemSizeTooLarge:
er = "com.amazonaws.dynamodb.v20111205#ValidationException"
return self.error(er, ItemSizeTooLarge.message)
except ValueError:
except ValueError as ve:
er = "com.amazonaws.dynamodb.v20111205#ConditionalCheckFailedException"
return self.error(
er, "A condition specified in the operation could not be evaluated."
)
return self.error(er, str(ve))
if result:
item_dict = result.to_json()
@ -828,3 +829,67 @@ class DynamoHandler(BaseResponse):
ttl_spec = self.dynamodb_backend.describe_ttl(name)
return json.dumps({"TimeToLiveDescription": ttl_spec})
def transact_get_items(self):
transact_items = self.body["TransactItems"]
responses = list()
if len(transact_items) > TRANSACTION_MAX_ITEMS:
msg = "1 validation error detected: Value '["
err_list = list()
request_id = 268435456
for _ in transact_items:
request_id += 1
hex_request_id = format(request_id, "x")
err_list.append(
"com.amazonaws.dynamodb.v20120810.TransactGetItem@%s"
% hex_request_id
)
msg += ", ".join(err_list)
msg += (
"'] at 'transactItems' failed to satisfy constraint: "
"Member must have length less than or equal to %s"
% TRANSACTION_MAX_ITEMS
)
return self.error("ValidationException", msg)
ret_consumed_capacity = self.body.get("ReturnConsumedCapacity", "NONE")
consumed_capacity = dict()
for transact_item in transact_items:
table_name = transact_item["Get"]["TableName"]
key = transact_item["Get"]["Key"]
try:
item = self.dynamodb_backend.get_item(table_name, key)
except ValueError:
er = "com.amazonaws.dynamodb.v20111205#ResourceNotFoundException"
return self.error(er, "Requested resource not found")
if not item:
continue
item_describe = item.describe_attrs(False)
responses.append(item_describe)
table_capacity = consumed_capacity.get(table_name, {})
table_capacity["TableName"] = table_name
capacity_units = table_capacity.get("CapacityUnits", 0) + 2.0
table_capacity["CapacityUnits"] = capacity_units
read_capacity_units = table_capacity.get("ReadCapacityUnits", 0) + 2.0
table_capacity["ReadCapacityUnits"] = read_capacity_units
consumed_capacity[table_name] = table_capacity
if ret_consumed_capacity == "INDEXES":
table_capacity["Table"] = {
"CapacityUnits": capacity_units,
"ReadCapacityUnits": read_capacity_units,
}
result = dict()
result.update({"Responses": responses})
if ret_consumed_capacity != "NONE":
result.update({"ConsumedCapacity": [v for v in consumed_capacity.values()]})
return dynamo_json_dump(result)

9
moto/ec2/utils.py
View File

@ -252,7 +252,8 @@ def dhcp_configuration_from_querystring(querystring, option="DhcpConfiguration")
def filters_from_querystring(querystring_dict):
response_values = {}
for key, value in querystring_dict.items():
last_tag_key = None
for key, value in sorted(querystring_dict.items()):
match = re.search(r"Filter.(\d).Name", key)
if match:
filter_index = match.groups()[0]
@ -262,6 +263,10 @@ def filters_from_querystring(querystring_dict):
for filter_key, filter_value in querystring_dict.items()
if filter_key.startswith(value_prefix)
]
if value[0] == "tag-key":
last_tag_key = "tag:" + filter_values[0]
elif last_tag_key and value[0] == "tag-value":
response_values[last_tag_key] = filter_values
response_values[value[0]] = filter_values
return response_values
@ -329,6 +334,8 @@ def tag_filter_matches(obj, filter_name, filter_values):
tag_values = get_obj_tag_names(obj)
elif filter_name == "tag-value":
tag_values = get_obj_tag_values(obj)
elif filter_name.startswith("tag:"):
tag_values = get_obj_tag_values(obj)
else:
tag_values = [get_obj_tag(obj, filter_name) or ""]

2
moto/ecr/models.py
View File

@ -403,7 +403,7 @@ class ECRBackend(BaseBackend):
# If we have a digest, is it valid?
if "imageDigest" in image_id:
pattern = re.compile("^[0-9a-zA-Z_+\.-]+:[0-9a-fA-F]{64}")
pattern = re.compile(r"^[0-9a-zA-Z_+\.-]+:[0-9a-fA-F]{64}")
if not pattern.match(image_id.get("imageDigest")):
response["failures"].append(
{

4
moto/events/responses.py
View File

@ -62,7 +62,9 @@ class EventsHandler(BaseResponse):
rule = self.events_backend.describe_rule(name)
if not rule:
return self.error("ResourceNotFoundException", "Rule test does not exist.")
return self.error(
"ResourceNotFoundException", "Rule " + name + " does not exist."
)
rule_dict = self._generate_rule_dict(rule)
return json.dumps(rule_dict), self.response_headers

6
moto/iam/models.py Normal file → Executable file
View File

@ -12,6 +12,7 @@ import re
from cryptography import x509
from cryptography.hazmat.backends import default_backend
from six.moves.urllib.parse import urlparse
from uuid import uuid4
from moto.core.exceptions import RESTError
from moto.core import BaseBackend, BaseModel, ACCOUNT_ID
@ -330,9 +331,12 @@ class Role(BaseModel):
cls, resource_name, cloudformation_json, region_name
):
properties = cloudformation_json["Properties"]
role_name = (
properties["RoleName"] if "RoleName" in properties else str(uuid4())[0:5]
)
role = iam_backend.create_role(
role_name=resource_name,
role_name=role_name,
assume_role_policy_document=properties["AssumeRolePolicyDocument"],
path=properties.get("Path", "/"),
permissions_boundary=properties.get("PermissionsBoundary", ""),

2
moto/s3/utils.py
View File

@ -52,7 +52,7 @@ def parse_region_from_url(url):
def metadata_from_headers(headers):
metadata = {}
meta_regex = re.compile("^x-amz-meta-([a-zA-Z0-9\-_]+)$", flags=re.IGNORECASE)
meta_regex = re.compile(r"^x-amz-meta-([a-zA-Z0-9\-_]+)$", flags=re.IGNORECASE)
for header, value in headers.items():
if isinstance(header, six.string_types):
result = meta_regex.match(header)

25
moto/ssm/exceptions.py
View File

@ -23,6 +23,31 @@ class InvalidFilterValue(JsonRESTError):
super(InvalidFilterValue, self).__init__("InvalidFilterValue", message)
class ParameterNotFound(JsonRESTError):
code = 400
def __init__(self, message):
super(ParameterNotFound, self).__init__("ParameterNotFound", message)
class ParameterVersionNotFound(JsonRESTError):
code = 400
def __init__(self, message):
super(ParameterVersionNotFound, self).__init__(
"ParameterVersionNotFound", message
)
class ParameterVersionLabelLimitExceeded(JsonRESTError):
code = 400
def __init__(self, message):
super(ParameterVersionLabelLimitExceeded, self).__init__(
"ParameterVersionLabelLimitExceeded", message
)
class ValidationException(JsonRESTError):
code = 400

68
moto/ssm/models.py
View File

@ -19,6 +19,9 @@ from .exceptions import (
InvalidFilterValue,
InvalidFilterOption,
InvalidFilterKey,
ParameterVersionLabelLimitExceeded,
ParameterVersionNotFound,
ParameterNotFound,
)
@ -41,6 +44,7 @@ class Parameter(BaseModel):
self.keyid = keyid
self.last_modified_date = last_modified_date
self.version = version
self.labels = []
if self.type == "SecureString":
if not self.keyid:
@ -75,7 +79,7 @@ class Parameter(BaseModel):
return r
def describe_response_object(self, decrypt=False):
def describe_response_object(self, decrypt=False, include_labels=False):
r = self.response_object(decrypt)
r["LastModifiedDate"] = round(self.last_modified_date, 3)
r["LastModifiedUser"] = "N/A"
@ -89,6 +93,9 @@ class Parameter(BaseModel):
if self.allowed_pattern:
r["AllowedPattern"] = self.allowed_pattern
if include_labels:
r["Labels"] = self.labels
return r
@ -614,6 +621,65 @@ class SimpleSystemManagerBackend(BaseBackend):
return self._parameters[name][-1]
return None
def label_parameter_version(self, name, version, labels):
previous_parameter_versions = self._parameters[name]
if not previous_parameter_versions:
raise ParameterNotFound("Parameter %s not found." % name)
found_parameter = None
labels_needing_removal = []
if not version:
version = 1
for parameter in previous_parameter_versions:
if parameter.version >= version:
version = parameter.version
for parameter in previous_parameter_versions:
if parameter.version == version:
found_parameter = parameter
else:
for label in labels:
if label in parameter.labels:
labels_needing_removal.append(label)
if not found_parameter:
raise ParameterVersionNotFound(
"Systems Manager could not find version %s of %s. "
"Verify the version and try again." % (version, name)
)
labels_to_append = []
invalid_labels = []
for label in labels:
if (
label.startswith("aws")
or label.startswith("ssm")
or label[:1].isdigit()
or not re.match("^[a-zA-z0-9_\.\-]*$", label)
):
invalid_labels.append(label)
continue
if len(label) > 100:
raise ValidationException(
"1 validation error detected: "
"Value '[%s]' at 'labels' failed to satisfy constraint: "
"Member must satisfy constraint: "
"[Member must have length less than or equal to 100, Member must have length greater than or equal to 1]"
% label
)
continue
if label not in found_parameter.labels:
labels_to_append.append(label)
if (len(found_parameter.labels) + len(labels_to_append)) > 10:
raise ParameterVersionLabelLimitExceeded(
"An error occurred (ParameterVersionLabelLimitExceeded) when calling the LabelParameterVersion operation: "
"A parameter version can have maximum 10 labels."
"Move one or more labels to another version and try again."
)
found_parameter.labels = found_parameter.labels + labels_to_append
for parameter in previous_parameter_versions:
if parameter.version != version:
for label in parameter.labels[:]:
if label in labels_needing_removal:
parameter.labels.remove(label)
return [invalid_labels, version]
def put_parameter(
self, name, description, value, type, allowed_pattern, keyid, overwrite
):

14
moto/ssm/responses.py
View File

@ -168,12 +168,24 @@ class SimpleSystemManagerResponse(BaseResponse):
response = {"Parameters": []}
for parameter_version in result:
param_data = parameter_version.describe_response_object(
decrypt=with_decryption
decrypt=with_decryption, include_labels=True
)
response["Parameters"].append(param_data)
return json.dumps(response)
def label_parameter_version(self):
name = self._get_param("Name")
version = self._get_param("ParameterVersion")
labels = self._get_param("Labels")
invalid_labels, version = self.ssm_backend.label_parameter_version(
name, version, labels
)
response = {"InvalidLabels": invalid_labels, "ParameterVersion": version}
return json.dumps(response)
def add_tags_to_resource(self):
resource_id = self._get_param("ResourceId")
resource_type = self._get_param("ResourceType")

4
requirements-dev.txt
View File

@ -1,5 +1,5 @@
-r requirements.txt
mock==3.0.5 # Last version compatible with Python 2.7
mock<=3.0.5 # Last version compatible with Python 2.7
nose
black; python_version >= '3.6'
regex==2019.11.1; python_version >= '3.6' # Needed for black
@ -10,7 +10,7 @@ freezegun
flask
boto>=2.45.0
boto3>=1.4.4
botocore>=1.12.13
botocore>=1.15.13
six>=1.9
parameterized>=0.7.0
prompt-toolkit==1.0.14

6
setup.py
View File

@ -29,7 +29,7 @@ def get_version():
install_requires = [
"setuptools==44.0.0",
"Jinja2==2.11.0",
"Jinja2<3.0.0,>=2.10.1",
"boto>=2.36.0",
"boto3>=1.9.201",
"botocore>=1.12.201",
@ -42,12 +42,12 @@ install_requires = [
"pytz",
"python-dateutil<3.0.0,>=2.1",
"python-jose<4.0.0",
"mock==3.0.5",
"mock<=3.0.5",
"docker>=2.5.1",
"jsondiff>=1.1.2",
"aws-xray-sdk!=0.96,>=0.93",
"responses>=0.9.0",
"idna<2.9,>=2.5",
"idna<3,>=2.5",
"cfn-lint>=0.4.0",
"sshpubkeys>=3.1.0,<4.0",
"zipp==0.6.0",

39
tests/test_batch/test_batch.py
View File

@ -10,17 +10,6 @@ import functools
import nose
def expected_failure(test):
@functools.wraps(test)
def inner(*args, **kwargs):
try:
test(*args, **kwargs)
except Exception as err:
raise nose.SkipTest
return inner
DEFAULT_REGION = "eu-central-1"
@ -692,7 +681,8 @@ def test_submit_job_by_name():
# SLOW TESTS
@expected_failure
@mock_logs
@mock_ec2
@mock_ecs
@ -720,13 +710,13 @@ def test_submit_job():
queue_arn = resp["jobQueueArn"]
resp = batch_client.register_job_definition(
jobDefinitionName="sleep10",
jobDefinitionName="sayhellotomylittlefriend",
type="container",
containerProperties={
"image": "busybox",
"image": "busybox:latest",
"vcpus": 1,
"memory": 128,
"command": ["sleep", "10"],
"command": ["echo", "hello"],
},
)
job_def_arn = resp["jobDefinitionArn"]
@ -740,13 +730,6 @@ def test_submit_job():
while datetime.datetime.now() < future:
resp = batch_client.describe_jobs(jobs=[job_id])
print(
"{0}:{1} {2}".format(
resp["jobs"][0]["jobName"],
resp["jobs"][0]["jobId"],
resp["jobs"][0]["status"],
)
)
if resp["jobs"][0]["status"] == "FAILED":
raise RuntimeError("Batch job failed")
@ -763,10 +746,9 @@ def test_submit_job():
resp = logs_client.get_log_events(
logGroupName="/aws/batch/job", logStreamName=ls_name
)
len(resp["events"]).should.be.greater_than(5)
[event["message"] for event in resp["events"]].should.equal(["hello"])
@expected_failure
@mock_logs
@mock_ec2
@mock_ecs
@ -794,13 +776,13 @@ def test_list_jobs():
queue_arn = resp["jobQueueArn"]
resp = batch_client.register_job_definition(
jobDefinitionName="sleep10",
jobDefinitionName="sleep5",
type="container",
containerProperties={
"image": "busybox",
"image": "busybox:latest",
"vcpus": 1,
"memory": 128,
"command": ["sleep", "10"],
"command": ["sleep", "5"],
},
)
job_def_arn = resp["jobDefinitionArn"]
@ -843,7 +825,6 @@ def test_list_jobs():
len(resp_finished_jobs2["jobSummaryList"]).should.equal(2)
@expected_failure
@mock_logs
@mock_ec2
@mock_ecs
@ -874,7 +855,7 @@ def test_terminate_job():
jobDefinitionName="sleep10",
type="container",
containerProperties={
"image": "busybox",
"image": "busybox:latest",
"vcpus": 1,
"memory": 128,
"command": ["sleep", "10"],

11
tests/test_cloudformation/test_cloudformation_stack_integration.py
View File

@ -909,6 +909,7 @@ def test_iam_roles():
},
"my-role-no-path": {
"Properties": {
"RoleName": "my-role-no-path-name",
"AssumeRolePolicyDocument": {
"Statement": [
{
@ -917,7 +918,7 @@ def test_iam_roles():
"Principal": {"Service": ["ec2.amazonaws.com"]},
}
]
}
},
},
"Type": "AWS::IAM::Role",
},
@ -936,13 +937,15 @@ def test_iam_roles():
role_name_to_id = {}
for role_result in role_results:
role = iam_conn.get_role(role_result.role_name)
role.role_name.should.contain("my-role")
if "with-path" in role.role_name:
if "my-role" not in role.role_name:
role_name_to_id["with-path"] = role.role_id
role.path.should.equal("my-path")
len(role.role_name).should.equal(
5
) # Role name is not specified, so randomly generated - can't check exact name
else:
role_name_to_id["no-path"] = role.role_id
role.role_name.should.contain("no-path")
role.role_name.should.equal("my-role-no-path-name")
role.path.should.equal("/")
instance_profile_responses = iam_conn.list_instance_profiles()[

272
tests/test_cloudwatch/test_cloudwatch.py
View File

@ -1,117 +1,155 @@
import boto
from boto.ec2.cloudwatch.alarm import MetricAlarm
import sure # noqa
from moto import mock_cloudwatch_deprecated
def alarm_fixture(name="tester", action=None):
action = action or ["arn:alarm"]
return MetricAlarm(
name=name,
namespace="{0}_namespace".format(name),
metric="{0}_metric".format(name),
comparison=">=",
threshold=2.0,
period=60,
evaluation_periods=5,
statistic="Average",
description="A test",
dimensions={"InstanceId": ["i-0123456,i-0123457"]},
alarm_actions=action,
ok_actions=["arn:ok"],
insufficient_data_actions=["arn:insufficient"],
unit="Seconds",
)
@mock_cloudwatch_deprecated
def test_create_alarm():
conn = boto.connect_cloudwatch()
alarm = alarm_fixture()
conn.create_alarm(alarm)
alarms = conn.describe_alarms()
alarms.should.have.length_of(1)
alarm = alarms[0]
alarm.name.should.equal("tester")
alarm.namespace.should.equal("tester_namespace")
alarm.metric.should.equal("tester_metric")
alarm.comparison.should.equal(">=")
alarm.threshold.should.equal(2.0)
alarm.period.should.equal(60)
alarm.evaluation_periods.should.equal(5)
alarm.statistic.should.equal("Average")
alarm.description.should.equal("A test")
dict(alarm.dimensions).should.equal({"InstanceId": ["i-0123456,i-0123457"]})
list(alarm.alarm_actions).should.equal(["arn:alarm"])
list(alarm.ok_actions).should.equal(["arn:ok"])
list(alarm.insufficient_data_actions).should.equal(["arn:insufficient"])
alarm.unit.should.equal("Seconds")
@mock_cloudwatch_deprecated
def test_delete_alarm():
conn = boto.connect_cloudwatch()
alarms = conn.describe_alarms()
alarms.should.have.length_of(0)
alarm = alarm_fixture()
conn.create_alarm(alarm)
alarms = conn.describe_alarms()
alarms.should.have.length_of(1)
alarms[0].delete()
alarms = conn.describe_alarms()
alarms.should.have.length_of(0)
@mock_cloudwatch_deprecated
def test_put_metric_data():
conn = boto.connect_cloudwatch()
conn.put_metric_data(
namespace="tester",
name="metric",
value=1.5,
dimensions={"InstanceId": ["i-0123456,i-0123457"]},
)
metrics = conn.list_metrics()
metrics.should.have.length_of(1)
metric = metrics[0]
metric.namespace.should.equal("tester")
metric.name.should.equal("metric")
dict(metric.dimensions).should.equal({"InstanceId": ["i-0123456,i-0123457"]})
@mock_cloudwatch_deprecated
def test_describe_alarms():
conn = boto.connect_cloudwatch()
alarms = conn.describe_alarms()
alarms.should.have.length_of(0)
conn.create_alarm(alarm_fixture(name="nfoobar", action="afoobar"))
conn.create_alarm(alarm_fixture(name="nfoobaz", action="afoobaz"))
conn.create_alarm(alarm_fixture(name="nbarfoo", action="abarfoo"))
conn.create_alarm(alarm_fixture(name="nbazfoo", action="abazfoo"))
alarms = conn.describe_alarms()
alarms.should.have.length_of(4)
alarms = conn.describe_alarms(alarm_name_prefix="nfoo")
alarms.should.have.length_of(2)
alarms = conn.describe_alarms(alarm_names=["nfoobar", "nbarfoo", "nbazfoo"])
alarms.should.have.length_of(3)
alarms = conn.describe_alarms(action_prefix="afoo")
alarms.should.have.length_of(2)
for alarm in conn.describe_alarms():
alarm.delete()
alarms = conn.describe_alarms()
alarms.should.have.length_of(0)
import boto
from boto.ec2.cloudwatch.alarm import MetricAlarm
from datetime import datetime
import sure # noqa
from moto import mock_cloudwatch_deprecated
def alarm_fixture(name="tester", action=None):
action = action or ["arn:alarm"]
return MetricAlarm(
name=name,
namespace="{0}_namespace".format(name),
metric="{0}_metric".format(name),
comparison=">=",
threshold=2.0,
period=60,
evaluation_periods=5,
statistic="Average",
description="A test",
dimensions={"InstanceId": ["i-0123456,i-0123457"]},
alarm_actions=action,
ok_actions=["arn:ok"],
insufficient_data_actions=["arn:insufficient"],
unit="Seconds",
)
@mock_cloudwatch_deprecated
def test_create_alarm():
conn = boto.connect_cloudwatch()
alarm = alarm_fixture()
conn.create_alarm(alarm)
alarms = conn.describe_alarms()
alarms.should.have.length_of(1)
alarm = alarms[0]
alarm.name.should.equal("tester")
alarm.namespace.should.equal("tester_namespace")
alarm.metric.should.equal("tester_metric")
alarm.comparison.should.equal(">=")
alarm.threshold.should.equal(2.0)
alarm.period.should.equal(60)
alarm.evaluation_periods.should.equal(5)
alarm.statistic.should.equal("Average")
alarm.description.should.equal("A test")
dict(alarm.dimensions).should.equal({"InstanceId": ["i-0123456,i-0123457"]})
list(alarm.alarm_actions).should.equal(["arn:alarm"])
list(alarm.ok_actions).should.equal(["arn:ok"])
list(alarm.insufficient_data_actions).should.equal(["arn:insufficient"])
alarm.unit.should.equal("Seconds")
@mock_cloudwatch_deprecated
def test_delete_alarm():
conn = boto.connect_cloudwatch()
alarms = conn.describe_alarms()
alarms.should.have.length_of(0)
alarm = alarm_fixture()
conn.create_alarm(alarm)
alarms = conn.describe_alarms()
alarms.should.have.length_of(1)
alarms[0].delete()
alarms = conn.describe_alarms()
alarms.should.have.length_of(0)
@mock_cloudwatch_deprecated
def test_put_metric_data():
conn = boto.connect_cloudwatch()
conn.put_metric_data(
namespace="tester",
name="metric",
value=1.5,
dimensions={"InstanceId": ["i-0123456,i-0123457"]},
)
metrics = conn.list_metrics()
metrics.should.have.length_of(1)
metric = metrics[0]
metric.namespace.should.equal("tester")
metric.name.should.equal("metric")
dict(metric.dimensions).should.equal({"InstanceId": ["i-0123456,i-0123457"]})
@mock_cloudwatch_deprecated
def test_describe_alarms():
conn = boto.connect_cloudwatch()
alarms = conn.describe_alarms()
alarms.should.have.length_of(0)
conn.create_alarm(alarm_fixture(name="nfoobar", action="afoobar"))
conn.create_alarm(alarm_fixture(name="nfoobaz", action="afoobaz"))
conn.create_alarm(alarm_fixture(name="nbarfoo", action="abarfoo"))
conn.create_alarm(alarm_fixture(name="nbazfoo", action="abazfoo"))
enabled = alarm_fixture(name="enabled1", action=["abarfoo"])
enabled.add_alarm_action("arn:alarm")
conn.create_alarm(enabled)
alarms = conn.describe_alarms()
alarms.should.have.length_of(5)
alarms = conn.describe_alarms(alarm_name_prefix="nfoo")
alarms.should.have.length_of(2)
alarms = conn.describe_alarms(alarm_names=["nfoobar", "nbarfoo", "nbazfoo"])
alarms.should.have.length_of(3)
alarms = conn.describe_alarms(action_prefix="afoo")
alarms.should.have.length_of(2)
alarms = conn.describe_alarms(alarm_name_prefix="enabled")
alarms.should.have.length_of(1)
alarms[0].actions_enabled.should.equal("true")
for alarm in conn.describe_alarms():
alarm.delete()
alarms = conn.describe_alarms()
alarms.should.have.length_of(0)
@mock_cloudwatch_deprecated
def test_get_metric_statistics():
conn = boto.connect_cloudwatch()
metric_timestamp = datetime(2018, 4, 9, 13, 0, 0, 0)
conn.put_metric_data(
namespace="tester",
name="metric",
value=1.5,
dimensions={"InstanceId": ["i-0123456,i-0123457"]},
timestamp=metric_timestamp,
)
metric_kwargs = dict(
namespace="tester",
metric_name="metric",
start_time=metric_timestamp,
end_time=datetime.now(),
period=3600,
statistics=["Minimum"],
)
datapoints = conn.get_metric_statistics(**metric_kwargs)
datapoints.should.have.length_of(1)
datapoint = datapoints[0]
datapoint.should.have.key("Minimum").which.should.equal(1.5)
datapoint.should.have.key("Timestamp").which.should.equal(metric_timestamp)

3
tests/test_cloudwatch/test_cloudwatch_boto3.py
View File

@ -104,6 +104,7 @@ def test_alarm_state():
Statistic="Average",
Threshold=2,
ComparisonOperator="GreaterThanThreshold",
ActionsEnabled=True,
)
client.put_metric_alarm(
AlarmName="testalarm2",
@ -128,11 +129,13 @@ def test_alarm_state():
len(resp["MetricAlarms"]).should.equal(1)
resp["MetricAlarms"][0]["AlarmName"].should.equal("testalarm1")
resp["MetricAlarms"][0]["StateValue"].should.equal("ALARM")
resp["MetricAlarms"][0]["ActionsEnabled"].should.equal(True)
resp = client.describe_alarms(StateValue="OK")
len(resp["MetricAlarms"]).should.equal(1)
resp["MetricAlarms"][0]["AlarmName"].should.equal("testalarm2")
resp["MetricAlarms"][0]["StateValue"].should.equal("OK")
resp["MetricAlarms"][0]["ActionsEnabled"].should.equal(False)
# Just for sanity
resp = client.describe_alarms()

34
tests/test_core/test_auth.py
View File

@ -298,6 +298,40 @@ def test_access_denied_with_not_allowing_policy():
)
@set_initial_no_auth_action_count(3)
@mock_ec2
def test_access_denied_for_run_instances():
# https://github.com/spulec/moto/issues/2774
# The run-instances method was broken between botocore versions 1.15.8 and 1.15.12
# This was due to the inclusion of '"idempotencyToken":true' in the response, somehow altering the signature and breaking the authentication
# Keeping this test in place in case botocore decides to break again
user_name = "test-user"
inline_policy_document = {
"Version": "2012-10-17",
"Statement": [
{"Effect": "Allow", "Action": ["ec2:Describe*"], "Resource": "*"}
],
}
access_key = create_user_with_access_key_and_inline_policy(
user_name, inline_policy_document
)
client = boto3.client(
"ec2",
region_name="us-east-1",
aws_access_key_id=access_key["AccessKeyId"],
aws_secret_access_key=access_key["SecretAccessKey"],
)
with assert_raises(ClientError) as ex:
client.run_instances(MaxCount=1, MinCount=1)
ex.exception.response["Error"]["Code"].should.equal("AccessDenied")
ex.exception.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(403)
ex.exception.response["Error"]["Message"].should.equal(
"User: arn:aws:iam::{account_id}:user/{user_name} is not authorized to perform: {operation}".format(
account_id=ACCOUNT_ID, user_name=user_name, operation="ec2:RunInstances",
)
)
@set_initial_no_auth_action_count(3)
@mock_ec2
def test_access_denied_with_denying_policy():

236
tests/test_dynamodb2/test_dynamodb.py
View File

@ -6,8 +6,9 @@ import six
import boto
import boto3
from boto3.dynamodb.conditions import Attr, Key
import sure # noqa
import re
import requests
import sure # noqa
from moto import mock_dynamodb2, mock_dynamodb2_deprecated
from moto.dynamodb2 import dynamodb_backend2, dynamodb_backends2
from boto.exception import JSONResponseError
@ -1344,6 +1345,24 @@ def test_get_item_returns_consumed_capacity():
assert "TableName" in response["ConsumedCapacity"]
@mock_dynamodb2
def test_put_item_nonexisting_hash_key():
dynamodb = boto3.resource("dynamodb", region_name="us-east-1")
dynamodb.create_table(
AttributeDefinitions=[{"AttributeName": "structure_id", "AttributeType": "S"},],
TableName="test",
KeySchema=[{"AttributeName": "structure_id", "KeyType": "HASH"},],
ProvisionedThroughput={"ReadCapacityUnits": 123, "WriteCapacityUnits": 123},
)
table = dynamodb.Table("test")
with assert_raises(ClientError) as ex:
table.put_item(Item={"a_terribly_misguided_id_attribute": "abcdef"})
ex.exception.response["Error"]["Message"].should.equal(
"One or more parameter values were invalid: Missing the key structure_id in the item"
)
def test_filter_expression():
row1 = moto.dynamodb2.models.Item(
None,
@ -3792,3 +3811,218 @@ def test_query_catches_when_no_filters():
ex.exception.response["Error"]["Message"].should.equal(
"Either KeyConditions or QueryFilter should be present"
)
@mock_dynamodb2
def test_invalid_transact_get_items():
dynamodb = boto3.resource("dynamodb", region_name="us-east-1")
dynamodb.create_table(
TableName="test1",
KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],
ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
)
table = dynamodb.Table("test1")
table.put_item(
Item={"id": "1", "val": "1",}
)
table.put_item(
Item={"id": "1", "val": "2",}
)
client = boto3.client("dynamodb", region_name="us-east-1")
with assert_raises(ClientError) as ex:
client.transact_get_items(
TransactItems=[
{"Get": {"Key": {"id": {"S": "1"}}, "TableName": "test1"}}
for i in range(26)
]
)
ex.exception.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
ex.exception.response["Error"]["Message"].should.match(
r"failed to satisfy constraint: Member must have length less than or equal to 25",
re.I,
)
with assert_raises(ClientError) as ex:
client.transact_get_items(
TransactItems=[
{"Get": {"Key": {"id": {"S": "1"},}, "TableName": "test1"}},
{"Get": {"Key": {"id": {"S": "1"},}, "TableName": "non_exists_table"}},
]
)
ex.exception.response["Error"]["Code"].should.equal("ResourceNotFoundException")
ex.exception.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
ex.exception.response["Error"]["Message"].should.equal(
"Requested resource not found"
)
@mock_dynamodb2
def test_valid_transact_get_items():
dynamodb = boto3.resource("dynamodb", region_name="us-east-1")
dynamodb.create_table(
TableName="test1",
KeySchema=[
{"AttributeName": "id", "KeyType": "HASH"},
{"AttributeName": "sort_key", "KeyType": "RANGE"},
],
AttributeDefinitions=[
{"AttributeName": "id", "AttributeType": "S"},
{"AttributeName": "sort_key", "AttributeType": "S"},
],
ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
)
table1 = dynamodb.Table("test1")
table1.put_item(
Item={"id": "1", "sort_key": "1",}
)
table1.put_item(
Item={"id": "1", "sort_key": "2",}
)
dynamodb.create_table(
TableName="test2",
KeySchema=[
{"AttributeName": "id", "KeyType": "HASH"},
{"AttributeName": "sort_key", "KeyType": "RANGE"},
],
AttributeDefinitions=[
{"AttributeName": "id", "AttributeType": "S"},
{"AttributeName": "sort_key", "AttributeType": "S"},
],
ProvisionedThroughput={"ReadCapacityUnits": 5, "WriteCapacityUnits": 5},
)
table2 = dynamodb.Table("test2")
table2.put_item(
Item={"id": "1", "sort_key": "1",}
)
client = boto3.client("dynamodb", region_name="us-east-1")
res = client.transact_get_items(
TransactItems=[
{
"Get": {
"Key": {"id": {"S": "1"}, "sort_key": {"S": "1"}},
"TableName": "test1",
}
},
{
"Get": {
"Key": {"id": {"S": "non_exists_key"}, "sort_key": {"S": "2"}},
"TableName": "test1",
}
},
]
)
res["Responses"][0]["Item"].should.equal({"id": {"S": "1"}, "sort_key": {"S": "1"}})
len(res["Responses"]).should.equal(1)
res = client.transact_get_items(
TransactItems=[
{
"Get": {
"Key": {"id": {"S": "1"}, "sort_key": {"S": "1"}},
"TableName": "test1",
}
},
{
"Get": {
"Key": {"id": {"S": "1"}, "sort_key": {"S": "2"}},
"TableName": "test1",
}
},
{
"Get": {
"Key": {"id": {"S": "1"}, "sort_key": {"S": "1"}},
"TableName": "test2",
}
},
]
)
res["Responses"][0]["Item"].should.equal({"id": {"S": "1"}, "sort_key": {"S": "1"}})
res["Responses"][1]["Item"].should.equal({"id": {"S": "1"}, "sort_key": {"S": "2"}})
res["Responses"][2]["Item"].should.equal({"id": {"S": "1"}, "sort_key": {"S": "1"}})
res = client.transact_get_items(
TransactItems=[
{
"Get": {
"Key": {"id": {"S": "1"}, "sort_key": {"S": "1"}},
"TableName": "test1",
}
},
{
"Get": {
"Key": {"id": {"S": "1"}, "sort_key": {"S": "2"}},
"TableName": "test1",
}
},
{
"Get": {
"Key": {"id": {"S": "1"}, "sort_key": {"S": "1"}},
"TableName": "test2",
}
},
],
ReturnConsumedCapacity="TOTAL",
)
res["ConsumedCapacity"][0].should.equal(
{"TableName": "test1", "CapacityUnits": 4.0, "ReadCapacityUnits": 4.0}
)
res["ConsumedCapacity"][1].should.equal(
{"TableName": "test2", "CapacityUnits": 2.0, "ReadCapacityUnits": 2.0}
)
res = client.transact_get_items(
TransactItems=[
{
"Get": {
"Key": {"id": {"S": "1"}, "sort_key": {"S": "1"}},
"TableName": "test1",
}
},
{
"Get": {
"Key": {"id": {"S": "1"}, "sort_key": {"S": "2"}},
"TableName": "test1",
}
},
{
"Get": {
"Key": {"id": {"S": "1"}, "sort_key": {"S": "1"}},
"TableName": "test2",
}
},
],
ReturnConsumedCapacity="INDEXES",
)
res["ConsumedCapacity"][0].should.equal(
{
"TableName": "test1",
"CapacityUnits": 4.0,
"ReadCapacityUnits": 4.0,
"Table": {"CapacityUnits": 4.0, "ReadCapacityUnits": 4.0,},
}
)
res["ConsumedCapacity"][1].should.equal(
{
"TableName": "test2",
"CapacityUnits": 2.0,
"ReadCapacityUnits": 2.0,
"Table": {"CapacityUnits": 2.0, "ReadCapacityUnits": 2.0,},
}
)

33
tests/test_ec2/test_tags.py
View File

@ -468,3 +468,36 @@ def test_delete_tag_empty_resource():
ex.exception.response["Error"]["Message"].should.equal(
"The request must contain the parameter resourceIdSet"
)
@mock_ec2
def test_retrieve_resource_with_multiple_tags():
ec2 = boto3.resource("ec2", region_name="us-west-1")
blue, green = ec2.create_instances(ImageId="ANY_ID", MinCount=2, MaxCount=2)
ec2.create_tags(
Resources=[blue.instance_id],
Tags=[
{"Key": "environment", "Value": "blue"},
{"Key": "application", "Value": "api"},
],
)
ec2.create_tags(
Resources=[green.instance_id],
Tags=[
{"Key": "environment", "Value": "green"},
{"Key": "application", "Value": "api"},
],
)
green_instances = list(ec2.instances.filter(Filters=(get_filter("green"))))
green_instances.should.equal([green])
blue_instances = list(ec2.instances.filter(Filters=(get_filter("blue"))))
blue_instances.should.equal([blue])
def get_filter(color):
return [
{"Name": "tag-key", "Values": ["application"]},
{"Name": "tag-value", "Values": ["api"]},
{"Name": "tag-key", "Values": ["environment"]},
{"Name": "tag-value", "Values": [color]},
]

17
tests/test_s3/test_s3.py
View File

@ -11,6 +11,7 @@ from six.moves.urllib.error import HTTPError
from functools import wraps
from gzip import GzipFile
from io import BytesIO
import mimetypes
import zlib
import pickle
@ -2024,6 +2025,22 @@ def test_boto3_get_object():
e.exception.response["Error"]["Code"].should.equal("NoSuchKey")
@mock_s3
def test_boto3_s3_content_type():
s3 = boto3.resource("s3", region_name=DEFAULT_REGION_NAME)
my_bucket = s3.Bucket("my-cool-bucket")
my_bucket.create()
s3_path = "test_s3.py"
s3 = boto3.resource("s3", verify=False)
content_type = "text/python-x"
s3.Object(my_bucket.name, s3_path).put(
ContentType=content_type, Body=b"some python code", ACL="public-read"
)
s3.Object(my_bucket.name, s3_path).content_type.should.equal(content_type)
@mock_s3
def test_boto3_get_missing_object_with_part_number():
s3 = boto3.resource("s3", region_name=DEFAULT_REGION_NAME)

419
tests/test_ssm/test_ssm_boto3.py
View File

@ -897,6 +897,7 @@ def test_get_parameter_history():
param["Value"].should.equal("value-%d" % index)
param["Version"].should.equal(index + 1)
param["Description"].should.equal("A test parameter version %d" % index)
param["Labels"].should.equal([])
len(parameters_response).should.equal(3)
@ -938,6 +939,424 @@ def test_get_parameter_history_with_secure_string():
len(parameters_response).should.equal(3)
@mock_ssm
def test_label_parameter_version():
client = boto3.client("ssm", region_name="us-east-1")
test_parameter_name = "test"
client.put_parameter(
Name=test_parameter_name,
Description="A test parameter",
Value="value",
Type="String",
)
response = client.label_parameter_version(
Name=test_parameter_name, Labels=["test-label"]
)
response["InvalidLabels"].should.equal([])
response["ParameterVersion"].should.equal(1)
@mock_ssm
def test_label_parameter_version_with_specific_version():
client = boto3.client("ssm", region_name="us-east-1")
test_parameter_name = "test"
client.put_parameter(
Name=test_parameter_name,
Description="A test parameter",
Value="value",
Type="String",
)
response = client.label_parameter_version(
Name=test_parameter_name, ParameterVersion=1, Labels=["test-label"]
)
response["InvalidLabels"].should.equal([])
response["ParameterVersion"].should.equal(1)
@mock_ssm
def test_label_parameter_version_twice():
client = boto3.client("ssm", region_name="us-east-1")
test_parameter_name = "test"
test_labels = ["test-label"]
client.put_parameter(
Name=test_parameter_name,
Description="A test parameter",
Value="value",
Type="String",
)
response = client.label_parameter_version(
Name=test_parameter_name, ParameterVersion=1, Labels=test_labels
)
response["InvalidLabels"].should.equal([])
response["ParameterVersion"].should.equal(1)
response = client.label_parameter_version(
Name=test_parameter_name, ParameterVersion=1, Labels=test_labels
)
response["InvalidLabels"].should.equal([])
response["ParameterVersion"].should.equal(1)
response = client.get_parameter_history(Name=test_parameter_name)
len(response["Parameters"]).should.equal(1)
response["Parameters"][0]["Labels"].should.equal(test_labels)
@mock_ssm
def test_label_parameter_moving_versions():
client = boto3.client("ssm", region_name="us-east-1")
test_parameter_name = "test"
test_labels = ["test-label"]
for i in range(3):
client.put_parameter(
Name=test_parameter_name,
Description="A test parameter version %d" % i,
Value="value-%d" % i,
Type="String",
Overwrite=True,
)
response = client.label_parameter_version(
Name=test_parameter_name, ParameterVersion=1, Labels=test_labels
)
response["InvalidLabels"].should.equal([])
response["ParameterVersion"].should.equal(1)
response = client.label_parameter_version(
Name=test_parameter_name, ParameterVersion=2, Labels=test_labels
)
response["InvalidLabels"].should.equal([])
response["ParameterVersion"].should.equal(2)
response = client.get_parameter_history(Name=test_parameter_name)
parameters_response = response["Parameters"]
for index, param in enumerate(parameters_response):
param["Name"].should.equal(test_parameter_name)
param["Type"].should.equal("String")
param["Value"].should.equal("value-%d" % index)
param["Version"].should.equal(index + 1)
param["Description"].should.equal("A test parameter version %d" % index)
labels = test_labels if param["Version"] == 2 else []
param["Labels"].should.equal(labels)
len(parameters_response).should.equal(3)
@mock_ssm
def test_label_parameter_moving_versions_complex():
client = boto3.client("ssm", region_name="us-east-1")
test_parameter_name = "test"
for i in range(3):
client.put_parameter(
Name=test_parameter_name,
Description="A test parameter version %d" % i,
Value="value-%d" % i,
Type="String",
Overwrite=True,
)
response = client.label_parameter_version(
Name=test_parameter_name,
ParameterVersion=1,
Labels=["test-label1", "test-label2", "test-label3"],
)
response["InvalidLabels"].should.equal([])
response["ParameterVersion"].should.equal(1)
response = client.label_parameter_version(
Name=test_parameter_name,
ParameterVersion=2,
Labels=["test-label2", "test-label3"],
)
response["InvalidLabels"].should.equal([])
response["ParameterVersion"].should.equal(2)
response = client.get_parameter_history(Name=test_parameter_name)
parameters_response = response["Parameters"]
for index, param in enumerate(parameters_response):
param["Name"].should.equal(test_parameter_name)
param["Type"].should.equal("String")
param["Value"].should.equal("value-%d" % index)
param["Version"].should.equal(index + 1)
param["Description"].should.equal("A test parameter version %d" % index)
labels = (
["test-label2", "test-label3"]
if param["Version"] == 2
else (["test-label1"] if param["Version"] == 1 else [])
)
param["Labels"].should.equal(labels)
len(parameters_response).should.equal(3)
@mock_ssm
def test_label_parameter_version_exception_ten_labels_at_once():
client = boto3.client("ssm", region_name="us-east-1")
test_parameter_name = "test"
test_labels = [
"test-label1",
"test-label2",
"test-label3",
"test-label4",
"test-label5",
"test-label6",
"test-label7",
"test-label8",
"test-label9",
"test-label10",
"test-label11",
]
client.put_parameter(
Name=test_parameter_name,
Description="A test parameter",
Value="value",
Type="String",
)
client.label_parameter_version.when.called_with(
Name="test", ParameterVersion=1, Labels=test_labels
).should.throw(
ClientError,
"An error occurred (ParameterVersionLabelLimitExceeded) when calling the LabelParameterVersion operation: "
"A parameter version can have maximum 10 labels."
"Move one or more labels to another version and try again.",
)
@mock_ssm
def test_label_parameter_version_exception_ten_labels_over_multiple_calls():
client = boto3.client("ssm", region_name="us-east-1")
test_parameter_name = "test"
client.put_parameter(
Name=test_parameter_name,
Description="A test parameter",
Value="value",
Type="String",
)
client.label_parameter_version(
Name=test_parameter_name,
ParameterVersion=1,
Labels=[
"test-label1",
"test-label2",
"test-label3",
"test-label4",
"test-label5",
],
)
client.label_parameter_version.when.called_with(
Name="test",
ParameterVersion=1,
Labels=[
"test-label6",
"test-label7",
"test-label8",
"test-label9",
"test-label10",
"test-label11",
],
).should.throw(
ClientError,
"An error occurred (ParameterVersionLabelLimitExceeded) when calling the LabelParameterVersion operation: "
"A parameter version can have maximum 10 labels."
"Move one or more labels to another version and try again.",
)
@mock_ssm
def test_label_parameter_version_invalid_name():
client = boto3.client("ssm", region_name="us-east-1")
test_parameter_name = "test"
response = client.label_parameter_version.when.called_with(
Name=test_parameter_name, Labels=["test-label"]
).should.throw(
ClientError,
"An error occurred (ParameterNotFound) when calling the LabelParameterVersion operation: "
"Parameter test not found.",
)
@mock_ssm
def test_label_parameter_version_invalid_parameter_version():
client = boto3.client("ssm", region_name="us-east-1")
test_parameter_name = "test"
client.put_parameter(
Name=test_parameter_name,
Description="A test parameter",
Value="value",
Type="String",
)
response = client.label_parameter_version.when.called_with(
Name=test_parameter_name, Labels=["test-label"], ParameterVersion=5
).should.throw(
ClientError,
"An error occurred (ParameterVersionNotFound) when calling the LabelParameterVersion operation: "
"Systems Manager could not find version 5 of test. "
"Verify the version and try again.",
)
@mock_ssm
def test_label_parameter_version_invalid_label():
client = boto3.client("ssm", region_name="us-east-1")
test_parameter_name = "test"
client.put_parameter(
Name=test_parameter_name,
Description="A test parameter",
Value="value",
Type="String",
)
response = client.label_parameter_version(
Name=test_parameter_name, ParameterVersion=1, Labels=["awsabc"]
)
response["InvalidLabels"].should.equal(["awsabc"])
response = client.label_parameter_version(
Name=test_parameter_name, ParameterVersion=1, Labels=["ssmabc"]
)
response["InvalidLabels"].should.equal(["ssmabc"])
response = client.label_parameter_version(
Name=test_parameter_name, ParameterVersion=1, Labels=["9abc"]
)
response["InvalidLabels"].should.equal(["9abc"])
response = client.label_parameter_version(
Name=test_parameter_name, ParameterVersion=1, Labels=["abc/123"]
)
response["InvalidLabels"].should.equal(["abc/123"])
client.label_parameter_version.when.called_with(
Name=test_parameter_name, ParameterVersion=1, Labels=["a" * 101]
).should.throw(
ClientError,
"1 validation error detected: "
"Value '[%s]' at 'labels' failed to satisfy constraint: "
"Member must satisfy constraint: "
"[Member must have length less than or equal to 100, Member must have length greater than or equal to 1]"
% ("a" * 101),
)
@mock_ssm
def test_get_parameter_history_with_label():
client = boto3.client("ssm", region_name="us-east-1")
test_parameter_name = "test"
test_labels = ["test-label"]
for i in range(3):
client.put_parameter(
Name=test_parameter_name,
Description="A test parameter version %d" % i,
Value="value-%d" % i,
Type="String",
Overwrite=True,
)
client.label_parameter_version(
Name=test_parameter_name, ParameterVersion=1, Labels=test_labels
)
response = client.get_parameter_history(Name=test_parameter_name)
parameters_response = response["Parameters"]
for index, param in enumerate(parameters_response):
param["Name"].should.equal(test_parameter_name)
param["Type"].should.equal("String")
param["Value"].should.equal("value-%d" % index)
param["Version"].should.equal(index + 1)
param["Description"].should.equal("A test parameter version %d" % index)
labels = test_labels if param["Version"] == 1 else []
param["Labels"].should.equal(labels)
len(parameters_response).should.equal(3)
@mock_ssm
def test_get_parameter_history_with_label_non_latest():
client = boto3.client("ssm", region_name="us-east-1")
test_parameter_name = "test"
test_labels = ["test-label"]
for i in range(3):
client.put_parameter(
Name=test_parameter_name,
Description="A test parameter version %d" % i,
Value="value-%d" % i,
Type="String",
Overwrite=True,
)
client.label_parameter_version(
Name=test_parameter_name, ParameterVersion=2, Labels=test_labels
)
response = client.get_parameter_history(Name=test_parameter_name)
parameters_response = response["Parameters"]
for index, param in enumerate(parameters_response):
param["Name"].should.equal(test_parameter_name)
param["Type"].should.equal("String")
param["Value"].should.equal("value-%d" % index)
param["Version"].should.equal(index + 1)
param["Description"].should.equal("A test parameter version %d" % index)
labels = test_labels if param["Version"] == 2 else []
param["Labels"].should.equal(labels)
len(parameters_response).should.equal(3)
@mock_ssm
def test_get_parameter_history_with_label_latest_assumed():
client = boto3.client("ssm", region_name="us-east-1")
test_parameter_name = "test"
test_labels = ["test-label"]
for i in range(3):
client.put_parameter(
Name=test_parameter_name,
Description="A test parameter version %d" % i,
Value="value-%d" % i,
Type="String",
Overwrite=True,
)
client.label_parameter_version(Name=test_parameter_name, Labels=test_labels)
response = client.get_parameter_history(Name=test_parameter_name)
parameters_response = response["Parameters"]
for index, param in enumerate(parameters_response):
param["Name"].should.equal(test_parameter_name)
param["Type"].should.equal("String")
param["Value"].should.equal("value-%d" % index)
param["Version"].should.equal(index + 1)
param["Description"].should.equal("A test parameter version %d" % index)
labels = test_labels if param["Version"] == 3 else []
param["Labels"].should.equal(labels)
len(parameters_response).should.equal(3)
@mock_ssm
def test_get_parameter_history_missing_parameter():
client = boto3.client("ssm", region_name="us-east-1")