catalin/moto
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Merge remote-tracking branch 'spulec/master' into improve_coverage

Browse source
This commit is contained in:
Bert Blommers 2020-05-12 15:02:24 +01:00
commit d680b1e025
45 changed files with 2446 additions and 159 deletions
Download patch file Download diff file Expand all files Collapse all files

60
tests/test_apigateway/test_apigateway.py
View file

@ -1,5 +1,6 @@
from __future__ import unicode_literals
import json
import boto3
from freezegun import freeze_time
@ -1230,6 +1231,65 @@ def test_put_integration_response_requires_responseTemplate():
)
@mock_apigateway
def test_put_integration_response_with_response_template():
client = boto3.client("apigateway", region_name="us-west-2")
response = client.create_rest_api(name="my_api", description="this is my api")
api_id = response["id"]
resources = client.get_resources(restApiId=api_id)
root_id = [resource for resource in resources["items"] if resource["path"] == "/"][
0
]["id"]
client.put_method(
restApiId=api_id, resourceId=root_id, httpMethod="GET", authorizationType="NONE"
)
client.put_method_response(
restApiId=api_id, resourceId=root_id, httpMethod="GET", statusCode="200"
)
client.put_integration(
restApiId=api_id,
resourceId=root_id,
httpMethod="GET",
type="HTTP",
uri="http://httpbin.org/robots.txt",
integrationHttpMethod="POST",
)
with assert_raises(ClientError) as ex:
client.put_integration_response(
restApiId=api_id, resourceId=root_id, httpMethod="GET", statusCode="200"
)
ex.exception.response["Error"]["Code"].should.equal("BadRequestException")
ex.exception.response["Error"]["Message"].should.equal("Invalid request input")
client.put_integration_response(
restApiId=api_id,
resourceId=root_id,
httpMethod="GET",
statusCode="200",
selectionPattern="foobar",
responseTemplates={"application/json": json.dumps({"data": "test"})},
)
response = client.get_integration_response(
restApiId=api_id, resourceId=root_id, httpMethod="GET", statusCode="200"
)
# this is hard to match against, so remove it
response["ResponseMetadata"].pop("HTTPHeaders", None)
response["ResponseMetadata"].pop("RetryAttempts", None)
response.should.equal(
{
"statusCode": "200",
"selectionPattern": "foobar",
"ResponseMetadata": {"HTTPStatusCode": 200},
"responseTemplates": {"application/json": json.dumps({"data": "test"})},
}
)
@mock_apigateway
def test_put_integration_validation():
client = boto3.client("apigateway", region_name="us-west-2")

232
tests/test_autoscaling/test_autoscaling.py
View file

