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

Merge branch 'master' into feature/dynamodb_item_limit

Browse source
This commit is contained in:
Bert Blommers 2020-04-16 07:07:59 +01:00
commit 47d80621f9
45 changed files with 2220 additions and 393 deletions
Download patch file Download diff file Expand all files Collapse all files

175
tests/test_apigateway/test_apigateway.py
View file

@ -1483,6 +1483,181 @@ def test_deployment():
stage["description"].should.equal("_new_description_")
@mock_apigateway
def test_create_domain_names():
client = boto3.client("apigateway", region_name="us-west-2")
domain_name = "testDomain"
test_certificate_name = "test.certificate"
test_certificate_private_key = "testPrivateKey"
# success case with valid params
response = client.create_domain_name(
domainName=domain_name,
certificateName=test_certificate_name,
certificatePrivateKey=test_certificate_private_key,
)
response["domainName"].should.equal(domain_name)
response["certificateName"].should.equal(test_certificate_name)
# without domain name it should throw BadRequestException
with assert_raises(ClientError) as ex:
client.create_domain_name(domainName="")
ex.exception.response["Error"]["Message"].should.equal("No Domain Name specified")
ex.exception.response["Error"]["Code"].should.equal("BadRequestException")
@mock_apigateway
def test_get_domain_names():
client = boto3.client("apigateway", region_name="us-west-2")
# without any domain names already present
result = client.get_domain_names()
result["items"].should.equal([])
domain_name = "testDomain"
test_certificate_name = "test.certificate"
response = client.create_domain_name(
domainName=domain_name, certificateName=test_certificate_name
)
response["domainName"].should.equal(domain_name)
response["certificateName"].should.equal(test_certificate_name)
response["domainNameStatus"].should.equal("AVAILABLE")
# after adding a new domain name
result = client.get_domain_names()
result["items"][0]["domainName"].should.equal(domain_name)
result["items"][0]["certificateName"].should.equal(test_certificate_name)
result["items"][0]["domainNameStatus"].should.equal("AVAILABLE")
@mock_apigateway
def test_get_domain_name():
client = boto3.client("apigateway", region_name="us-west-2")
domain_name = "testDomain"
# quering an invalid domain name which is not present
with assert_raises(ClientError) as ex:
client.get_domain_name(domainName=domain_name)
ex.exception.response["Error"]["Message"].should.equal(
"Invalid Domain Name specified"
)
ex.exception.response["Error"]["Code"].should.equal("NotFoundException")
# adding a domain name
client.create_domain_name(domainName=domain_name)
# retrieving the data of added domain name.
result = client.get_domain_name(domainName=domain_name)
result["domainName"].should.equal(domain_name)
result["domainNameStatus"].should.equal("AVAILABLE")
@mock_apigateway
def test_create_model():
client = boto3.client("apigateway", region_name="us-west-2")
response = client.create_rest_api(name="my_api", description="this is my api")
rest_api_id = response["id"]
dummy_rest_api_id = "a12b3c4d"
model_name = "testModel"
description = "test model"
content_type = "application/json"
# success case with valid params
response = client.create_model(
restApiId=rest_api_id,
name=model_name,
description=description,
contentType=content_type,
)
response["name"].should.equal(model_name)
response["description"].should.equal(description)
# with an invalid rest_api_id it should throw NotFoundException
with assert_raises(ClientError) as ex:
client.create_model(
restApiId=dummy_rest_api_id,
name=model_name,
description=description,
contentType=content_type,
)
ex.exception.response["Error"]["Message"].should.equal(
"Invalid Rest API Id specified"
)
ex.exception.response["Error"]["Code"].should.equal("NotFoundException")
with assert_raises(ClientError) as ex:
client.create_model(
restApiId=rest_api_id,
name="",
description=description,
contentType=content_type,
)
ex.exception.response["Error"]["Message"].should.equal("No Model Name specified")
ex.exception.response["Error"]["Code"].should.equal("BadRequestException")
@mock_apigateway
def test_get_api_models():
client = boto3.client("apigateway", region_name="us-west-2")
response = client.create_rest_api(name="my_api", description="this is my api")
rest_api_id = response["id"]
model_name = "testModel"
description = "test model"
content_type = "application/json"
# when no models are present
result = client.get_models(restApiId=rest_api_id)
result["items"].should.equal([])
# add a model
client.create_model(
restApiId=rest_api_id,
name=model_name,
description=description,
contentType=content_type,
)
# get models after adding
result = client.get_models(restApiId=rest_api_id)
result["items"][0]["name"] = model_name
result["items"][0]["description"] = description
@mock_apigateway
def test_get_model_by_name():
client = boto3.client("apigateway", region_name="us-west-2")
response = client.create_rest_api(name="my_api", description="this is my api")
rest_api_id = response["id"]
dummy_rest_api_id = "a12b3c4d"
model_name = "testModel"
description = "test model"
content_type = "application/json"
# add a model
client.create_model(
restApiId=rest_api_id,
name=model_name,
description=description,
contentType=content_type,
)
# get models after adding
result = client.get_model(restApiId=rest_api_id, modelName=model_name)
result["name"] = model_name
result["description"] = description
with assert_raises(ClientError) as ex:
client.get_model(restApiId=dummy_rest_api_id, modelName=model_name)
ex.exception.response["Error"]["Message"].should.equal(
"Invalid Rest API Id specified"
)
ex.exception.response["Error"]["Code"].should.equal("NotFoundException")
@mock_apigateway
def test_get_model_with_invalid_name():
client = boto3.client("apigateway", region_name="us-west-2")
response = client.create_rest_api(name="my_api", description="this is my api")
rest_api_id = response["id"]
# test with an invalid model name
with assert_raises(ClientError) as ex:
client.get_model(restApiId=rest_api_id, modelName="fake")
ex.exception.response["Error"]["Message"].should.equal(
"Invalid Model Name specified"
)
ex.exception.response["Error"]["Code"].should.equal("NotFoundException")
@mock_apigateway
def test_http_proxying_integration():
responses.add(

614
tests/test_autoscaling/test_autoscaling.py
View file

@ -843,13 +843,41 @@ def test_describe_autoscaling_instances_boto3():
NewInstancesProtectedFromScaleIn=True,
)
response = client.describe_auto_scaling_instances()
len(response["AutoScalingInstances"]).should.equal(5)
for instance in response["AutoScalingInstances"]:
instance["AutoScalingGroupName"].should.equal("test_asg")
instance["AvailabilityZone"].should.equal("us-east-1a")
instance["ProtectedFromScaleIn"].should.equal(True)
@mock_autoscaling
def test_describe_autoscaling_instances_instanceid_filter():
mocked_networking = setup_networking()
client = boto3.client("autoscaling", region_name="us-east-1")
_ = client.create_launch_configuration(
LaunchConfigurationName="test_launch_configuration"
)
_ = client.create_auto_scaling_group(
AutoScalingGroupName="test_asg",
LaunchConfigurationName="test_launch_configuration",
MinSize=0,
MaxSize=20,
DesiredCapacity=5,
VPCZoneIdentifier=mocked_networking["subnet1"],
NewInstancesProtectedFromScaleIn=True,
)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
instance_ids = [
instance["InstanceId"]
for instance in response["AutoScalingGroups"][0]["Instances"]
]
response = client.describe_auto_scaling_instances(InstanceIds=instance_ids)
response = client.describe_auto_scaling_instances(
InstanceIds=instance_ids[0:2]
) # Filter by first 2 of 5
len(response["AutoScalingInstances"]).should.equal(2)
for instance in response["AutoScalingInstances"]:
instance["AutoScalingGroupName"].should.equal("test_asg")
instance["AvailabilityZone"].should.equal("us-east-1a")
@ -1074,8 +1102,6 @@ def test_detach_one_instance_decrement():
ec2_client = boto3.client("ec2", region_name="us-east-1")
response = ec2_client.describe_instances(InstanceIds=[instance_to_detach])
response = client.detach_instances(
AutoScalingGroupName="test_asg",
InstanceIds=[instance_to_detach],
@ -1128,8 +1154,6 @@ def test_detach_one_instance():
ec2_client = boto3.client("ec2", region_name="us-east-1")
response = ec2_client.describe_instances(InstanceIds=[instance_to_detach])
response = client.detach_instances(
AutoScalingGroupName="test_asg",
InstanceIds=[instance_to_detach],
@ -1150,6 +1174,516 @@ def test_detach_one_instance():
tags.should.have.length_of(2)
@mock_autoscaling
@mock_ec2
def test_standby_one_instance_decrement():
mocked_networking = setup_networking()
client = boto3.client("autoscaling", region_name="us-east-1")
_ = client.create_launch_configuration(
LaunchConfigurationName="test_launch_configuration"
)
client.create_auto_scaling_group(
AutoScalingGroupName="test_asg",
LaunchConfigurationName="test_launch_configuration",
MinSize=0,
MaxSize=2,
DesiredCapacity=2,
Tags=[
{
"ResourceId": "test_asg",
"ResourceType": "auto-scaling-group",
"Key": "propogated-tag-key",
"Value": "propagate-tag-value",
"PropagateAtLaunch": True,
}
],
VPCZoneIdentifier=mocked_networking["subnet1"],
)
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"]
ec2_client = boto3.client("ec2", region_name="us-east-1")
response = client.enter_standby(
AutoScalingGroupName="test_asg",
InstanceIds=[instance_to_standby],
ShouldDecrementDesiredCapacity=True,
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
response["AutoScalingGroups"][0]["Instances"].should.have.length_of(2)
response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(1)
response = client.describe_auto_scaling_instances(InstanceIds=[instance_to_standby])
response["AutoScalingInstances"][0]["LifecycleState"].should.equal("Standby")
# test to ensure tag has been retained (standby instance is still part of the ASG)
response = ec2_client.describe_instances()
for reservation in response["Reservations"]:
for instance in reservation["Instances"]:
tags = instance["Tags"]
tags.should.have.length_of(2)
@mock_autoscaling
@mock_ec2
def test_standby_one_instance():
mocked_networking = setup_networking()
client = boto3.client("autoscaling", region_name="us-east-1")
_ = client.create_launch_configuration(
LaunchConfigurationName="test_launch_configuration"
)
client.create_auto_scaling_group(
AutoScalingGroupName="test_asg",
LaunchConfigurationName="test_launch_configuration",
MinSize=0,
MaxSize=2,
DesiredCapacity=2,
Tags=[
{
"ResourceId": "test_asg",
"ResourceType": "auto-scaling-group",
"Key": "propogated-tag-key",
"Value": "propagate-tag-value",
"PropagateAtLaunch": True,
}
],
VPCZoneIdentifier=mocked_networking["subnet1"],
)
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"]
ec2_client = boto3.client("ec2", region_name="us-east-1")
response = client.enter_standby(
AutoScalingGroupName="test_asg",
InstanceIds=[instance_to_standby],
ShouldDecrementDesiredCapacity=False,
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
response["AutoScalingGroups"][0]["Instances"].should.have.length_of(3)
response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(2)
response = client.describe_auto_scaling_instances(InstanceIds=[instance_to_standby])
response["AutoScalingInstances"][0]["LifecycleState"].should.equal("Standby")
# test to ensure tag has been retained (standby instance is still part of the ASG)
response = ec2_client.describe_instances()
for reservation in response["Reservations"]:
for instance in reservation["Instances"]:
tags = instance["Tags"]
tags.should.have.length_of(2)
@mock_elb
@mock_autoscaling
@mock_ec2
def test_standby_elb_update():
mocked_networking = setup_networking()
client = boto3.client("autoscaling", region_name="us-east-1")
_ = client.create_launch_configuration(
LaunchConfigurationName="test_launch_configuration"
)
client.create_auto_scaling_group(
AutoScalingGroupName="test_asg",
LaunchConfigurationName="test_launch_configuration",
MinSize=0,
MaxSize=2,
DesiredCapacity=2,
Tags=[
{
"ResourceId": "test_asg",
"ResourceType": "auto-scaling-group",
"Key": "propogated-tag-key",
"Value": "propagate-tag-value",
"PropagateAtLaunch": True,
}
],
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"]
response = client.enter_standby(
AutoScalingGroupName="test_asg",
InstanceIds=[instance_to_standby],
ShouldDecrementDesiredCapacity=False,
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
response["AutoScalingGroups"][0]["Instances"].should.have.length_of(3)
response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(2)
response = client.describe_auto_scaling_instances(InstanceIds=[instance_to_standby])
response["AutoScalingInstances"][0]["LifecycleState"].should.equal("Standby")
response = elb_client.describe_load_balancers(LoadBalancerNames=["my-lb"])
list(response["LoadBalancerDescriptions"][0]["Instances"]).should.have.length_of(2)
@mock_autoscaling
@mock_ec2
def test_standby_terminate_instance_decrement():
mocked_networking = setup_networking()
client = boto3.client("autoscaling", region_name="us-east-1")
_ = client.create_launch_configuration(
LaunchConfigurationName="test_launch_configuration"
)
client.create_auto_scaling_group(
AutoScalingGroupName="test_asg",
LaunchConfigurationName="test_launch_configuration",
MinSize=0,
MaxSize=3,
DesiredCapacity=2,
Tags=[
{
"ResourceId": "test_asg",
"ResourceType": "auto-scaling-group",
"Key": "propogated-tag-key",
"Value": "propagate-tag-value",
"PropagateAtLaunch": True,
}
],
VPCZoneIdentifier=mocked_networking["subnet1"],
)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
instance_to_standby_terminate = response["AutoScalingGroups"][0]["Instances"][0][
"InstanceId"
]
ec2_client = boto3.client("ec2", region_name="us-east-1")
response = client.enter_standby(
AutoScalingGroupName="test_asg",
InstanceIds=[instance_to_standby_terminate],
ShouldDecrementDesiredCapacity=False,
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
response["AutoScalingGroups"][0]["Instances"].should.have.length_of(3)
response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(2)
response = client.describe_auto_scaling_instances(
InstanceIds=[instance_to_standby_terminate]
)
response["AutoScalingInstances"][0]["LifecycleState"].should.equal("Standby")
response = client.terminate_instance_in_auto_scaling_group(
InstanceId=instance_to_standby_terminate, ShouldDecrementDesiredCapacity=True
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
# AWS still decrements desired capacity ASG if requested, even if the terminated instance is in standby
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
response["AutoScalingGroups"][0]["Instances"].should.have.length_of(1)
response["AutoScalingGroups"][0]["Instances"][0]["InstanceId"].should_not.equal(
instance_to_standby_terminate
)
response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(1)
response = ec2_client.describe_instances(
InstanceIds=[instance_to_standby_terminate]
)
response["Reservations"][0]["Instances"][0]["State"]["Name"].should.equal(
"terminated"
)
@mock_autoscaling
@mock_ec2
def test_standby_terminate_instance_no_decrement():
mocked_networking = setup_networking()
client = boto3.client("autoscaling", region_name="us-east-1")
_ = client.create_launch_configuration(
LaunchConfigurationName="test_launch_configuration"
)
client.create_auto_scaling_group(
AutoScalingGroupName="test_asg",
LaunchConfigurationName="test_launch_configuration",
MinSize=0,
MaxSize=3,
DesiredCapacity=2,
Tags=[
{
"ResourceId": "test_asg",
"ResourceType": "auto-scaling-group",
"Key": "propogated-tag-key",
"Value": "propagate-tag-value",
"PropagateAtLaunch": True,
}
],
VPCZoneIdentifier=mocked_networking["subnet1"],
)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
instance_to_standby_terminate = response["AutoScalingGroups"][0]["Instances"][0][
"InstanceId"
]
ec2_client = boto3.client("ec2", region_name="us-east-1")
response = client.enter_standby(
AutoScalingGroupName="test_asg",
InstanceIds=[instance_to_standby_terminate],
ShouldDecrementDesiredCapacity=False,
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
response["AutoScalingGroups"][0]["Instances"].should.have.length_of(3)
response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(2)
response = client.describe_auto_scaling_instances(
InstanceIds=[instance_to_standby_terminate]
)
response["AutoScalingInstances"][0]["LifecycleState"].should.equal("Standby")
response = client.terminate_instance_in_auto_scaling_group(
InstanceId=instance_to_standby_terminate, ShouldDecrementDesiredCapacity=False
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
group = response["AutoScalingGroups"][0]
group["Instances"].should.have.length_of(2)
instance_to_standby_terminate.shouldnt.be.within(
[x["InstanceId"] for x in group["Instances"]]
)
group["DesiredCapacity"].should.equal(2)
response = ec2_client.describe_instances(
InstanceIds=[instance_to_standby_terminate]
)
response["Reservations"][0]["Instances"][0]["State"]["Name"].should.equal(
"terminated"
)
@mock_autoscaling
@mock_ec2
def test_standby_detach_instance_decrement():
mocked_networking = setup_networking()
client = boto3.client("autoscaling", region_name="us-east-1")
_ = client.create_launch_configuration(
LaunchConfigurationName="test_launch_configuration"
)
client.create_auto_scaling_group(
AutoScalingGroupName="test_asg",
LaunchConfigurationName="test_launch_configuration",
MinSize=0,
MaxSize=3,
DesiredCapacity=2,
Tags=[
{
"ResourceId": "test_asg",
"ResourceType": "auto-scaling-group",
"Key": "propogated-tag-key",
"Value": "propagate-tag-value",
"PropagateAtLaunch": True,
}
],
VPCZoneIdentifier=mocked_networking["subnet1"],
)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
instance_to_standby_detach = response["AutoScalingGroups"][0]["Instances"][0][
"InstanceId"
]
ec2_client = boto3.client("ec2", region_name="us-east-1")
response = client.enter_standby(
AutoScalingGroupName="test_asg",
InstanceIds=[instance_to_standby_detach],
ShouldDecrementDesiredCapacity=False,
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
response["AutoScalingGroups"][0]["Instances"].should.have.length_of(3)
response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(2)
response = client.describe_auto_scaling_instances(
InstanceIds=[instance_to_standby_detach]
)
response["AutoScalingInstances"][0]["LifecycleState"].should.equal("Standby")
response = client.detach_instances(
AutoScalingGroupName="test_asg",
InstanceIds=[instance_to_standby_detach],
ShouldDecrementDesiredCapacity=True,
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
# AWS still decrements desired capacity ASG if requested, even if the detached instance was in standby
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
response["AutoScalingGroups"][0]["Instances"].should.have.length_of(1)
response["AutoScalingGroups"][0]["Instances"][0]["InstanceId"].should_not.equal(
instance_to_standby_detach
)
response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(1)
response = ec2_client.describe_instances(InstanceIds=[instance_to_standby_detach])
response["Reservations"][0]["Instances"][0]["State"]["Name"].should.equal("running")
@mock_autoscaling
@mock_ec2
def test_standby_detach_instance_no_decrement():
mocked_networking = setup_networking()
client = boto3.client("autoscaling", region_name="us-east-1")
_ = client.create_launch_configuration(
LaunchConfigurationName="test_launch_configuration"
)
client.create_auto_scaling_group(
AutoScalingGroupName="test_asg",
LaunchConfigurationName="test_launch_configuration",
MinSize=0,
MaxSize=3,
DesiredCapacity=2,
Tags=[
{
"ResourceId": "test_asg",
"ResourceType": "auto-scaling-group",
"Key": "propogated-tag-key",
"Value": "propagate-tag-value",
"PropagateAtLaunch": True,
}
],
VPCZoneIdentifier=mocked_networking["subnet1"],
)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
instance_to_standby_detach = response["AutoScalingGroups"][0]["Instances"][0][
"InstanceId"
]
ec2_client = boto3.client("ec2", region_name="us-east-1")
response = client.enter_standby(
AutoScalingGroupName="test_asg",
InstanceIds=[instance_to_standby_detach],
ShouldDecrementDesiredCapacity=False,
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
response["AutoScalingGroups"][0]["Instances"].should.have.length_of(3)
response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(2)
response = client.describe_auto_scaling_instances(
InstanceIds=[instance_to_standby_detach]
)
response["AutoScalingInstances"][0]["LifecycleState"].should.equal("Standby")
response = client.detach_instances(
AutoScalingGroupName="test_asg",
InstanceIds=[instance_to_standby_detach],
ShouldDecrementDesiredCapacity=False,
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
group = response["AutoScalingGroups"][0]
group["Instances"].should.have.length_of(2)
instance_to_standby_detach.shouldnt.be.within(
[x["InstanceId"] for x in group["Instances"]]
)
group["DesiredCapacity"].should.equal(2)
response = ec2_client.describe_instances(InstanceIds=[instance_to_standby_detach])
response["Reservations"][0]["Instances"][0]["State"]["Name"].should.equal("running")
@mock_autoscaling
@mock_ec2
def test_standby_exit_standby():
mocked_networking = setup_networking()
client = boto3.client("autoscaling", region_name="us-east-1")
_ = client.create_launch_configuration(
LaunchConfigurationName="test_launch_configuration"
)
client.create_auto_scaling_group(
AutoScalingGroupName="test_asg",
LaunchConfigurationName="test_launch_configuration",
MinSize=0,
MaxSize=3,
DesiredCapacity=2,
Tags=[
{
"ResourceId": "test_asg",
"ResourceType": "auto-scaling-group",
"Key": "propogated-tag-key",
"Value": "propagate-tag-value",
"PropagateAtLaunch": True,
}
],
VPCZoneIdentifier=mocked_networking["subnet1"],
)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
instance_to_standby_exit_standby = response["AutoScalingGroups"][0]["Instances"][0][
"InstanceId"
]
ec2_client = boto3.client("ec2", region_name="us-east-1")
response = client.enter_standby(
AutoScalingGroupName="test_asg",
InstanceIds=[instance_to_standby_exit_standby],
ShouldDecrementDesiredCapacity=False,
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
response["AutoScalingGroups"][0]["Instances"].should.have.length_of(3)
response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(2)
response = client.describe_auto_scaling_instances(
InstanceIds=[instance_to_standby_exit_standby]
)
response["AutoScalingInstances"][0]["LifecycleState"].should.equal("Standby")
response = client.exit_standby(
AutoScalingGroupName="test_asg", InstanceIds=[instance_to_standby_exit_standby],
)
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
group = response["AutoScalingGroups"][0]
group["Instances"].should.have.length_of(3)
instance_to_standby_exit_standby.should.be.within(
[x["InstanceId"] for x in group["Instances"]]
)
group["DesiredCapacity"].should.equal(3)
response = ec2_client.describe_instances(
InstanceIds=[instance_to_standby_exit_standby]
)
response["Reservations"][0]["Instances"][0]["State"]["Name"].should.equal("running")
@mock_autoscaling
@mock_ec2
def test_attach_one_instance():
@ -1383,7 +1917,7 @@ def test_set_desired_capacity_down_boto3():
@mock_autoscaling
@mock_ec2
def test_terminate_instance_in_autoscaling_group():
def test_terminate_instance_via_ec2_in_autoscaling_group():
mocked_networking = setup_networking()
client = boto3.client("autoscaling", region_name="us-east-1")
_ = client.create_launch_configuration(
@ -1412,3 +1946,71 @@ def test_terminate_instance_in_autoscaling_group():
for instance in response["AutoScalingGroups"][0]["Instances"]
)
replaced_instance_id.should_not.equal(original_instance_id)
@mock_autoscaling
@mock_ec2
def test_terminate_instance_in_auto_scaling_group_decrement():
mocked_networking = setup_networking()
client = boto3.client("autoscaling", region_name="us-east-1")
_ = client.create_launch_configuration(
LaunchConfigurationName="test_launch_configuration"
)
_ = client.create_auto_scaling_group(
AutoScalingGroupName="test_asg",
LaunchConfigurationName="test_launch_configuration",
MinSize=0,
DesiredCapacity=1,
MaxSize=2,
VPCZoneIdentifier=mocked_networking["subnet1"],
NewInstancesProtectedFromScaleIn=False,
)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
original_instance_id = next(
instance["InstanceId"]
for instance in response["AutoScalingGroups"][0]["Instances"]
)
client.terminate_instance_in_auto_scaling_group(
InstanceId=original_instance_id, ShouldDecrementDesiredCapacity=True
)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
response["AutoScalingGroups"][0]["Instances"].should.equal([])
response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(0)
@mock_autoscaling
@mock_ec2
def test_terminate_instance_in_auto_scaling_group_no_decrement():
mocked_networking = setup_networking()
client = boto3.client("autoscaling", region_name="us-east-1")
_ = client.create_launch_configuration(
LaunchConfigurationName="test_launch_configuration"
)
_ = client.create_auto_scaling_group(
AutoScalingGroupName="test_asg",
LaunchConfigurationName="test_launch_configuration",
MinSize=0,
DesiredCapacity=1,
MaxSize=2,
VPCZoneIdentifier=mocked_networking["subnet1"],
NewInstancesProtectedFromScaleIn=False,
)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
original_instance_id = next(
instance["InstanceId"]
for instance in response["AutoScalingGroups"][0]["Instances"]
)
client.terminate_instance_in_auto_scaling_group(
InstanceId=original_instance_id, ShouldDecrementDesiredCapacity=False
)
response = client.describe_auto_scaling_groups(AutoScalingGroupNames=["test_asg"])
replaced_instance_id = next(
instance["InstanceId"]
for instance in response["AutoScalingGroups"][0]["Instances"]
)
replaced_instance_id.should_not.equal(original_instance_id)
response["AutoScalingGroups"][0]["DesiredCapacity"].should.equal(1)

125
tests/test_cloudformation/test_cloudformation_stack_integration.py
View file

@ -29,6 +29,7 @@ from moto import (
mock_ec2_deprecated,
mock_elb,
mock_elb_deprecated,
mock_events,
mock_iam_deprecated,
mock_kms,
mock_lambda,
@ -2372,10 +2373,128 @@ def test_create_log_group_using_fntransform():
}
cf_conn = boto3.client("cloudformation", "us-west-2")
cf_conn.create_stack(
StackName="test_stack", TemplateBody=json.dumps(template),
)
cf_conn.create_stack(StackName="test_stack", TemplateBody=json.dumps(template))
logs_conn = boto3.client("logs", region_name="us-west-2")
log_group = logs_conn.describe_log_groups()["logGroups"][0]
log_group["logGroupName"].should.equal("some-log-group")
log_group["retentionInDays"].should.be.equal(90)
@mock_cloudformation
@mock_events
def test_stack_events_create_rule_integration():
events_template = {
"AWSTemplateFormatVersion": "2010-09-09",
"Resources": {
"Event": {
"Type": "AWS::Events::Rule",
"Properties": {
"Name": "quick-fox",
"State": "ENABLED",
"ScheduleExpression": "rate(5 minutes)",
},
}
},
}
cf_conn = boto3.client("cloudformation", "us-west-2")
cf_conn.create_stack(
StackName="test_stack", TemplateBody=json.dumps(events_template)
)
rules = boto3.client("events", "us-west-2").list_rules()
rules["Rules"].should.have.length_of(1)
rules["Rules"][0]["Name"].should.equal("quick-fox")
rules["Rules"][0]["State"].should.equal("ENABLED")
rules["Rules"][0]["ScheduleExpression"].should.equal("rate(5 minutes)")
@mock_cloudformation
@mock_events
def test_stack_events_delete_rule_integration():
events_template = {
"AWSTemplateFormatVersion": "2010-09-09",
"Resources": {
"Event": {
"Type": "AWS::Events::Rule",
"Properties": {
"Name": "quick-fox",
"State": "ENABLED",
"ScheduleExpression": "rate(5 minutes)",
},
}
},
}
cf_conn = boto3.client("cloudformation", "us-west-2")
cf_conn.create_stack(
StackName="test_stack", TemplateBody=json.dumps(events_template)
)
rules = boto3.client("events", "us-west-2").list_rules()
rules["Rules"].should.have.length_of(1)
cf_conn.delete_stack(StackName="test_stack")
rules = boto3.client("events", "us-west-2").list_rules()
rules["Rules"].should.have.length_of(0)
@mock_cloudformation
@mock_events
def test_stack_events_create_rule_without_name_integration():
events_template = {
"AWSTemplateFormatVersion": "2010-09-09",
"Resources": {
"Event": {
"Type": "AWS::Events::Rule",
"Properties": {
"State": "ENABLED",
"ScheduleExpression": "rate(5 minutes)",
},
}
},
}
cf_conn = boto3.client("cloudformation", "us-west-2")
cf_conn.create_stack(
StackName="test_stack", TemplateBody=json.dumps(events_template)
)
rules = boto3.client("events", "us-west-2").list_rules()
rules["Rules"][0]["Name"].should.contain("test_stack-Event-")
@mock_cloudformation
@mock_events
@mock_logs
def test_stack_events_create_rule_as_target():
events_template = {
"AWSTemplateFormatVersion": "2010-09-09",
"Resources": {
"SecurityGroup": {
"Type": "AWS::Logs::LogGroup",
"Properties": {
"LogGroupName": {"Fn::GetAtt": ["Event", "Arn"]},
"RetentionInDays": 3,
},
},
"Event": {
"Type": "AWS::Events::Rule",
"Properties": {
"State": "ENABLED",
"ScheduleExpression": "rate(5 minutes)",
},
},
},
}
cf_conn = boto3.client("cloudformation", "us-west-2")
cf_conn.create_stack(
StackName="test_stack", TemplateBody=json.dumps(events_template)
)
rules = boto3.client("events", "us-west-2").list_rules()
log_groups = boto3.client("logs", "us-west-2").describe_log_groups()
rules["Rules"][0]["Name"].should.contain("test_stack-Event-")
log_groups["logGroups"][0]["logGroupName"].should.equal(rules["Rules"][0]["Arn"])
log_groups["logGroups"][0]["retentionInDays"].should.equal(3)

45
tests/test_dynamodb2/test_dynamodb.py
View file

@ -1345,6 +1345,25 @@ def test_get_item_returns_consumed_capacity():
assert "TableName" in response["ConsumedCapacity"]
@mock_dynamodb2
def test_put_empty_item():
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={})
ex.exception.response["Error"]["Message"].should.equal(
"One or more parameter values were invalid: Missing the key structure_id in the item"
)
ex.exception.response["Error"]["Code"].should.equal("ValidationException")
@mock_dynamodb2
def test_put_item_nonexisting_hash_key():
dynamodb = boto3.resource("dynamodb", region_name="us-east-1")
@ -1361,6 +1380,32 @@ def test_put_item_nonexisting_hash_key():
ex.exception.response["Error"]["Message"].should.equal(
"One or more parameter values were invalid: Missing the key structure_id in the item"
)
ex.exception.response["Error"]["Code"].should.equal("ValidationException")
@mock_dynamodb2
def test_put_item_nonexisting_range_key():
dynamodb = boto3.resource("dynamodb", region_name="us-east-1")
dynamodb.create_table(
AttributeDefinitions=[
{"AttributeName": "structure_id", "AttributeType": "S"},
{"AttributeName": "added_at", "AttributeType": "N"},
],
TableName="test",
KeySchema=[
{"AttributeName": "structure_id", "KeyType": "HASH"},
{"AttributeName": "added_at", "KeyType": "RANGE"},
],
ProvisionedThroughput={"ReadCapacityUnits": 123, "WriteCapacityUnits": 123},
)
table = dynamodb.Table("test")
with assert_raises(ClientError) as ex:
table.put_item(Item={"structure_id": "abcdef"})
ex.exception.response["Error"]["Message"].should.equal(
"One or more parameter values were invalid: Missing the key added_at in the item"
)
ex.exception.response["Error"]["Code"].should.equal("ValidationException")
def test_filter_expression():

1
tests/test_dynamodbstreams/test_dynamodbstreams.py
View file

@ -134,6 +134,7 @@ class TestCore:
"id": {"S": "entry1"},
"first_col": {"S": "bar"},
"second_col": {"S": "baz"},
"a": {"L": [{"M": {"b": {"S": "bar1"}}}]},
},
)
conn.delete_item(TableName="test-streams", Key={"id": {"S": "entry1"}})

12
tests/test_ec2/test_availability_zones_and_regions.py
View file

@ -52,3 +52,15 @@ def test_boto3_availability_zones():
resp = conn.describe_availability_zones()
for rec in resp["AvailabilityZones"]:
rec["ZoneName"].should.contain(region)
@mock_ec2
def test_boto3_zoneId_in_availability_zones():
conn = boto3.client("ec2", "us-east-1")
resp = conn.describe_availability_zones()
for rec in resp["AvailabilityZones"]:
rec.get("ZoneId").should.contain("use1")
conn = boto3.client("ec2", "us-west-1")
resp = conn.describe_availability_zones()
for rec in resp["AvailabilityZones"]:
rec.get("ZoneId").should.contain("usw1")

39
tests/test_ec2/test_elastic_block_store.py
View file

@ -53,6 +53,45 @@ def test_create_and_delete_volume():
cm.exception.request_id.should_not.be.none
@mock_ec2_deprecated
def test_delete_attached_volume():
conn = boto.ec2.connect_to_region("us-east-1")
reservation = conn.run_instances("ami-1234abcd")
# create an instance
instance = reservation.instances[0]
# create a volume
volume = conn.create_volume(80, "us-east-1a")
# attach volume to instance
volume.attach(instance.id, "/dev/sdh")
volume.update()
volume.volume_state().should.equal("in-use")
volume.attachment_state().should.equal("attached")
volume.attach_data.instance_id.should.equal(instance.id)
# attempt to delete volume
# assert raises VolumeInUseError
with assert_raises(EC2ResponseError) as ex:
volume.delete()
ex.exception.error_code.should.equal("VolumeInUse")
ex.exception.status.should.equal(400)
ex.exception.message.should.equal(
"Volume {0} is currently attached to {1}".format(volume.id, instance.id)
)
volume.detach()
volume.update()
volume.volume_state().should.equal("available")
volume.delete()
all_volumes = conn.get_all_volumes()
my_volume = [item for item in all_volumes if item.id == volume.id]
my_volume.should.have.length_of(0)
@mock_ec2_deprecated
def test_create_encrypted_volume_dryrun():
conn = boto.ec2.connect_to_region("us-east-1")

15
tests/test_ec2/test_instances.py
View file

@ -1166,6 +1166,21 @@ def test_describe_instance_status_with_instance_filter_deprecated():
cm.exception.request_id.should_not.be.none
@mock_ec2
def test_describe_instance_credit_specifications():
conn = boto3.client("ec2", region_name="us-west-1")
# We want to filter based on this one
reservation = conn.run_instances(ImageId="ami-1234abcd", MinCount=1, MaxCount=1)
result = conn.describe_instance_credit_specifications(
InstanceIds=[reservation["Instances"][0]["InstanceId"]]
)
assert (
result["InstanceCreditSpecifications"][0]["InstanceId"]
== reservation["Instances"][0]["InstanceId"]
)
@mock_ec2
def test_describe_instance_status_with_instance_filter():
conn = boto3.client("ec2", region_name="us-west-1")

65
tests/test_ecs/test_ecs_boto3.py
View file

@ -1122,6 +1122,71 @@ def test_run_task():
response["tasks"][0]["stoppedReason"].should.equal("")
@mock_ec2
@mock_ecs
def test_run_task_default_cluster():
client = boto3.client("ecs", region_name="us-east-1")
ec2 = boto3.resource("ec2", region_name="us-east-1")
test_cluster_name = "default"
_ = client.create_cluster(clusterName=test_cluster_name)
test_instance = ec2.create_instances(
ImageId="ami-1234abcd", MinCount=1, MaxCount=1
)[0]
instance_id_document = json.dumps(
ec2_utils.generate_instance_identity_document(test_instance)
)
response = client.register_container_instance(
cluster=test_cluster_name, instanceIdentityDocument=instance_id_document
)
_ = client.register_task_definition(
family="test_ecs_task",
containerDefinitions=[
{
"name": "hello_world",
"image": "docker/hello-world:latest",
"cpu": 1024,
"memory": 400,
"essential": True,
"environment": [
{"name": "AWS_ACCESS_KEY_ID", "value": "SOME_ACCESS_KEY"}
],
"logConfiguration": {"logDriver": "json-file"},
}
],
)
response = client.run_task(
launchType="FARGATE",
overrides={},
taskDefinition="test_ecs_task",
count=2,
startedBy="moto",
)
len(response["tasks"]).should.equal(2)
response["tasks"][0]["taskArn"].should.contain(
"arn:aws:ecs:us-east-1:012345678910:task/"
)
response["tasks"][0]["clusterArn"].should.equal(
"arn:aws:ecs:us-east-1:012345678910:cluster/default"
)
response["tasks"][0]["taskDefinitionArn"].should.equal(
"arn:aws:ecs:us-east-1:012345678910:task-definition/test_ecs_task:1"
)
response["tasks"][0]["containerInstanceArn"].should.contain(
"arn:aws:ecs:us-east-1:012345678910:container-instance/"
)
response["tasks"][0]["overrides"].should.equal({})
response["tasks"][0]["lastStatus"].should.equal("RUNNING")
response["tasks"][0]["desiredStatus"].should.equal("RUNNING")
response["tasks"][0]["startedBy"].should.equal("moto")
response["tasks"][0]["stoppedReason"].should.equal("")
@mock_ec2
@mock_ecs
def test_start_task():

29
tests/test_events/test_events.py
View file

@ -1,15 +1,16 @@
from moto.events.models import EventsBackend
from moto.events import mock_events
import json
import random
import unittest
import boto3
import sure # noqa
from botocore.exceptions import ClientError
from moto.core.exceptions import JsonRESTError
from nose.tools import assert_raises
from moto.core import ACCOUNT_ID
from moto.core.exceptions import JsonRESTError
from moto.events import mock_events
from moto.events.models import EventsBackend
RULES = [
{"Name": "test1", "ScheduleExpression": "rate(5 minutes)"},
@ -75,6 +76,28 @@ def generate_environment():
return client
@mock_events
def test_put_rule():
client = boto3.client("events", "us-west-2")
client.list_rules()["Rules"].should.have.length_of(0)
rule_data = {
"Name": "my-event",
"ScheduleExpression": "rate(5 minutes)",
"EventPattern": '{"source": ["test-source"]}',
}
client.put_rule(**rule_data)
rules = client.list_rules()["Rules"]
rules.should.have.length_of(1)
rules[0]["Name"].should.equal(rule_data["Name"])
rules[0]["ScheduleExpression"].should.equal(rule_data["ScheduleExpression"])
rules[0]["EventPattern"].should.equal(rule_data["EventPattern"])
rules[0]["State"].should.equal("ENABLED")
@mock_events
def test_list_rules():
client = generate_environment()

13
tests/test_logs/test_logs.py
View file

@ -12,17 +12,14 @@ _logs_region = "us-east-1" if settings.TEST_SERVER_MODE else "us-west-2"
@mock_logs
def test_log_group_create():
def test_create_log_group():
conn = boto3.client("logs", "us-west-2")
log_group_name = "dummy"
response = conn.create_log_group(logGroupName=log_group_name)
response = conn.describe_log_groups(logGroupNamePrefix=log_group_name)
assert len(response["logGroups"]) == 1
# AWS defaults to Never Expire for log group retention
assert response["logGroups"][0].get("retentionInDays") == None
response = conn.create_log_group(logGroupName="dummy")
response = conn.describe_log_groups()
response = conn.delete_log_group(logGroupName=log_group_name)
response["logGroups"].should.have.length_of(1)
response["logGroups"][0].should_not.have.key("retentionInDays")
@mock_logs

18
tests/test_s3/test_s3.py
View file

@ -2413,6 +2413,24 @@ def test_boto3_put_bucket_tagging():
"Cannot provide multiple Tags with the same key"
)
# Cannot put tags that are "system" tags - i.e. tags that start with "aws:"
with assert_raises(ClientError) as ce:
s3.put_bucket_tagging(
Bucket=bucket_name,
Tagging={"TagSet": [{"Key": "aws:sometag", "Value": "nope"}]},
)
e = ce.exception
e.response["Error"]["Code"].should.equal("InvalidTag")
e.response["Error"]["Message"].should.equal(
"System tags cannot be added/updated by requester"
)
# This is OK though:
s3.put_bucket_tagging(
Bucket=bucket_name,
Tagging={"TagSet": [{"Key": "something:aws:stuff", "Value": "this is fine"}]},
)
@mock_s3
def test_boto3_get_bucket_tagging():

54
tests/test_sns/test_publishing_boto3.py
View file

@ -148,34 +148,42 @@ def test_publish_to_sqs_msg_attr_byte_value():
conn.create_topic(Name="some-topic")
response = conn.list_topics()
topic_arn = response["Topics"][0]["TopicArn"]
sqs_conn = boto3.resource("sqs", region_name="us-east-1")
queue = sqs_conn.create_queue(QueueName="test-queue")
sqs = boto3.resource("sqs", region_name="us-east-1")
queue = sqs.create_queue(QueueName="test-queue")
conn.subscribe(
TopicArn=topic_arn, Protocol="sqs", Endpoint=queue.attributes["QueueArn"],
)
queue_raw = sqs.create_queue(QueueName="test-queue-raw")
conn.subscribe(
TopicArn=topic_arn,
Protocol="sqs",
Endpoint="arn:aws:sqs:us-east-1:{}:test-queue".format(ACCOUNT_ID),
Endpoint=queue_raw.attributes["QueueArn"],
Attributes={"RawMessageDelivery": "true"},
)
message = "my message"
conn.publish(
TopicArn=topic_arn,
Message=message,
Message="my message",
MessageAttributes={
"store": {"DataType": "Binary", "BinaryValue": b"\x02\x03\x04"}
},
)
messages = queue.receive_messages(MaxNumberOfMessages=5)
message_attributes = [json.loads(m.body)["MessageAttributes"] for m in messages]
message_attributes.should.equal(
[
{
"store": {
"Type": "Binary",
"Value": base64.b64encode(b"\x02\x03\x04").decode(),
}
message = json.loads(queue.receive_messages()[0].body)
message["Message"].should.equal("my message")
message["MessageAttributes"].should.equal(
{
"store": {
"Type": "Binary",
"Value": base64.b64encode(b"\x02\x03\x04").decode(),
}
]
}
)
message = queue_raw.receive_messages()[0]
message.body.should.equal("my message")
message.message_attributes.should.equal(
{"store": {"DataType": "Binary", "BinaryValue": b"\x02\x03\x04"}}
)
@ -187,6 +195,12 @@ def test_publish_to_sqs_msg_attr_number_type():
sqs = boto3.resource("sqs", region_name="us-east-1")
queue = sqs.create_queue(QueueName="test-queue")
topic.subscribe(Protocol="sqs", Endpoint=queue.attributes["QueueArn"])
queue_raw = sqs.create_queue(QueueName="test-queue-raw")
topic.subscribe(
Protocol="sqs",
Endpoint=queue_raw.attributes["QueueArn"],
Attributes={"RawMessageDelivery": "true"},
)
topic.publish(
Message="test message",
@ -199,6 +213,12 @@ def test_publish_to_sqs_msg_attr_number_type():
{"retries": {"Type": "Number", "Value": 0}}
)
message = queue_raw.receive_messages()[0]
message.body.should.equal("test message")
message.message_attributes.should.equal(
{"retries": {"DataType": "Number", "StringValue": "0"}}
)
@mock_sns
def test_publish_sms():

2
tests/test_stepfunctions/test_stepfunctions.py
View file

@ -516,7 +516,7 @@ def test_state_machine_describe_execution_after_stoppage():
description = client.describe_execution(executionArn=execution["executionArn"])
#
description["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
description["status"].should.equal("SUCCEEDED")
description["status"].should.equal("ABORTED")
description["stopDate"].should.be.a(datetime)