@ -1071,6 +1071,7 @@ def test_autoscaling_describe_policies_boto3():
response["ScalingPolicies"][0]["PolicyName"].should.equal("test_policy_down")
@mock_elb
@mock_autoscaling
@mock_ec2
def test_detach_one_instance_decrement():
@ -1096,6 +1097,19 @@ def test_detach_one_instance_decrement():
],
VPCZoneIdentifier=mocked_networking["subnet1"],
)
elb_client = boto3.client("elb", region_name="us-east-1")
elb_client.create_load_balancer(
LoadBalancerName="my-lb",
Listeners=[{"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080}],
AvailabilityZones=["us-east-1a", "us-east-1b"],
)
response = client.attach_load_balancers(
AutoScalingGroupName="test_asg", LoadBalancerNames=["my-lb"]
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
instance_to_detach = response["AutoScalingGroups"][0]["Instances"][0]["InstanceId"]
instance_to_keep = response["AutoScalingGroups"][0]["Instances"][1]["InstanceId"]
@ -1111,6 +1125,9 @@ def test_detach_one_instance_decrement():
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
response["AutoScalingGroups"][0]["Instances"].should.have.length_of(1)
instance_to_detach.shouldnt.be.within(
[x["InstanceId"] for x in response["AutoScalingGroups"][0]["Instances"]]
)
# test to ensure tag has been removed
response = ec2_client.describe_instances(InstanceIds=[instance_to_detach])
@ -1122,7 +1139,14 @@ def test_detach_one_instance_decrement():
tags = response["Reservations"][0]["Instances"][0]["Tags"]
tags.should.have.length_of(2)
response = elb_client.describe_load_balancers(LoadBalancerNames=["my-lb"])
list(response["LoadBalancerDescriptions"][0]["Instances"]).should.have.length_of(1)
instance_to_detach.shouldnt.be.within(
[x["InstanceId"] for x in response["LoadBalancerDescriptions"][0]["Instances"]]
)
@mock_elb
@mock_autoscaling
@mock_ec2
def test_detach_one_instance():
@ -1148,6 +1172,19 @@ def test_detach_one_instance():
],
VPCZoneIdentifier=mocked_networking["subnet1"],
)
elb_client = boto3.client("elb", region_name="us-east-1")
elb_client.create_load_balancer(
LoadBalancerName="my-lb",
Listeners=[{"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080}],
AvailabilityZones=["us-east-1a", "us-east-1b"],
)
response = client.attach_load_balancers(
AutoScalingGroupName="test_asg", LoadBalancerNames=["my-lb"]
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
instance_to_detach = response["AutoScalingGroups"][0]["Instances"][0]["InstanceId"]
instance_to_keep = response["AutoScalingGroups"][0]["Instances"][1]["InstanceId"]
@ -1173,7 +1210,14 @@ def test_detach_one_instance():
tags = response["Reservations"][0]["Instances"][0]["Tags"]
tags.should.have.length_of(2)
response = elb_client.describe_load_balancers(LoadBalancerNames=["my-lb"])
list(response["LoadBalancerDescriptions"][0]["Instances"]).should.have.length_of(2)
instance_to_detach.shouldnt.be.within(
[x["InstanceId"] for x in response["LoadBalancerDescriptions"][0]["Instances"]]
)
@mock_elb
@mock_autoscaling
@mock_ec2
def test_standby_one_instance_decrement():
@ -1199,6 +1243,19 @@ def test_standby_one_instance_decrement():
],
VPCZoneIdentifier=mocked_networking["subnet1"],
)
elb_client = boto3.client("elb", region_name="us-east-1")
elb_client.create_load_balancer(
LoadBalancerName="my-lb",
Listeners=[{"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080}],
AvailabilityZones=["us-east-1a", "us-east-1b"],
)
response = client.attach_load_balancers(
AutoScalingGroupName="test_asg", LoadBalancerNames=["my-lb"]
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
instance_to_standby = response["AutoScalingGroups"][0]["Instances"][0]["InstanceId"]
instance_to_keep = response["AutoScalingGroups"][0]["Instances"][1]["InstanceId"]
@ -1226,7 +1283,14 @@ def test_standby_one_instance_decrement():
tags = instance["Tags"]
tags.should.have.length_of(2)
response = elb_client.describe_load_balancers(LoadBalancerNames=["my-lb"])
list(response["LoadBalancerDescriptions"][0]["Instances"]).should.have.length_of(1)
instance_to_standby.shouldnt.be.within(
[x["InstanceId"] for x in response["LoadBalancerDescriptions"][0]["Instances"]]
)
@mock_elb
@mock_autoscaling
@mock_ec2
def test_standby_one_instance():
@ -1252,6 +1316,19 @@ def test_standby_one_instance():
],
VPCZoneIdentifier=mocked_networking["subnet1"],
)
elb_client = boto3.client("elb", region_name="us-east-1")
elb_client.create_load_balancer(
LoadBalancerName="my-lb",
Listeners=[{"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080}],
AvailabilityZones=["us-east-1a", "us-east-1b"],
)
response = client.attach_load_balancers(
AutoScalingGroupName="test_asg", LoadBalancerNames=["my-lb"]
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
instance_to_standby = response["AutoScalingGroups"][0]["Instances"][0]["InstanceId"]
instance_to_keep = response["AutoScalingGroups"][0]["Instances"][1]["InstanceId"]
@ -1279,6 +1356,12 @@ def test_standby_one_instance():
tags = instance["Tags"]
tags.should.have.length_of(2)
response = elb_client.describe_load_balancers(LoadBalancerNames=["my-lb"])
list(response["LoadBalancerDescriptions"][0]["Instances"]).should.have.length_of(2)
instance_to_standby.shouldnt.be.within(
[x["InstanceId"] for x in response["LoadBalancerDescriptions"][0]["Instances"]]
)
@mock_elb
@mock_autoscaling
@ -1338,8 +1421,12 @@ def test_standby_elb_update():
response = elb_client.describe_load_balancers(LoadBalancerNames=["my-lb"])
list(response["LoadBalancerDescriptions"][0]["Instances"]).should.have.length_of(2)
instance_to_standby.shouldnt.be.within(
[x["InstanceId"] for x in response["LoadBalancerDescriptions"][0]["Instances"]]
)
@mock_elb
@mock_autoscaling
@mock_ec2
def test_standby_terminate_instance_decrement():
@ -1366,6 +1453,18 @@ def test_standby_terminate_instance_decrement():
VPCZoneIdentifier=mocked_networking["subnet1"],
)
elb_client = boto3.client("elb", region_name="us-east-1")
elb_client.create_load_balancer(
LoadBalancerName="my-lb",
Listeners=[{"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080}],
AvailabilityZones=["us-east-1a", "us-east-1b"],
)
response = client.attach_load_balancers(
AutoScalingGroupName="test_asg", LoadBalancerNames=["my-lb"]
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
instance_to_standby_terminate = response["AutoScalingGroups"][0]["Instances"][0][
"InstanceId"
@ -1409,7 +1508,14 @@ def test_standby_terminate_instance_decrement():
"terminated"
)
response = elb_client.describe_load_balancers(LoadBalancerNames=["my-lb"])
list(response["LoadBalancerDescriptions"][0]["Instances"]).should.have.length_of(1)
instance_to_standby_terminate.shouldnt.be.within(
[x["InstanceId"] for x in response["LoadBalancerDescriptions"][0]["Instances"]]
)
@mock_elb
@mock_autoscaling
@mock_ec2
def test_standby_terminate_instance_no_decrement():
@ -1436,6 +1542,18 @@ def test_standby_terminate_instance_no_decrement():
VPCZoneIdentifier=mocked_networking["subnet1"],
)
elb_client = boto3.client("elb", region_name="us-east-1")
elb_client.create_load_balancer(
LoadBalancerName="my-lb",
Listeners=[{"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080}],
AvailabilityZones=["us-east-1a", "us-east-1b"],
)
response = client.attach_load_balancers(
AutoScalingGroupName="test_asg", LoadBalancerNames=["my-lb"]
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
instance_to_standby_terminate = response["AutoScalingGroups"][0]["Instances"][0][
"InstanceId"
@ -1479,7 +1597,14 @@ def test_standby_terminate_instance_no_decrement():
"terminated"
)
response = elb_client.describe_load_balancers(LoadBalancerNames=["my-lb"])
list(response["LoadBalancerDescriptions"][0]["Instances"]).should.have.length_of(2)
instance_to_standby_terminate.shouldnt.be.within(
[x["InstanceId"] for x in response["LoadBalancerDescriptions"][0]["Instances"]]
)
@mock_elb
@mock_autoscaling
@mock_ec2
def test_standby_detach_instance_decrement():
@ -1506,6 +1631,18 @@ def test_standby_detach_instance_decrement():
VPCZoneIdentifier=mocked_networking["subnet1"],
)
elb_client = boto3.client("elb", region_name="us-east-1")
elb_client.create_load_balancer(
LoadBalancerName="my-lb",
Listeners=[{"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080}],
AvailabilityZones=["us-east-1a", "us-east-1b"],
)
response = client.attach_load_balancers(
AutoScalingGroupName="test_asg", LoadBalancerNames=["my-lb"]
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
instance_to_standby_detach = response["AutoScalingGroups"][0]["Instances"][0][
"InstanceId"
@ -1547,7 +1684,14 @@ def test_standby_detach_instance_decrement():
response = ec2_client.describe_instances(InstanceIds=[instance_to_standby_detach])
response["Reservations"][0]["Instances"][0]["State"]["Name"].should.equal("running")
response = elb_client.describe_load_balancers(LoadBalancerNames=["my-lb"])
list(response["LoadBalancerDescriptions"][0]["Instances"]).should.have.length_of(1)
instance_to_standby_detach.shouldnt.be.within(
[x["InstanceId"] for x in response["LoadBalancerDescriptions"][0]["Instances"]]
)
@mock_elb
@mock_autoscaling
@mock_ec2
def test_standby_detach_instance_no_decrement():
@ -1574,6 +1718,18 @@ def test_standby_detach_instance_no_decrement():
VPCZoneIdentifier=mocked_networking["subnet1"],
)
elb_client = boto3.client("elb", region_name="us-east-1")
elb_client.create_load_balancer(
LoadBalancerName="my-lb",
Listeners=[{"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080}],
AvailabilityZones=["us-east-1a", "us-east-1b"],
)
response = client.attach_load_balancers(
AutoScalingGroupName="test_asg", LoadBalancerNames=["my-lb"]
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
instance_to_standby_detach = response["AutoScalingGroups"][0]["Instances"][0][
"InstanceId"
@ -1615,7 +1771,14 @@ def test_standby_detach_instance_no_decrement():
response = ec2_client.describe_instances(InstanceIds=[instance_to_standby_detach])
response["Reservations"][0]["Instances"][0]["State"]["Name"].should.equal("running")
response = elb_client.describe_load_balancers(LoadBalancerNames=["my-lb"])
list(response["LoadBalancerDescriptions"][0]["Instances"]).should.have.length_of(2)
instance_to_standby_detach.shouldnt.be.within(
[x["InstanceId"] for x in response["LoadBalancerDescriptions"][0]["Instances"]]
)
@mock_elb
@mock_autoscaling
@mock_ec2
def test_standby_exit_standby():
@ -1642,6 +1805,18 @@ def test_standby_exit_standby():
VPCZoneIdentifier=mocked_networking["subnet1"],
)
elb_client = boto3.client("elb", region_name="us-east-1")
elb_client.create_load_balancer(
LoadBalancerName="my-lb",
Listeners=[{"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080}],
AvailabilityZones=["us-east-1a", "us-east-1b"],
)
response = client.attach_load_balancers(
AutoScalingGroupName="test_asg", LoadBalancerNames=["my-lb"]
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
instance_to_standby_exit_standby = response["AutoScalingGroups"][0]["Instances"][0][
"InstanceId"
@ -1683,7 +1858,14 @@ def test_standby_exit_standby():
)
response["Reservations"][0]["Instances"][0]["State"]["Name"].should.equal("running")
response = elb_client.describe_load_balancers(LoadBalancerNames=["my-lb"])
list(response["LoadBalancerDescriptions"][0]["Instances"]).should.have.length_of(3)
instance_to_standby_exit_standby.should.be.within(
[x["InstanceId"] for x in response["LoadBalancerDescriptions"][0]["Instances"]]
)
@mock_elb
@mock_autoscaling
@mock_ec2
def test_attach_one_instance():
@ -1711,6 +1893,18 @@ def test_attach_one_instance():
NewInstancesProtectedFromScaleIn=True,
)
elb_client = boto3.client("elb", region_name="us-east-1")
elb_client.create_load_balancer(
LoadBalancerName="my-lb",
Listeners=[{"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080}],
AvailabilityZones=["us-east-1a", "us-east-1b"],
)
response = client.attach_load_balancers(
AutoScalingGroupName="test_asg", LoadBalancerNames=["my-lb"]
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
ec2 = boto3.resource("ec2", "us-east-1")
instances_to_add = [
x.id for x in ec2.create_instances(ImageId="", MinCount=1, MaxCount=1)
@ -1727,6 +1921,9 @@ def test_attach_one_instance():
for instance in instances:
instance["ProtectedFromScaleIn"].should.equal(True)
response = elb_client.describe_load_balancers(LoadBalancerNames=["my-lb"])
list(response["LoadBalancerDescriptions"][0]["Instances"]).should.have.length_of(3)
@mock_autoscaling
@mock_ec2
@ -1948,6 +2145,7 @@ def test_terminate_instance_via_ec2_in_autoscaling_group():
replaced_instance_id.should_not.equal(original_instance_id)
@mock_elb
@mock_autoscaling
@mock_ec2
def test_terminate_instance_in_auto_scaling_group_decrement():
@ -1966,6 +2164,18 @@ def test_terminate_instance_in_auto_scaling_group_decrement():
NewInstancesProtectedFromScaleIn=False,
)
elb_client = boto3.client("elb", region_name="us-east-1")
elb_client.create_load_balancer(
LoadBalancerName="my-lb",
Listeners=[{"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080}],
AvailabilityZones=["us-east-1a", "us-east-1b"],
)
response = client.attach_load_balancers(
AutoScalingGroupName="test_asg", LoadBalancerNames=["my-lb"]
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
original_instance_id = next(
instance["InstanceId"]
@ -1979,7 +2189,11 @@ def test_terminate_instance_in_auto_scaling_group_decrement():
response["AutoScalingGroups"][0]["Instances"].should.equal([])
response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(0)
response = elb_client.describe_load_balancers(LoadBalancerNames=["my-lb"])
list(response["LoadBalancerDescriptions"][0]["Instances"]).should.have.length_of(0)
@mock_elb
@mock_autoscaling
@mock_ec2
def test_terminate_instance_in_auto_scaling_group_no_decrement():
@ -1998,6 +2212,18 @@ def test_terminate_instance_in_auto_scaling_group_no_decrement():
NewInstancesProtectedFromScaleIn=False,
)
elb_client = boto3.client("elb", region_name="us-east-1")
elb_client.create_load_balancer(
LoadBalancerName="my-lb",
Listeners=[{"Protocol": "tcp", "LoadBalancerPort": 80, "InstancePort": 8080}],
AvailabilityZones=["us-east-1a", "us-east-1b"],
)
response = client.attach_load_balancers(
AutoScalingGroupName="test_asg", LoadBalancerNames=["my-lb"]
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
original_instance_id = next(
instance["InstanceId"]
@ -2014,3 +2240,9 @@ def test_terminate_instance_in_auto_scaling_group_no_decrement():
)
replaced_instance_id.should_not.equal(original_instance_id)
response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(1)
response = elb_client.describe_load_balancers(LoadBalancerNames=["my-lb"])
list(response["LoadBalancerDescriptions"][0]["Instances"]).should.have.length_of(1)
original_instance_id.shouldnt.be.within(
[x["InstanceId"] for x in response["LoadBalancerDescriptions"][0]["Instances"]]
)

214
tests/test_dynamodb2/test_dynamodb.py
View file

@ -1,5 +1,6 @@
from __future__ import unicode_literals, print_function
from datetime import datetime
from decimal import Decimal
import boto
@ -2049,6 +2050,141 @@ def test_set_ttl():
resp["TimeToLiveDescription"]["TimeToLiveStatus"].should.equal("DISABLED")
@mock_dynamodb2
def test_describe_continuous_backups():
# given
client = boto3.client("dynamodb", region_name="us-east-1")
table_name = client.create_table(
TableName="test",
AttributeDefinitions=[
{"AttributeName": "client", "AttributeType": "S"},
{"AttributeName": "app", "AttributeType": "S"},
],
KeySchema=[
{"AttributeName": "client", "KeyType": "HASH"},
{"AttributeName": "app", "KeyType": "RANGE"},
],
BillingMode="PAY_PER_REQUEST",
)["TableDescription"]["TableName"]
# when
response = client.describe_continuous_backups(TableName=table_name)
# then
response["ContinuousBackupsDescription"].should.equal(
{
"ContinuousBackupsStatus": "ENABLED",
"PointInTimeRecoveryDescription": {"PointInTimeRecoveryStatus": "DISABLED"},
}
)
@mock_dynamodb2
def test_describe_continuous_backups_errors():
# given
client = boto3.client("dynamodb", region_name="us-east-1")
# when
with assert_raises(Exception) as e:
client.describe_continuous_backups(TableName="not-existing-table")
# then
ex = e.exception
ex.operation_name.should.equal("DescribeContinuousBackups")
ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
ex.response["Error"]["Code"].should.contain("TableNotFoundException")
ex.response["Error"]["Message"].should.equal("Table not found: not-existing-table")
@mock_dynamodb2
def test_update_continuous_backups():
# given
client = boto3.client("dynamodb", region_name="us-east-1")
table_name = client.create_table(
TableName="test",
AttributeDefinitions=[
{"AttributeName": "client", "AttributeType": "S"},
{"AttributeName": "app", "AttributeType": "S"},
],
KeySchema=[
{"AttributeName": "client", "KeyType": "HASH"},
{"AttributeName": "app", "KeyType": "RANGE"},
],
BillingMode="PAY_PER_REQUEST",
)["TableDescription"]["TableName"]
# when
response = client.update_continuous_backups(
TableName=table_name,
PointInTimeRecoverySpecification={"PointInTimeRecoveryEnabled": True},
)
# then
response["ContinuousBackupsDescription"]["ContinuousBackupsStatus"].should.equal(
"ENABLED"
)
point_in_time = response["ContinuousBackupsDescription"][
"PointInTimeRecoveryDescription"
]
earliest_datetime = point_in_time["EarliestRestorableDateTime"]
earliest_datetime.should.be.a(datetime)
latest_datetime = point_in_time["LatestRestorableDateTime"]
latest_datetime.should.be.a(datetime)
point_in_time["PointInTimeRecoveryStatus"].should.equal("ENABLED")
# when
# a second update should not change anything
response = client.update_continuous_backups(
TableName=table_name,
PointInTimeRecoverySpecification={"PointInTimeRecoveryEnabled": True},
)
# then
response["ContinuousBackupsDescription"]["ContinuousBackupsStatus"].should.equal(
"ENABLED"
)
point_in_time = response["ContinuousBackupsDescription"][
"PointInTimeRecoveryDescription"
]
point_in_time["EarliestRestorableDateTime"].should.equal(earliest_datetime)
point_in_time["LatestRestorableDateTime"].should.equal(latest_datetime)
point_in_time["PointInTimeRecoveryStatus"].should.equal("ENABLED")
# when
response = client.update_continuous_backups(
TableName=table_name,
PointInTimeRecoverySpecification={"PointInTimeRecoveryEnabled": False},
)
# then
response["ContinuousBackupsDescription"].should.equal(
{
"ContinuousBackupsStatus": "ENABLED",
"PointInTimeRecoveryDescription": {"PointInTimeRecoveryStatus": "DISABLED"},
}
)
@mock_dynamodb2
def test_update_continuous_backups_errors():
# given
client = boto3.client("dynamodb", region_name="us-east-1")
# when
with assert_raises(Exception) as e:
client.update_continuous_backups(
TableName="not-existing-table",
PointInTimeRecoverySpecification={"PointInTimeRecoveryEnabled": True},
)
# then
ex = e.exception
ex.operation_name.should.equal("UpdateContinuousBackups")
ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
ex.response["Error"]["Code"].should.contain("TableNotFoundException")
ex.response["Error"]["Message"].should.equal("Table not found: not-existing-table")
# https://github.com/spulec/moto/issues/1043
@mock_dynamodb2
def test_query_missing_expr_names():
@ -5029,3 +5165,81 @@ def test_update_item_atomic_counter_return_values():
"v" in response["Attributes"]
), "v has been updated, and should be returned here"
response["Attributes"]["v"]["N"].should.equal("8")
@mock_dynamodb2
def test_update_item_atomic_counter_from_zero():
table = "table_t"
ddb_mock = boto3.client("dynamodb", region_name="eu-west-1")
ddb_mock.create_table(
TableName=table,
KeySchema=[{"AttributeName": "t_id", "KeyType": "HASH"}],
AttributeDefinitions=[{"AttributeName": "t_id", "AttributeType": "S"}],
BillingMode="PAY_PER_REQUEST",
)
key = {"t_id": {"S": "item1"}}
ddb_mock.put_item(
TableName=table, Item=key,
)
ddb_mock.update_item(
TableName=table,
Key=key,
UpdateExpression="add n_i :inc1, n_f :inc2",
ExpressionAttributeValues={":inc1": {"N": "1.2"}, ":inc2": {"N": "-0.5"}},
)
updated_item = ddb_mock.get_item(TableName=table, Key=key)["Item"]
assert updated_item["n_i"]["N"] == "1.2"
assert updated_item["n_f"]["N"] == "-0.5"
@mock_dynamodb2
def test_update_item_add_to_non_existent_set():
table = "table_t"
ddb_mock = boto3.client("dynamodb", region_name="eu-west-1")
ddb_mock.create_table(
TableName=table,
KeySchema=[{"AttributeName": "t_id", "KeyType": "HASH"}],
AttributeDefinitions=[{"AttributeName": "t_id", "AttributeType": "S"}],
BillingMode="PAY_PER_REQUEST",
)
key = {"t_id": {"S": "item1"}}
ddb_mock.put_item(
TableName=table, Item=key,
)
ddb_mock.update_item(
TableName=table,
Key=key,
UpdateExpression="add s_i :s1",
ExpressionAttributeValues={":s1": {"SS": ["hello"]}},
)
updated_item = ddb_mock.get_item(TableName=table, Key=key)["Item"]
assert updated_item["s_i"]["SS"] == ["hello"]
@mock_dynamodb2
def test_update_item_add_to_non_existent_number_set():
table = "table_t"
ddb_mock = boto3.client("dynamodb", region_name="eu-west-1")
ddb_mock.create_table(
TableName=table,
KeySchema=[{"AttributeName": "t_id", "KeyType": "HASH"}],
AttributeDefinitions=[{"AttributeName": "t_id", "AttributeType": "S"}],
BillingMode="PAY_PER_REQUEST",
)
key = {"t_id": {"S": "item1"}}
ddb_mock.put_item(
TableName=table, Item=key,
)
ddb_mock.update_item(
TableName=table,
Key=key,
UpdateExpression="add s_i :s1",
ExpressionAttributeValues={":s1": {"NS": ["3"]}},
)
updated_item = ddb_mock.get_item(TableName=table, Key=key)["Item"]
assert updated_item["s_i"]["NS"] == ["3"]

23
tests/test_dynamodb2/test_dynamodb_table_with_range_key.py
View file

@ -1307,16 +1307,16 @@ def test_update_item_add_with_expression():
ExpressionAttributeValues={":v": {"item4"}},
)
current_item["str_set"] = current_item["str_set"].union({"item4"})
dict(table.get_item(Key=item_key)["Item"]).should.equal(current_item)
assert dict(table.get_item(Key=item_key)["Item"]) == current_item
# Update item to add a string value to a non-existing set
# Should throw: 'The provided key element does not match the schema'
assert_failure_due_to_key_not_in_schema(
table.update_item,
table.update_item(
Key=item_key,
UpdateExpression="ADD non_existing_str_set :v",
ExpressionAttributeValues={":v": {"item4"}},
)
current_item["non_existing_str_set"] = {"item4"}
assert dict(table.get_item(Key=item_key)["Item"]) == current_item
# Update item to add a num value to a num set
table.update_item(
@ -1325,7 +1325,7 @@ def test_update_item_add_with_expression():
ExpressionAttributeValues={":v": {6}},
)
current_item["num_set"] = current_item["num_set"].union({6})
dict(table.get_item(Key=item_key)["Item"]).should.equal(current_item)
assert dict(table.get_item(Key=item_key)["Item"]) == current_item
# Update item to add a value to a number value
table.update_item(
@ -1334,7 +1334,7 @@ def test_update_item_add_with_expression():
ExpressionAttributeValues={":v": 20},
)
current_item["num_val"] = current_item["num_val"] + 20
dict(table.get_item(Key=item_key)["Item"]).should.equal(current_item)
assert dict(table.get_item(Key=item_key)["Item"]) == current_item
# Attempt to add a number value to a string set, should raise Client Error
table.update_item.when.called_with(
@ -1342,7 +1342,7 @@ def test_update_item_add_with_expression():
UpdateExpression="ADD str_set :v",
ExpressionAttributeValues={":v": 20},
).should.have.raised(ClientError)
dict(table.get_item(Key=item_key)["Item"]).should.equal(current_item)
assert dict(table.get_item(Key=item_key)["Item"]) == current_item
# Attempt to add a number set to the string set, should raise a ClientError
table.update_item.when.called_with(
@ -1350,7 +1350,7 @@ def test_update_item_add_with_expression():
UpdateExpression="ADD str_set :v",
ExpressionAttributeValues={":v": {20}},
).should.have.raised(ClientError)
dict(table.get_item(Key=item_key)["Item"]).should.equal(current_item)
assert dict(table.get_item(Key=item_key)["Item"]) == current_item
# Attempt to update with a bad expression
table.update_item.when.called_with(
@ -1388,17 +1388,18 @@ def test_update_item_add_with_nested_sets():
current_item["nested"]["str_set"] = current_item["nested"]["str_set"].union(
{"item4"}
)
dict(table.get_item(Key=item_key)["Item"]).should.equal(current_item)
assert dict(table.get_item(Key=item_key)["Item"]) == current_item
# Update item to add a string value to a non-existing set
# Should raise
assert_failure_due_to_key_not_in_schema(
table.update_item,
table.update_item(
Key=item_key,
UpdateExpression="ADD #ns.#ne :v",
ExpressionAttributeNames={"#ns": "nested", "#ne": "non_existing_str_set"},
ExpressionAttributeValues={":v": {"new_item"}},
)
current_item["nested"]["non_existing_str_set"] = {"new_item"}
assert dict(table.get_item(Key=item_key)["Item"]) == current_item
@mock_dynamodb2

105
tests/test_ec2/test_instances.py
View file

@ -1126,6 +1126,111 @@ def test_run_instance_with_keypair():
instance.key_name.should.equal("keypair_name")
@mock_ec2
def test_run_instance_with_block_device_mappings():
ec2_client = boto3.client("ec2", region_name="us-east-1")
kwargs = {
"MinCount": 1,
"MaxCount": 1,
"ImageId": "ami-d3adb33f",
"KeyName": "the_key",
"InstanceType": "t1.micro",
"BlockDeviceMappings": [{"DeviceName": "/dev/sda2", "Ebs": {"VolumeSize": 50}}],
}
ec2_client.run_instances(**kwargs)
instances = ec2_client.describe_instances()
volume = instances["Reservations"][0]["Instances"][0]["BlockDeviceMappings"][0][
"Ebs"
]
volumes = ec2_client.describe_volumes(VolumeIds=[volume["VolumeId"]])
volumes["Volumes"][0]["Size"].should.equal(50)
@mock_ec2
def test_run_instance_with_block_device_mappings_missing_ebs():
ec2_client = boto3.client("ec2", region_name="us-east-1")
kwargs = {
"MinCount": 1,
"MaxCount": 1,
"ImageId": "ami-d3adb33f",
"KeyName": "the_key",
"InstanceType": "t1.micro",
"BlockDeviceMappings": [{"DeviceName": "/dev/sda2"}],
}
with assert_raises(ClientError) as ex:
ec2_client.run_instances(**kwargs)
ex.exception.response["Error"]["Code"].should.equal("MissingParameter")
ex.exception.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
ex.exception.response["Error"]["Message"].should.equal(
"The request must contain the parameter ebs"
)
@mock_ec2
def test_run_instance_with_block_device_mappings_missing_size():
ec2_client = boto3.client("ec2", region_name="us-east-1")
kwargs = {
"MinCount": 1,
"MaxCount": 1,
"ImageId": "ami-d3adb33f",
"KeyName": "the_key",
"InstanceType": "t1.micro",
"BlockDeviceMappings": [
{"DeviceName": "/dev/sda2", "Ebs": {"VolumeType": "standard"}}
],
}
with assert_raises(ClientError) as ex:
ec2_client.run_instances(**kwargs)
ex.exception.response["Error"]["Code"].should.equal("MissingParameter")
ex.exception.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
ex.exception.response["Error"]["Message"].should.equal(
"The request must contain the parameter size or snapshotId"
)
@mock_ec2
def test_run_instance_with_block_device_mappings_from_snapshot():
ec2_client = boto3.client("ec2", region_name="us-east-1")
ec2_resource = boto3.resource("ec2", region_name="us-east-1")
volume_details = {
"AvailabilityZone": "1a",
"Size": 30,
}
volume = ec2_resource.create_volume(**volume_details)
snapshot = volume.create_snapshot()
kwargs = {
"MinCount": 1,
"MaxCount": 1,
"ImageId": "ami-d3adb33f",
"KeyName": "the_key",
"InstanceType": "t1.micro",
"BlockDeviceMappings": [
{"DeviceName": "/dev/sda2", "Ebs": {"SnapshotId": snapshot.snapshot_id}}
],
}
ec2_client.run_instances(**kwargs)
instances = ec2_client.describe_instances()
volume = instances["Reservations"][0]["Instances"][0]["BlockDeviceMappings"][0][
"Ebs"
]
volumes = ec2_client.describe_volumes(VolumeIds=[volume["VolumeId"]])
volumes["Volumes"][0]["Size"].should.equal(30)
volumes["Volumes"][0]["SnapshotId"].should.equal(snapshot.snapshot_id)
@mock_ec2_deprecated
def test_describe_instance_status_no_instances():
conn = boto.connect_ec2("the_key", "the_secret")

256
tests/test_iot/test_iot.py
View file

@ -9,6 +9,38 @@ from botocore.exceptions import ClientError
from nose.tools import assert_raises
def generate_thing_group_tree(iot_client, tree_dict, _parent=None):
"""
Generates a thing group tree given the input tree structure.
:param iot_client: the iot client for boto3
:param tree_dict: dictionary with the key being the group_name, and the value being a sub tree.
tree_dict = {
"group_name_1a":{
"group_name_2a":{
"group_name_3a":{} or None
},
},
"group_name_1b":{}
}
:return: a dictionary of created groups, keyed by group name
"""
if tree_dict is None:
tree_dict = {}
created_dict = {}
for group_name in tree_dict.keys():
params = {"thingGroupName": group_name}
if _parent:
params["parentGroupName"] = _parent
created_group = iot_client.create_thing_group(**params)
created_dict[group_name] = created_group
subtree_dict = generate_thing_group_tree(
iot_client=iot_client, tree_dict=tree_dict[group_name], _parent=group_name
)
created_dict.update(created_dict)
created_dict.update(subtree_dict)
return created_dict
@mock_iot
def test_attach_policy():
client = boto3.client("iot", region_name="ap-northeast-1")
@ -756,25 +788,143 @@ def test_delete_principal_thing():
client.delete_certificate(certificateId=cert_id)
class TestListThingGroup:
group_name_1a = "my-group-name-1a"
group_name_1b = "my-group-name-1b"
group_name_2a = "my-group-name-2a"
group_name_2b = "my-group-name-2b"
group_name_3a = "my-group-name-3a"
group_name_3b = "my-group-name-3b"
group_name_3c = "my-group-name-3c"
group_name_3d = "my-group-name-3d"
tree_dict = {
group_name_1a: {
group_name_2a: {group_name_3a: {}, group_name_3b: {}},
group_name_2b: {group_name_3c: {}, group_name_3d: {}},
},
group_name_1b: {},
}
@mock_iot
def test_should_list_all_groups(self):
# setup
client = boto3.client("iot", region_name="ap-northeast-1")
group_catalog = generate_thing_group_tree(client, self.tree_dict)
# test
resp = client.list_thing_groups()
resp.should.have.key("thingGroups")
resp["thingGroups"].should.have.length_of(8)
@mock_iot
def test_should_list_all_groups_non_recursively(self):
# setup
client = boto3.client("iot", region_name="ap-northeast-1")
group_catalog = generate_thing_group_tree(client, self.tree_dict)
# test
resp = client.list_thing_groups(recursive=False)
resp.should.have.key("thingGroups")
resp["thingGroups"].should.have.length_of(2)
@mock_iot
def test_should_list_all_groups_filtered_by_parent(self):
# setup
client = boto3.client("iot", region_name="ap-northeast-1")
group_catalog = generate_thing_group_tree(client, self.tree_dict)
# test
resp = client.list_thing_groups(parentGroup=self.group_name_1a)
resp.should.have.key("thingGroups")
resp["thingGroups"].should.have.length_of(6)
resp = client.list_thing_groups(parentGroup=self.group_name_2a)
resp.should.have.key("thingGroups")
resp["thingGroups"].should.have.length_of(2)
resp = client.list_thing_groups(parentGroup=self.group_name_1b)
resp.should.have.key("thingGroups")
resp["thingGroups"].should.have.length_of(0)
with assert_raises(ClientError) as e:
client.list_thing_groups(parentGroup="inexistant-group-name")
e.exception.response["Error"]["Code"].should.equal(
"ResourceNotFoundException"
)
@mock_iot
def test_should_list_all_groups_filtered_by_parent_non_recursively(self):
# setup
client = boto3.client("iot", region_name="ap-northeast-1")
group_catalog = generate_thing_group_tree(client, self.tree_dict)
# test
resp = client.list_thing_groups(parentGroup=self.group_name_1a, recursive=False)
resp.should.have.key("thingGroups")
resp["thingGroups"].should.have.length_of(2)
resp = client.list_thing_groups(parentGroup=self.group_name_2a, recursive=False)
resp.should.have.key("thingGroups")
resp["thingGroups"].should.have.length_of(2)
@mock_iot
def test_should_list_all_groups_filtered_by_name_prefix(self):
# setup
client = boto3.client("iot", region_name="ap-northeast-1")
group_catalog = generate_thing_group_tree(client, self.tree_dict)
# test
resp = client.list_thing_groups(namePrefixFilter="my-group-name-1")
resp.should.have.key("thingGroups")
resp["thingGroups"].should.have.length_of(2)
resp = client.list_thing_groups(namePrefixFilter="my-group-name-3")
resp.should.have.key("thingGroups")
resp["thingGroups"].should.have.length_of(4)
resp = client.list_thing_groups(namePrefixFilter="prefix-which-doesn-not-match")
resp.should.have.key("thingGroups")
resp["thingGroups"].should.have.length_of(0)
@mock_iot
def test_should_list_all_groups_filtered_by_name_prefix_non_recursively(self):
# setup
client = boto3.client("iot", region_name="ap-northeast-1")
group_catalog = generate_thing_group_tree(client, self.tree_dict)
# test
resp = client.list_thing_groups(
namePrefixFilter="my-group-name-1", recursive=False
)
resp.should.have.key("thingGroups")
resp["thingGroups"].should.have.length_of(2)
resp = client.list_thing_groups(
namePrefixFilter="my-group-name-3", recursive=False
)
resp.should.have.key("thingGroups")
resp["thingGroups"].should.have.length_of(0)
@mock_iot
def test_should_list_all_groups_filtered_by_name_prefix_and_parent(self):
# setup
client = boto3.client("iot", region_name="ap-northeast-1")
group_catalog = generate_thing_group_tree(client, self.tree_dict)
# test
resp = client.list_thing_groups(
namePrefixFilter="my-group-name-2", parentGroup=self.group_name_1a
)
resp.should.have.key("thingGroups")
resp["thingGroups"].should.have.length_of(2)
resp = client.list_thing_groups(
namePrefixFilter="my-group-name-3", parentGroup=self.group_name_1a
)
resp.should.have.key("thingGroups")
resp["thingGroups"].should.have.length_of(4)
resp = client.list_thing_groups(
namePrefixFilter="prefix-which-doesn-not-match",
parentGroup=self.group_name_1a,
)
resp.should.have.key("thingGroups")
resp["thingGroups"].should.have.length_of(0)
@mock_iot
def test_delete_thing_group():
client = boto3.client("iot", region_name="ap-northeast-1")
group_name_1a = "my-group-name-1a"
group_name_2a = "my-group-name-2a"
# --1a
# |--2a
# create thing groups tree
# 1
thing_group1a = client.create_thing_group(thingGroupName=group_name_1a)
thing_group1a.should.have.key("thingGroupName").which.should.equal(group_name_1a)
thing_group1a.should.have.key("thingGroupArn")
# 2
thing_group2a = client.create_thing_group(
thingGroupName=group_name_2a, parentGroupName=group_name_1a
)
thing_group2a.should.have.key("thingGroupName").which.should.equal(group_name_2a)
thing_group2a.should.have.key("thingGroupArn")
tree_dict = {
group_name_1a: {group_name_2a: {},},
}
group_catalog = generate_thing_group_tree(client, tree_dict)
# delete group with child
try:
@ -809,56 +959,14 @@ def test_describe_thing_group_metadata_hierarchy():
group_name_3c = "my-group-name-3c"
group_name_3d = "my-group-name-3d"
# --1a
# |--2a
# | |--3a
# | |--3b
# |
# |--2b
# |--3c
# |--3d
# --1b
# create thing groups tree
# 1
thing_group1a = client.create_thing_group(thingGroupName=group_name_1a)
thing_group1a.should.have.key("thingGroupName").which.should.equal(group_name_1a)
thing_group1a.should.have.key("thingGroupArn")
thing_group1b = client.create_thing_group(thingGroupName=group_name_1b)
thing_group1b.should.have.key("thingGroupName").which.should.equal(group_name_1b)
thing_group1b.should.have.key("thingGroupArn")
# 2
thing_group2a = client.create_thing_group(
thingGroupName=group_name_2a, parentGroupName=group_name_1a
)
thing_group2a.should.have.key("thingGroupName").which.should.equal(group_name_2a)
thing_group2a.should.have.key("thingGroupArn")
thing_group2b = client.create_thing_group(
thingGroupName=group_name_2b, parentGroupName=group_name_1a
)
thing_group2b.should.have.key("thingGroupName").which.should.equal(group_name_2b)
thing_group2b.should.have.key("thingGroupArn")
# 3
thing_group3a = client.create_thing_group(
thingGroupName=group_name_3a, parentGroupName=group_name_2a
)
thing_group3a.should.have.key("thingGroupName").which.should.equal(group_name_3a)
thing_group3a.should.have.key("thingGroupArn")
thing_group3b = client.create_thing_group(
thingGroupName=group_name_3b, parentGroupName=group_name_2a
)
thing_group3b.should.have.key("thingGroupName").which.should.equal(group_name_3b)
thing_group3b.should.have.key("thingGroupArn")
thing_group3c = client.create_thing_group(
thingGroupName=group_name_3c, parentGroupName=group_name_2b
)
thing_group3c.should.have.key("thingGroupName").which.should.equal(group_name_3c)
thing_group3c.should.have.key("thingGroupArn")
thing_group3d = client.create_thing_group(
thingGroupName=group_name_3d, parentGroupName=group_name_2b
)
thing_group3d.should.have.key("thingGroupName").which.should.equal(group_name_3d)
thing_group3d.should.have.key("thingGroupArn")
tree_dict = {
group_name_1a: {
group_name_2a: {group_name_3a: {}, group_name_3b: {}},
group_name_2b: {group_name_3c: {}, group_name_3d: {}},
},
group_name_1b: {},
}
group_catalog = generate_thing_group_tree(client, tree_dict)
# describe groups
# groups level 1
@ -910,7 +1018,7 @@ def test_describe_thing_group_metadata_hierarchy():
].should.match(group_name_1a)
thing_group_description2a["thingGroupMetadata"]["rootToParentThingGroups"][0][
"groupArn"
].should.match(thing_group1a["thingGroupArn"])
].should.match(group_catalog[group_name_1a]["thingGroupArn"])
thing_group_description2a.should.have.key("version")
# 2b
thing_group_description2b = client.describe_thing_group(
@ -936,7 +1044,7 @@ def test_describe_thing_group_metadata_hierarchy():
].should.match(group_name_1a)
thing_group_description2b["thingGroupMetadata"]["rootToParentThingGroups"][0][
"groupArn"
].should.match(thing_group1a["thingGroupArn"])
].should.match(group_catalog[group_name_1a]["thingGroupArn"])
thing_group_description2b.should.have.key("version")
# groups level 3
# 3a
@ -963,13 +1071,13 @@ def test_describe_thing_group_metadata_hierarchy():
].should.match(group_name_1a)
thing_group_description3a["thingGroupMetadata"]["rootToParentThingGroups"][0][
"groupArn"
].should.match(thing_group1a["thingGroupArn"])
].should.match(group_catalog[group_name_1a]["thingGroupArn"])
thing_group_description3a["thingGroupMetadata"]["rootToParentThingGroups"][1][
"groupName"
].should.match(group_name_2a)
thing_group_description3a["thingGroupMetadata"]["rootToParentThingGroups"][1][
"groupArn"
].should.match(thing_group2a["thingGroupArn"])
].should.match(group_catalog[group_name_2a]["thingGroupArn"])
thing_group_description3a.should.have.key("version")
# 3b
thing_group_description3b = client.describe_thing_group(
@ -995,13 +1103,13 @@ def test_describe_thing_group_metadata_hierarchy():
].should.match(group_name_1a)
thing_group_description3b["thingGroupMetadata"]["rootToParentThingGroups"][0][
"groupArn"
].should.match(thing_group1a["thingGroupArn"])
].should.match(group_catalog[group_name_1a]["thingGroupArn"])
thing_group_description3b["thingGroupMetadata"]["rootToParentThingGroups"][1][
"groupName"
].should.match(group_name_2a)
thing_group_description3b["thingGroupMetadata"]["rootToParentThingGroups"][1][
"groupArn"
].should.match(thing_group2a["thingGroupArn"])
].should.match(group_catalog[group_name_2a]["thingGroupArn"])
thing_group_description3b.should.have.key("version")
# 3c
thing_group_description3c = client.describe_thing_group(
@ -1027,13 +1135,13 @@ def test_describe_thing_group_metadata_hierarchy():
].should.match(group_name_1a)
thing_group_description3c["thingGroupMetadata"]["rootToParentThingGroups"][0][
"groupArn"
].should.match(thing_group1a["thingGroupArn"])
].should.match(group_catalog[group_name_1a]["thingGroupArn"])
thing_group_description3c["thingGroupMetadata"]["rootToParentThingGroups"][1][
"groupName"
].should.match(group_name_2b)
thing_group_description3c["thingGroupMetadata"]["rootToParentThingGroups"][1][
"groupArn"
].should.match(thing_group2b["thingGroupArn"])
].should.match(group_catalog[group_name_2b]["thingGroupArn"])
thing_group_description3c.should.have.key("version")
# 3d
thing_group_description3d = client.describe_thing_group(
@ -1059,13 +1167,13 @@ def test_describe_thing_group_metadata_hierarchy():
].should.match(group_name_1a)
thing_group_description3d["thingGroupMetadata"]["rootToParentThingGroups"][0][
"groupArn"
].should.match(thing_group1a["thingGroupArn"])
].should.match(group_catalog[group_name_1a]["thingGroupArn"])
thing_group_description3d["thingGroupMetadata"]["rootToParentThingGroups"][1][
"groupName"
].should.match(group_name_2b)
thing_group_description3d["thingGroupMetadata"]["rootToParentThingGroups"][1][
"groupArn"
].should.match(thing_group2b["thingGroupArn"])
].should.match(group_catalog[group_name_2b]["thingGroupArn"])
thing_group_description3d.should.have.key("version")

414
tests/test_logs/test_logs.py
View file

@ -1,10 +1,17 @@
import base64
import json
import time
import zlib
from io import BytesIO
from zipfile import ZipFile, ZIP_DEFLATED
import boto3
import os
import sure # noqa
import six
from botocore.exceptions import ClientError
from moto import mock_logs, settings
from moto import mock_logs, settings, mock_lambda, mock_iam
from nose.tools import assert_raises
from nose import SkipTest
@ -425,3 +432,408 @@ def test_untag_log_group():
assert response["tags"] == remaining_tags
response = conn.delete_log_group(logGroupName=log_group_name)
@mock_logs
def test_describe_subscription_filters():
# given
client = boto3.client("logs", "us-east-1")
log_group_name = "/test"
client.create_log_group(logGroupName=log_group_name)
# when
response = client.describe_subscription_filters(logGroupName=log_group_name)
# then
response["subscriptionFilters"].should.have.length_of(0)
@mock_logs
def test_describe_subscription_filters_errors():
# given
client = boto3.client("logs", "us-east-1")
# when
with assert_raises(ClientError) as e:
client.describe_subscription_filters(logGroupName="not-existing-log-group",)
# then
ex = e.exception
ex.operation_name.should.equal("DescribeSubscriptionFilters")
ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
ex.response["Error"]["Code"].should.contain("ResourceNotFoundException")
ex.response["Error"]["Message"].should.equal(
"The specified log group does not exist"
)
@mock_lambda
@mock_logs
def test_put_subscription_filter_update():
# given
region_name = "us-east-1"
client_lambda = boto3.client("lambda", region_name)
client_logs = boto3.client("logs", region_name)
log_group_name = "/test"
log_stream_name = "stream"
client_logs.create_log_group(logGroupName=log_group_name)
client_logs.create_log_stream(
logGroupName=log_group_name, logStreamName=log_stream_name
)
function_arn = client_lambda.create_function(
FunctionName="test",
Runtime="python3.8",
Role=_get_role_name(region_name),
Handler="lambda_function.lambda_handler",
Code={"ZipFile": _get_test_zip_file()},
Description="test lambda function",
Timeout=3,
MemorySize=128,
Publish=True,
)["FunctionArn"]
# when
client_logs.put_subscription_filter(
logGroupName=log_group_name,
filterName="test",
filterPattern="",
destinationArn=function_arn,
)
# then
response = client_logs.describe_subscription_filters(logGroupName=log_group_name)
response["subscriptionFilters"].should.have.length_of(1)
filter = response["subscriptionFilters"][0]
creation_time = filter["creationTime"]
creation_time.should.be.a(int)
filter["destinationArn"] = "arn:aws:lambda:us-east-1:123456789012:function:test"
filter["distribution"] = "ByLogStream"
filter["logGroupName"] = "/test"
filter["filterName"] = "test"
filter["filterPattern"] = ""
# when
# to update an existing subscription filter the 'filerName' must be identical
client_logs.put_subscription_filter(
logGroupName=log_group_name,
filterName="test",
filterPattern="[]",
destinationArn=function_arn,
)
# then
response = client_logs.describe_subscription_filters(logGroupName=log_group_name)
response["subscriptionFilters"].should.have.length_of(1)
filter = response["subscriptionFilters"][0]
filter["creationTime"].should.equal(creation_time)
filter["destinationArn"] = "arn:aws:lambda:us-east-1:123456789012:function:test"
filter["distribution"] = "ByLogStream"
filter["logGroupName"] = "/test"
filter["filterName"] = "test"
filter["filterPattern"] = "[]"
# when
# only one subscription filter can be associated with a log group
with assert_raises(ClientError) as e:
client_logs.put_subscription_filter(
logGroupName=log_group_name,
filterName="test-2",
filterPattern="",
destinationArn=function_arn,
)
# then
ex = e.exception
ex.operation_name.should.equal("PutSubscriptionFilter")
ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
ex.response["Error"]["Code"].should.contain("LimitExceededException")
ex.response["Error"]["Message"].should.equal("Resource limit exceeded.")
@mock_lambda
@mock_logs
def test_put_subscription_filter_with_lambda():
# given
region_name = "us-east-1"
client_lambda = boto3.client("lambda", region_name)
client_logs = boto3.client("logs", region_name)
log_group_name = "/test"
log_stream_name = "stream"
client_logs.create_log_group(logGroupName=log_group_name)
client_logs.create_log_stream(
logGroupName=log_group_name, logStreamName=log_stream_name
)
function_arn = client_lambda.create_function(
FunctionName="test",
Runtime="python3.8",
Role=_get_role_name(region_name),
Handler="lambda_function.lambda_handler",
Code={"ZipFile": _get_test_zip_file()},
Description="test lambda function",
Timeout=3,
MemorySize=128,
Publish=True,
)["FunctionArn"]
# when
client_logs.put_subscription_filter(
logGroupName=log_group_name,
filterName="test",
filterPattern="",
destinationArn=function_arn,
)
# then
response = client_logs.describe_subscription_filters(logGroupName=log_group_name)
response["subscriptionFilters"].should.have.length_of(1)
filter = response["subscriptionFilters"][0]
filter["creationTime"].should.be.a(int)
filter["destinationArn"] = "arn:aws:lambda:us-east-1:123456789012:function:test"
filter["distribution"] = "ByLogStream"
filter["logGroupName"] = "/test"
filter["filterName"] = "test"
filter["filterPattern"] = ""
# when
client_logs.put_log_events(
logGroupName=log_group_name,
logStreamName=log_stream_name,
logEvents=[
{"timestamp": 0, "message": "test"},
{"timestamp": 0, "message": "test 2"},
],
)
# then
msg_showed_up, received_message = _wait_for_log_msg(
client_logs, "/aws/lambda/test", "awslogs"
)
assert msg_showed_up, "CloudWatch log event was not found. All logs: {}".format(
received_message
)
data = json.loads(received_message)["awslogs"]["data"]
response = json.loads(
zlib.decompress(base64.b64decode(data), 16 + zlib.MAX_WBITS).decode("utf-8")
)
response["messageType"].should.equal("DATA_MESSAGE")
response["owner"].should.equal("123456789012")
response["logGroup"].should.equal("/test")
response["logStream"].should.equal("stream")
response["subscriptionFilters"].should.equal(["test"])
log_events = sorted(response["logEvents"], key=lambda log_event: log_event["id"])
log_events.should.have.length_of(2)
log_events[0]["id"].should.be.a(int)
log_events[0]["message"].should.equal("test")
log_events[0]["timestamp"].should.equal(0)
log_events[1]["id"].should.be.a(int)
log_events[1]["message"].should.equal("test 2")
log_events[1]["timestamp"].should.equal(0)
@mock_logs
def test_put_subscription_filter_errors():
# given
client = boto3.client("logs", "us-east-1")
log_group_name = "/test"
client.create_log_group(logGroupName=log_group_name)
# when
with assert_raises(ClientError) as e:
client.put_subscription_filter(
logGroupName="not-existing-log-group",
filterName="test",
filterPattern="",
destinationArn="arn:aws:lambda:us-east-1:123456789012:function:test",
)
# then
ex = e.exception
ex.operation_name.should.equal("PutSubscriptionFilter")
ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
ex.response["Error"]["Code"].should.contain("ResourceNotFoundException")
ex.response["Error"]["Message"].should.equal(
"The specified log group does not exist"
)
# when
with assert_raises(ClientError) as e:
client.put_subscription_filter(
logGroupName="/test",
filterName="test",
filterPattern="",
destinationArn="arn:aws:lambda:us-east-1:123456789012:function:not-existing",
)
# then
ex = e.exception
ex.operation_name.should.equal("PutSubscriptionFilter")
ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
ex.response["Error"]["Code"].should.contain("InvalidParameterException")
ex.response["Error"]["Message"].should.equal(
"Could not execute the lambda function. "
"Make sure you have given CloudWatch Logs permission to execute your function."
)
# when
with assert_raises(ClientError) as e:
client.put_subscription_filter(
logGroupName="/test",
filterName="test",
filterPattern="",
destinationArn="arn:aws:lambda:us-east-1:123456789012:function:not-existing",
)
# then
ex = e.exception
ex.operation_name.should.equal("PutSubscriptionFilter")
ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
ex.response["Error"]["Code"].should.contain("InvalidParameterException")
ex.response["Error"]["Message"].should.equal(
"Could not execute the lambda function. "
"Make sure you have given CloudWatch Logs permission to execute your function."
)
@mock_lambda
@mock_logs
def test_delete_subscription_filter_errors():
# given
region_name = "us-east-1"
client_lambda = boto3.client("lambda", region_name)
client_logs = boto3.client("logs", region_name)
log_group_name = "/test"
client_logs.create_log_group(logGroupName=log_group_name)
function_arn = client_lambda.create_function(
FunctionName="test",
Runtime="python3.8",
Role=_get_role_name(region_name),
Handler="lambda_function.lambda_handler",
Code={"ZipFile": _get_test_zip_file()},
Description="test lambda function",
Timeout=3,
MemorySize=128,
Publish=True,
)["FunctionArn"]
client_logs.put_subscription_filter(
logGroupName=log_group_name,
filterName="test",
filterPattern="",
destinationArn=function_arn,
)
# when
client_logs.delete_subscription_filter(
logGroupName="/test", filterName="test",
)
# then
response = client_logs.describe_subscription_filters(logGroupName=log_group_name)
response["subscriptionFilters"].should.have.length_of(0)
@mock_lambda
@mock_logs
def test_delete_subscription_filter_errors():
# given
region_name = "us-east-1"
client_lambda = boto3.client("lambda", region_name)
client_logs = boto3.client("logs", region_name)
log_group_name = "/test"
client_logs.create_log_group(logGroupName=log_group_name)
function_arn = client_lambda.create_function(
FunctionName="test",
Runtime="python3.8",
Role=_get_role_name(region_name),
Handler="lambda_function.lambda_handler",
Code={"ZipFile": _get_test_zip_file()},
Description="test lambda function",
Timeout=3,
MemorySize=128,
Publish=True,
)["FunctionArn"]
client_logs.put_subscription_filter(
logGroupName=log_group_name,
filterName="test",
filterPattern="",
destinationArn=function_arn,
)
# when
with assert_raises(ClientError) as e:
client_logs.delete_subscription_filter(
logGroupName="not-existing-log-group", filterName="test",
)
# then
ex = e.exception
ex.operation_name.should.equal("DeleteSubscriptionFilter")
ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
ex.response["Error"]["Code"].should.contain("ResourceNotFoundException")
ex.response["Error"]["Message"].should.equal(
"The specified log group does not exist"
)
# when
with assert_raises(ClientError) as e:
client_logs.delete_subscription_filter(
logGroupName="/test", filterName="wrong-filter-name",
)
# then
ex = e.exception
ex.operation_name.should.equal("DeleteSubscriptionFilter")
ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
ex.response["Error"]["Code"].should.contain("ResourceNotFoundException")
ex.response["Error"]["Message"].should.equal(
"The specified subscription filter does not exist."
)
def _get_role_name(region_name):
with mock_iam():
iam = boto3.client("iam", region_name=region_name)
try:
return iam.get_role(RoleName="test-role")["Role"]["Arn"]
except ClientError:
return iam.create_role(
RoleName="test-role", AssumeRolePolicyDocument="test policy", Path="/",
)["Role"]["Arn"]
def _get_test_zip_file():
func_str = """
def lambda_handler(event, context):
return event
"""
zip_output = BytesIO()
zip_file = ZipFile(zip_output, "w", ZIP_DEFLATED)
zip_file.writestr("lambda_function.py", func_str)
zip_file.close()
zip_output.seek(0)
return zip_output.read()
def _wait_for_log_msg(client, log_group_name, expected_msg_part):
received_messages = []
start = time.time()
while (time.time() - start) < 10:
result = client.describe_log_streams(logGroupName=log_group_name)
log_streams = result.get("logStreams")
if not log_streams:
time.sleep(1)
continue
for log_stream in log_streams:
result = client.get_log_events(
logGroupName=log_group_name, logStreamName=log_stream["logStreamName"],
)
received_messages.extend(
[event["message"] for event in result.get("events")]
)
for message in received_messages:
if expected_msg_part in message:
return True, message
time.sleep(1)
return False, received_messages

142
tests/test_managedblockchain/test_managedblockchain_networks.py Normal file
View file

@ -0,0 +1,142 @@
from __future__ import unicode_literals
import boto3
import sure # noqa
from moto.managedblockchain.exceptions import BadRequestException
from moto import mock_managedblockchain
default_frameworkconfiguration = {"Fabric": {"Edition": "STARTER"}}
default_votingpolicy = {
"ApprovalThresholdPolicy": {
"ThresholdPercentage": 50,
"ProposalDurationInHours": 24,
"ThresholdComparator": "GREATER_THAN_OR_EQUAL_TO",
}
}
default_memberconfiguration = {
"Name": "testmember1",
"Description": "Test Member 1",
"FrameworkConfiguration": {
"Fabric": {"AdminUsername": "admin", "AdminPassword": "Admin12345"}
},
"LogPublishingConfiguration": {
"Fabric": {"CaLogs": {"Cloudwatch": {"Enabled": False}}}
},
}
@mock_managedblockchain
def test_create_network():
conn = boto3.client("managedblockchain", region_name="us-east-1")
response = conn.create_network(
Name="testnetwork1",
Framework="HYPERLEDGER_FABRIC",
FrameworkVersion="1.2",
FrameworkConfiguration=default_frameworkconfiguration,
VotingPolicy=default_votingpolicy,
MemberConfiguration=default_memberconfiguration,
)
response["NetworkId"].should.match("n-[A-Z0-9]{26}")
response["MemberId"].should.match("m-[A-Z0-9]{26}")
# Find in full list
response = conn.list_networks()
mbcnetworks = response["Networks"]
mbcnetworks.should.have.length_of(1)
mbcnetworks[0]["Name"].should.equal("testnetwork1")
# Get network details
network_id = mbcnetworks[0]["Id"]
response = conn.get_network(NetworkId=network_id)
response["Network"]["Name"].should.equal("testnetwork1")
@mock_managedblockchain
def test_create_network_withopts():
conn = boto3.client("managedblockchain", region_name="us-east-1")
response = conn.create_network(
Name="testnetwork1",
Description="Test Network 1",
Framework="HYPERLEDGER_FABRIC",
FrameworkVersion="1.2",
FrameworkConfiguration=default_frameworkconfiguration,
VotingPolicy=default_votingpolicy,
MemberConfiguration=default_memberconfiguration,
)
response["NetworkId"].should.match("n-[A-Z0-9]{26}")
response["MemberId"].should.match("m-[A-Z0-9]{26}")
# Find in full list
response = conn.list_networks()
mbcnetworks = response["Networks"]
mbcnetworks.should.have.length_of(1)
mbcnetworks[0]["Description"].should.equal("Test Network 1")
# Get network details
network_id = mbcnetworks[0]["Id"]
response = conn.get_network(NetworkId=network_id)
response["Network"]["Description"].should.equal("Test Network 1")
@mock_managedblockchain
def test_create_network_noframework():
conn = boto3.client("managedblockchain", region_name="us-east-1")
response = conn.create_network.when.called_with(
Name="testnetwork1",
Description="Test Network 1",
Framework="HYPERLEDGER_VINYL",
FrameworkVersion="1.2",
FrameworkConfiguration=default_frameworkconfiguration,
VotingPolicy=default_votingpolicy,
MemberConfiguration=default_memberconfiguration,
).should.throw(Exception, "Invalid request body")
@mock_managedblockchain
def test_create_network_badframeworkver():
conn = boto3.client("managedblockchain", region_name="us-east-1")
response = conn.create_network.when.called_with(
Name="testnetwork1",
Description="Test Network 1",
Framework="HYPERLEDGER_FABRIC",
FrameworkVersion="1.X",
FrameworkConfiguration=default_frameworkconfiguration,
VotingPolicy=default_votingpolicy,
MemberConfiguration=default_memberconfiguration,
).should.throw(
Exception, "Invalid version 1.X requested for framework HYPERLEDGER_FABRIC"
)
@mock_managedblockchain
def test_create_network_badedition():
conn = boto3.client("managedblockchain", region_name="us-east-1")
frameworkconfiguration = {"Fabric": {"Edition": "SUPER"}}
response = conn.create_network.when.called_with(
Name="testnetwork1",
Description="Test Network 1",
Framework="HYPERLEDGER_FABRIC",
FrameworkVersion="1.2",
FrameworkConfiguration=frameworkconfiguration,
VotingPolicy=default_votingpolicy,
MemberConfiguration=default_memberconfiguration,
).should.throw(Exception, "Invalid request body")
@mock_managedblockchain
def test_get_network_badnetwork():
conn = boto3.client("managedblockchain", region_name="us-east-1")
response = conn.get_network.when.called_with(
NetworkId="n-BADNETWORK",
).should.throw(Exception, "Network n-BADNETWORK not found")

6
tests/test_opsworks/test_instances.py
View file

@ -195,6 +195,10 @@ def test_ec2_integration():
reservations = ec2.describe_instances()["Reservations"]
assert reservations.should.be.empty
# Before starting the instance, its status should be "stopped"
opsworks_instance = opsworks.describe_instances(StackId=stack_id)["Instances"][0]
opsworks_instance["Status"].should.equal("stopped")
# After starting the instance, it should be discoverable via ec2
opsworks.start_instance(InstanceId=instance_id)
reservations = ec2.describe_instances()["Reservations"]
@ -204,3 +208,5 @@ def test_ec2_integration():
instance["InstanceId"].should.equal(opsworks_instance["Ec2InstanceId"])
instance["PrivateIpAddress"].should.equal(opsworks_instance["PrivateIp"])
# After starting the instance, its status should be "online"
opsworks_instance["Status"].should.equal("online")

22
tests/test_redshift/test_redshift.py
View file

@ -915,6 +915,11 @@ def test_create_cluster_from_snapshot():
ClusterIdentifier=original_cluster_identifier,
)
client.restore_from_cluster_snapshot.when.called_with(
ClusterIdentifier=original_cluster_identifier,
SnapshotIdentifier=original_snapshot_identifier,
).should.throw(ClientError, "ClusterAlreadyExists")
response = client.restore_from_cluster_snapshot(
ClusterIdentifier=new_cluster_identifier,
SnapshotIdentifier=original_snapshot_identifier,
@ -1333,3 +1338,20 @@ def test_modify_snapshot_copy_retention_period():
response = client.describe_clusters(ClusterIdentifier="test")
cluster_snapshot_copy_status = response["Clusters"][0]["ClusterSnapshotCopyStatus"]
cluster_snapshot_copy_status["RetentionPeriod"].should.equal(5)
@mock_redshift
def test_create_duplicate_cluster_fails():
kwargs = {
"ClusterIdentifier": "test",
"ClusterType": "single-node",
"DBName": "test",
"MasterUsername": "user",
"MasterUserPassword": "password",
"NodeType": "ds2.xlarge",
}
client = boto3.client("redshift", region_name="us-east-1")
client.create_cluster(**kwargs)
client.create_cluster.when.called_with(**kwargs).should.throw(
ClientError, "ClusterAlreadyExists"
)

13
tests/test_s3/test_s3.py
View file

@ -2149,6 +2149,19 @@ def test_boto3_copy_object_with_versioning():
data.should.equal(b"test2")
@mock_s3
def test_s3_abort_multipart_data_with_invalid_upload_and_key():
client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
client.create_bucket(Bucket="blah")
with assert_raises(Exception) as err:
client.abort_multipart_upload(
Bucket="blah", Key="foobar", UploadId="dummy_upload_id"
)
err.exception.response["Error"]["Code"].should.equal("NoSuchUpload")
@mock_s3
def test_boto3_copy_object_from_unversioned_to_versioned_bucket():
client = boto3.client("s3", region_name=DEFAULT_REGION_NAME)

50
tests/test_ses/test_ses.py
View file

@ -127,3 +127,53 @@ def test_send_raw_email():
send_quota["GetSendQuotaResponse"]["GetSendQuotaResult"]["SentLast24Hours"]
)
sent_count.should.equal(1)
@mock_ses_deprecated
def test_get_send_statistics():
conn = boto.connect_ses("the_key", "the_secret")
conn.send_email.when.called_with(
"test@example.com",
"test subject",
"<span>test body</span>",
"test_to@example.com",
format="html",
).should.throw(BotoServerError)
# tests to verify rejects in get_send_statistics
result = conn.get_send_statistics()
reject_count = int(
result["GetSendStatisticsResponse"]["GetSendStatisticsResult"][
"SendDataPoints"
][0]["Rejects"]
)
delivery_count = int(
result["GetSendStatisticsResponse"]["GetSendStatisticsResult"][
"SendDataPoints"
][0]["DeliveryAttempts"]
)
reject_count.should.equal(1)
delivery_count.should.equal(0)
conn.verify_email_identity("test@example.com")
conn.send_email(
"test@example.com", "test subject", "test body", "test_to@example.com"
)
# tests to delivery attempts in get_send_statistics
result = conn.get_send_statistics()
reject_count = int(
result["GetSendStatisticsResponse"]["GetSendStatisticsResult"][
"SendDataPoints"
][0]["Rejects"]
)
delivery_count = int(
result["GetSendStatisticsResponse"]["GetSendStatisticsResult"][
"SendDataPoints"
][0]["DeliveryAttempts"]
)
reject_count.should.equal(1)
delivery_count.should.equal(1)

50
tests/test_ses/test_ses_boto3.py
View file

@ -4,6 +4,8 @@ import boto3
from botocore.exceptions import ClientError
from six.moves.email_mime_multipart import MIMEMultipart
from six.moves.email_mime_text import MIMEText
from nose.tools import assert_raises
import sure # noqa
@ -227,3 +229,51 @@ def test_send_email_notification_with_encoded_sender():
Message={"Subject": {"Data": "hi",}, "Body": {"Text": {"Data": "there",}}},
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
@mock_ses
def test_create_configuration_set():
conn = boto3.client("ses", region_name="us-east-1")
conn.create_configuration_set(ConfigurationSet=dict({"Name": "test"}))
conn.create_configuration_set_event_destination(
ConfigurationSetName="test",
EventDestination={
"Name": "snsEvent",
"Enabled": True,
"MatchingEventTypes": ["send",],
"SNSDestination": {
"TopicARN": "arn:aws:sns:us-east-1:123456789012:myTopic"
},
},
)
with assert_raises(ClientError) as ex:
conn.create_configuration_set_event_destination(
ConfigurationSetName="failtest",
EventDestination={
"Name": "snsEvent",
"Enabled": True,
"MatchingEventTypes": ["send",],
"SNSDestination": {
"TopicARN": "arn:aws:sns:us-east-1:123456789012:myTopic"
},
},
)
ex.exception.response["Error"]["Code"].should.equal("ConfigurationSetDoesNotExist")
with assert_raises(ClientError) as ex:
conn.create_configuration_set_event_destination(
ConfigurationSetName="test",
EventDestination={
"Name": "snsEvent",
"Enabled": True,
"MatchingEventTypes": ["send",],
"SNSDestination": {
"TopicARN": "arn:aws:sns:us-east-1:123456789012:myTopic"
},
},
)
ex.exception.response["Error"]["Code"].should.equal("EventDestinationAlreadyExists")

26
tests/test_ssm/test_ssm_boto3.py
View file

@ -1,5 +1,7 @@
from __future__ import unicode_literals
import string
import boto3
import botocore.exceptions
import sure # noqa
@ -300,6 +302,30 @@ def test_get_parameter():
)
@mock_ssm
def test_get_parameters_errors():
client = boto3.client("ssm", region_name="us-east-1")
ssm_parameters = {name: "value" for name in string.ascii_lowercase[:11]}
for name, value in ssm_parameters.items():
client.put_parameter(Name=name, Value=value, Type="String")
with assert_raises(ClientError) as e:
client.get_parameters(Names=list(ssm_parameters.keys()))
ex = e.exception
ex.operation_name.should.equal("GetParameters")
ex.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
ex.response["Error"]["Code"].should.contain("ValidationException")
ex.response["Error"]["Message"].should.equal(
"1 validation error detected: "
"Value '[{}]' at 'names' failed to satisfy constraint: "
"Member must have length less than or equal to 10.".format(
", ".join(ssm_parameters.keys())
)
)
@mock_ssm
def test_get_nonexistant_parameter():
client = boto3.client("ssm", region_name="us-east-1")