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

Merge branch 'master' into feature/dynamodb_transact_write_items

Browse source
This commit is contained in:
Bert Blommers 2020-04-26 10:24:27 +01:00
commit 56aa454397
101 changed files with 8946 additions and 663 deletions
Download patch file Download diff file Expand all files Collapse all files

191
tests/test_apigateway/test_apigateway.py
View file

@ -69,6 +69,22 @@ def test_create_rest_api_with_tags():
response["tags"].should.equal({"MY_TAG1": "MY_VALUE1"})
@mock_apigateway
def test_create_rest_api_with_policy():
client = boto3.client("apigateway", region_name="us-west-2")
policy = '{"Version": "2012-10-17","Statement": []}'
response = client.create_rest_api(
name="my_api", description="this is my api", policy=policy
)
api_id = response["id"]
response = client.get_rest_api(restApiId=api_id)
assert "policy" in response
response["policy"].should.equal(policy)
@mock_apigateway
def test_create_rest_api_invalid_apikeysource():
client = boto3.client("apigateway", region_name="us-west-2")
@ -1483,6 +1499,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)

36
tests/test_awslambda/test_lambda.py
View file

@ -1677,6 +1677,42 @@ def test_create_function_with_unknown_arn():
)
@mock_lambda
def test_remove_function_permission():
conn = boto3.client("lambda", _lambda_region)
zip_content = get_test_zip_file1()
conn.create_function(
FunctionName="testFunction",
Runtime="python2.7",
Role=(get_role_name()),
Handler="lambda_function.handler",
Code={"ZipFile": zip_content},
Description="test lambda function",
Timeout=3,
MemorySize=128,
Publish=True,
)
conn.add_permission(
FunctionName="testFunction",
StatementId="1",
Action="lambda:InvokeFunction",
Principal="432143214321",
SourceArn="arn:aws:lambda:us-west-2:account-id:function:helloworld",
SourceAccount="123412341234",
EventSourceToken="blah",
Qualifier="2",
)
remove = conn.remove_permission(
FunctionName="testFunction", StatementId="1", Qualifier="2",
)
remove["ResponseMetadata"]["HTTPStatusCode"].should.equal(204)
policy = conn.get_policy(FunctionName="testFunction", Qualifier="2")["Policy"]
policy = json.loads(policy)
policy["Statement"].should.equal([])
def create_invalid_lambda(role):
conn = boto3.client("lambda", _lambda_region)
zip_content = get_test_zip_file1()

17
tests/test_cloudformation/test_cloudformation_stack_crud_boto3.py
View file

@ -835,8 +835,10 @@ def test_describe_change_set():
)
stack = cf_conn.describe_change_set(ChangeSetName="NewChangeSet")
stack["ChangeSetName"].should.equal("NewChangeSet")
stack["StackName"].should.equal("NewStack")
stack["Status"].should.equal("REVIEW_IN_PROGRESS")
cf_conn.create_change_set(
StackName="NewStack",
@ -851,15 +853,30 @@ def test_describe_change_set():
@mock_cloudformation
@mock_ec2
def test_execute_change_set_w_arn():
cf_conn = boto3.client("cloudformation", region_name="us-east-1")
ec2 = boto3.client("ec2", region_name="us-east-1")
# Verify no instances exist at the moment
ec2.describe_instances()["Reservations"].should.have.length_of(0)
# Create a Change set, and verify no resources have been created yet
change_set = cf_conn.create_change_set(
StackName="NewStack",
TemplateBody=dummy_template_json,
ChangeSetName="NewChangeSet",
ChangeSetType="CREATE",
)
ec2.describe_instances()["Reservations"].should.have.length_of(0)
cf_conn.describe_change_set(ChangeSetName="NewChangeSet")["Status"].should.equal(
"REVIEW_IN_PROGRESS"
)
# Execute change set
cf_conn.execute_change_set(ChangeSetName=change_set["Id"])
# Verify that the status has changed, and the appropriate resources have been created
cf_conn.describe_change_set(ChangeSetName="NewChangeSet")["Status"].should.equal(
"CREATE_COMPLETE"
)
ec2.describe_instances()["Reservations"].should.have.length_of(1)
@mock_cloudformation

62
tests/test_cloudformation/test_cloudformation_stack_integration.py
View file

@ -495,7 +495,7 @@ def test_autoscaling_group_with_elb():
"my-as-group": {
"Type": "AWS::AutoScaling::AutoScalingGroup",
"Properties": {
"AvailabilityZones": ["us-east1"],
"AvailabilityZones": ["us-east-1a"],
"LaunchConfigurationName": {"Ref": "my-launch-config"},
"MinSize": "2",
"MaxSize": "2",
@ -522,7 +522,7 @@ def test_autoscaling_group_with_elb():
"my-elb": {
"Type": "AWS::ElasticLoadBalancing::LoadBalancer",
"Properties": {
"AvailabilityZones": ["us-east1"],
"AvailabilityZones": ["us-east-1a"],
"Listeners": [
{
"LoadBalancerPort": "80",
@ -545,10 +545,10 @@ def test_autoscaling_group_with_elb():
web_setup_template_json = json.dumps(web_setup_template)
conn = boto.cloudformation.connect_to_region("us-west-1")
conn = boto.cloudformation.connect_to_region("us-east-1")
conn.create_stack("web_stack", template_body=web_setup_template_json)
autoscale_conn = boto.ec2.autoscale.connect_to_region("us-west-1")
autoscale_conn = boto.ec2.autoscale.connect_to_region("us-east-1")
autoscale_group = autoscale_conn.get_all_groups()[0]
autoscale_group.launch_config_name.should.contain("my-launch-config")
autoscale_group.load_balancers[0].should.equal("my-elb")
@ -557,7 +557,7 @@ def test_autoscaling_group_with_elb():
autoscale_conn.get_all_launch_configurations().should.have.length_of(1)
# Confirm the ELB was actually created
elb_conn = boto.ec2.elb.connect_to_region("us-west-1")
elb_conn = boto.ec2.elb.connect_to_region("us-east-1")
elb_conn.get_all_load_balancers().should.have.length_of(1)
stack = conn.describe_stacks()[0]
@ -584,7 +584,7 @@ def test_autoscaling_group_with_elb():
elb_resource.physical_resource_id.should.contain("my-elb")
# confirm the instances were created with the right tags
ec2_conn = boto.ec2.connect_to_region("us-west-1")
ec2_conn = boto.ec2.connect_to_region("us-east-1")
reservations = ec2_conn.get_all_reservations()
len(reservations).should.equal(1)
reservation = reservations[0]
@ -604,7 +604,7 @@ def test_autoscaling_group_update():
"my-as-group": {
"Type": "AWS::AutoScaling::AutoScalingGroup",
"Properties": {
"AvailabilityZones": ["us-west-1"],
"AvailabilityZones": ["us-west-1a"],
"LaunchConfigurationName": {"Ref": "my-launch-config"},
"MinSize": "2",
"MaxSize": "2",
@ -2373,13 +2373,12 @@ 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
@ -2400,7 +2399,7 @@ def test_stack_events_create_rule_integration():
}
cf_conn = boto3.client("cloudformation", "us-west-2")
cf_conn.create_stack(
StackName="test_stack", TemplateBody=json.dumps(events_template),
StackName="test_stack", TemplateBody=json.dumps(events_template)
)
rules = boto3.client("events", "us-west-2").list_rules()
@ -2428,7 +2427,7 @@ def test_stack_events_delete_rule_integration():
}
cf_conn = boto3.client("cloudformation", "us-west-2")
cf_conn.create_stack(
StackName="test_stack", TemplateBody=json.dumps(events_template),
StackName="test_stack", TemplateBody=json.dumps(events_template)
)
rules = boto3.client("events", "us-west-2").list_rules()
@ -2457,8 +2456,45 @@ def test_stack_events_create_rule_without_name_integration():
}
cf_conn = boto3.client("cloudformation", "us-west-2")
cf_conn.create_stack(
StackName="test_stack", TemplateBody=json.dumps(events_template),
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)

40
tests/test_cloudwatch/test_cloudwatch.py
View file

@ -1,9 +1,10 @@
import boto
from boto.ec2.cloudwatch.alarm import MetricAlarm
from boto.s3.key import Key
from datetime import datetime
import sure # noqa
from moto import mock_cloudwatch_deprecated
from moto import mock_cloudwatch_deprecated, mock_s3_deprecated
def alarm_fixture(name="tester", action=None):
@ -83,10 +84,11 @@ def test_put_metric_data():
)
metrics = conn.list_metrics()
metrics.should.have.length_of(1)
metric_names = [m for m in metrics if m.name == "metric"]
metric_names.should.have(1)
metric = metrics[0]
metric.namespace.should.equal("tester")
metric.name.should.equal("metric")
metric.name.should.equal("Metric:metric")
dict(metric.dimensions).should.equal({"InstanceId": ["i-0123456,i-0123457"]})
@ -153,3 +155,35 @@ def test_get_metric_statistics():
datapoint = datapoints[0]
datapoint.should.have.key("Minimum").which.should.equal(1.5)
datapoint.should.have.key("Timestamp").which.should.equal(metric_timestamp)
@mock_s3_deprecated
@mock_cloudwatch_deprecated
def test_cloudwatch_return_s3_metrics():
region = "us-east-1"
cw = boto.ec2.cloudwatch.connect_to_region(region)
s3 = boto.s3.connect_to_region(region)
bucket_name_1 = "test-bucket-1"
bucket_name_2 = "test-bucket-2"
bucket1 = s3.create_bucket(bucket_name=bucket_name_1)
key = Key(bucket1)
key.key = "the-key"
key.set_contents_from_string("foobar" * 4)
s3.create_bucket(bucket_name=bucket_name_2)
metrics_s3_bucket_1 = cw.list_metrics(dimensions={"BucketName": bucket_name_1})
# Verify that the OOTB S3 metrics are available for the created buckets
len(metrics_s3_bucket_1).should.be(2)
metric_names = [m.name for m in metrics_s3_bucket_1]
sorted(metric_names).should.equal(
["Metric:BucketSizeBytes", "Metric:NumberOfObjects"]
)
# Explicit clean up - the metrics for these buckets are messing with subsequent tests
key.delete()
s3.delete_bucket(bucket_name_1)
s3.delete_bucket(bucket_name_2)

275
tests/test_cloudwatch/test_cloudwatch_boto3.py
View file

@ -3,6 +3,7 @@
import boto3
from botocore.exceptions import ClientError
from datetime import datetime, timedelta
from freezegun import freeze_time
from nose.tools import assert_raises
from uuid import uuid4
import pytz
@ -154,7 +155,7 @@ def test_put_metric_data_no_dimensions():
metrics.should.have.length_of(1)
metric = metrics[0]
metric["Namespace"].should.equal("tester")
metric["MetricName"].should.equal("metric")
metric["MetricName"].should.equal("Metric:metric")
@mock_cloudwatch
@ -182,7 +183,7 @@ def test_put_metric_data_with_statistics():
metrics.should.have.length_of(1)
metric = metrics[0]
metric["Namespace"].should.equal("tester")
metric["MetricName"].should.equal("statmetric")
metric["MetricName"].should.equal("Metric:statmetric")
# TODO: test statistics - https://github.com/spulec/moto/issues/1615
@ -211,6 +212,35 @@ def test_get_metric_statistics():
datapoint["Sum"].should.equal(1.5)
@mock_cloudwatch
@freeze_time("2020-02-10 18:44:05")
def test_custom_timestamp():
utc_now = datetime.now(tz=pytz.utc)
time = "2020-02-10T18:44:09Z"
cw = boto3.client("cloudwatch", "eu-west-1")
cw.put_metric_data(
Namespace="tester",
MetricData=[dict(MetricName="metric1", Value=1.5, Timestamp=time)],
)
cw.put_metric_data(
Namespace="tester",
MetricData=[
dict(MetricName="metric2", Value=1.5, Timestamp=datetime(2020, 2, 10))
],
)
stats = cw.get_metric_statistics(
Namespace="tester",
MetricName="metric",
StartTime=utc_now - timedelta(seconds=60),
EndTime=utc_now + timedelta(seconds=60),
Period=60,
Statistics=["SampleCount", "Sum"],
)
@mock_cloudwatch
def test_list_metrics():
cloudwatch = boto3.client("cloudwatch", "eu-west-1")
@ -233,8 +263,16 @@ def test_list_metrics():
# Verify format
res.should.equal(
[
{u"Namespace": "list_test_1/", u"Dimensions": [], u"MetricName": "metric1"},
{u"Namespace": "list_test_1/", u"Dimensions": [], u"MetricName": "metric1"},
{
u"Namespace": "list_test_1/",
u"Dimensions": [],
u"MetricName": "Metric:metric1",
},
{
u"Namespace": "list_test_1/",
u"Dimensions": [],
u"MetricName": "Metric:metric1",
},
]
)
# Verify unknown namespace still has no results
@ -292,3 +330,232 @@ def create_metrics(cloudwatch, namespace, metrics=5, data_points=5):
Namespace=namespace,
MetricData=[{"MetricName": metric_name, "Value": j, "Unit": "Seconds"}],
)
@mock_cloudwatch
def test_get_metric_data_within_timeframe():
utc_now = datetime.now(tz=pytz.utc)
cloudwatch = boto3.client("cloudwatch", "eu-west-1")
namespace1 = "my_namespace/"
# put metric data
values = [0, 2, 4, 3.5, 7, 100]
cloudwatch.put_metric_data(
Namespace=namespace1,
MetricData=[
{"MetricName": "metric1", "Value": val, "Unit": "Seconds"} for val in values
],
)
# get_metric_data
stats = ["Average", "Sum", "Minimum", "Maximum"]
response = cloudwatch.get_metric_data(
MetricDataQueries=[
{
"Id": "result_" + stat,
"MetricStat": {
"Metric": {"Namespace": namespace1, "MetricName": "metric1"},
"Period": 60,
"Stat": stat,
},
}
for stat in stats
],
StartTime=utc_now - timedelta(seconds=60),
EndTime=utc_now + timedelta(seconds=60),
)
#
# Assert Average/Min/Max/Sum is returned as expected
avg = [
res for res in response["MetricDataResults"] if res["Id"] == "result_Average"
][0]
avg["Label"].should.equal("metric1 Average")
avg["StatusCode"].should.equal("Complete")
[int(val) for val in avg["Values"]].should.equal([19])
sum_ = [res for res in response["MetricDataResults"] if res["Id"] == "result_Sum"][
0
]
sum_["Label"].should.equal("metric1 Sum")
sum_["StatusCode"].should.equal("Complete")
[val for val in sum_["Values"]].should.equal([sum(values)])
min_ = [
res for res in response["MetricDataResults"] if res["Id"] == "result_Minimum"
][0]
min_["Label"].should.equal("metric1 Minimum")
min_["StatusCode"].should.equal("Complete")
[int(val) for val in min_["Values"]].should.equal([0])
max_ = [
res for res in response["MetricDataResults"] if res["Id"] == "result_Maximum"
][0]
max_["Label"].should.equal("metric1 Maximum")
max_["StatusCode"].should.equal("Complete")
[int(val) for val in max_["Values"]].should.equal([100])
@mock_cloudwatch
def test_get_metric_data_partially_within_timeframe():
utc_now = datetime.now(tz=pytz.utc)
yesterday = utc_now - timedelta(days=1)
last_week = utc_now - timedelta(days=7)
cloudwatch = boto3.client("cloudwatch", "eu-west-1")
namespace1 = "my_namespace/"
# put metric data
values = [0, 2, 4, 3.5, 7, 100]
cloudwatch.put_metric_data(
Namespace=namespace1,
MetricData=[
{
"MetricName": "metric1",
"Value": 10,
"Unit": "Seconds",
"Timestamp": utc_now,
}
],
)
cloudwatch.put_metric_data(
Namespace=namespace1,
MetricData=[
{
"MetricName": "metric1",
"Value": 20,
"Unit": "Seconds",
"Timestamp": yesterday,
}
],
)
cloudwatch.put_metric_data(
Namespace=namespace1,
MetricData=[
{
"MetricName": "metric1",
"Value": 50,
"Unit": "Seconds",
"Timestamp": last_week,
}
],
)
# get_metric_data
response = cloudwatch.get_metric_data(
MetricDataQueries=[
{
"Id": "result",
"MetricStat": {
"Metric": {"Namespace": namespace1, "MetricName": "metric1"},
"Period": 60,
"Stat": "Sum",
},
}
],
StartTime=yesterday - timedelta(seconds=60),
EndTime=utc_now + timedelta(seconds=60),
)
#
# Assert Last week's data is not returned
len(response["MetricDataResults"]).should.equal(1)
sum_ = response["MetricDataResults"][0]
sum_["Label"].should.equal("metric1 Sum")
sum_["StatusCode"].should.equal("Complete")
sum_["Values"].should.equal([30.0])
@mock_cloudwatch
def test_get_metric_data_outside_timeframe():
utc_now = datetime.now(tz=pytz.utc)
last_week = utc_now - timedelta(days=7)
cloudwatch = boto3.client("cloudwatch", "eu-west-1")
namespace1 = "my_namespace/"
# put metric data
cloudwatch.put_metric_data(
Namespace=namespace1,
MetricData=[
{
"MetricName": "metric1",
"Value": 50,
"Unit": "Seconds",
"Timestamp": last_week,
}
],
)
# get_metric_data
response = cloudwatch.get_metric_data(
MetricDataQueries=[
{
"Id": "result",
"MetricStat": {
"Metric": {"Namespace": namespace1, "MetricName": "metric1"},
"Period": 60,
"Stat": "Sum",
},
}
],
StartTime=utc_now - timedelta(seconds=60),
EndTime=utc_now + timedelta(seconds=60),
)
#
# Assert Last week's data is not returned
len(response["MetricDataResults"]).should.equal(1)
response["MetricDataResults"][0]["Id"].should.equal("result")
response["MetricDataResults"][0]["StatusCode"].should.equal("Complete")
response["MetricDataResults"][0]["Values"].should.equal([])
@mock_cloudwatch
def test_get_metric_data_for_multiple_metrics():
utc_now = datetime.now(tz=pytz.utc)
cloudwatch = boto3.client("cloudwatch", "eu-west-1")
namespace = "my_namespace/"
# put metric data
cloudwatch.put_metric_data(
Namespace=namespace,
MetricData=[
{
"MetricName": "metric1",
"Value": 50,
"Unit": "Seconds",
"Timestamp": utc_now,
}
],
)
cloudwatch.put_metric_data(
Namespace=namespace,
MetricData=[
{
"MetricName": "metric2",
"Value": 25,
"Unit": "Seconds",
"Timestamp": utc_now,
}
],
)
# get_metric_data
response = cloudwatch.get_metric_data(
MetricDataQueries=[
{
"Id": "result1",
"MetricStat": {
"Metric": {"Namespace": namespace, "MetricName": "metric1"},
"Period": 60,
"Stat": "Sum",
},
},
{
"Id": "result2",
"MetricStat": {
"Metric": {"Namespace": namespace, "MetricName": "metric2"},
"Period": 60,
"Stat": "Sum",
},
},
],
StartTime=utc_now - timedelta(seconds=60),
EndTime=utc_now + timedelta(seconds=60),
)
#
len(response["MetricDataResults"]).should.equal(2)
res1 = [res for res in response["MetricDataResults"] if res["Id"] == "result1"][0]
res1["Values"].should.equal([50.0])
res2 = [res for res in response["MetricDataResults"] if res["Id"] == "result2"][0]
res2["Values"].should.equal([25.0])

8
tests/test_cognitoidentity/test_cognitoidentity.py
View file

@ -7,6 +7,7 @@ from nose.tools import assert_raises
from moto import mock_cognitoidentity
from moto.cognitoidentity.utils import get_random_identity_id
from moto.core import ACCOUNT_ID
from uuid import UUID
@mock_cognitoidentity
@ -83,8 +84,10 @@ def test_describe_identity_pool_with_invalid_id_raises_error():
# testing a helper function
def test_get_random_identity_id():
assert len(get_random_identity_id("us-west-2")) > 0
assert len(get_random_identity_id("us-west-2").split(":")[1]) == 19
identity_id = get_random_identity_id("us-west-2")
region, id = identity_id.split(":")
region.should.equal("us-west-2")
UUID(id, version=4) # Will throw an error if it's not a valid UUID
@mock_cognitoidentity
@ -96,7 +99,6 @@ def test_get_id():
IdentityPoolId="us-west-2:12345",
Logins={"someurl": "12345"},
)
print(result)
assert (
result.get("IdentityId", "").startswith("us-west-2")
or result.get("ResponseMetadata").get("HTTPStatusCode") == 200

1
tests/test_cognitoidentity/test_server.py
View file

@ -48,6 +48,5 @@ def test_get_id():
},
)
print(res.data)
json_data = json.loads(res.data.decode("utf-8"))
assert ":" in json_data["IdentityId"]

2
tests/test_config/test_config.py
View file

@ -11,6 +11,8 @@ from moto import mock_s3
from moto.config import mock_config
from moto.core import ACCOUNT_ID
import sure # noqa
@mock_config
def test_put_configuration_recorder():

178
tests/test_dynamodb2/test_dynamodb.py
View file

@ -1,5 +1,6 @@
from __future__ import unicode_literals, print_function
import re
from decimal import Decimal
import six
@ -1453,6 +1454,13 @@ def test_filter_expression():
filter_expr.expr(row1).should.be(True)
filter_expr.expr(row2).should.be(False)
# lowercase AND test
filter_expr = moto.dynamodb2.comparisons.get_filter_expression(
"Id > :v0 and Subs < :v1", {}, {":v0": {"N": "5"}, ":v1": {"N": "7"}}
)
filter_expr.expr(row1).should.be(True)
filter_expr.expr(row2).should.be(False)
# OR test
filter_expr = moto.dynamodb2.comparisons.get_filter_expression(
"Id = :v0 OR Id=:v1", {}, {":v0": {"N": "5"}, ":v1": {"N": "8"}}
@ -2146,13 +2154,33 @@ def test_update_item_on_map():
# Nonexistent nested attributes are supported for existing top-level attributes.
table.update_item(
Key={"forum_name": "the-key", "subject": "123"},
UpdateExpression="SET body.#nested.#data = :tb, body.nested.#nonexistentnested.#data = :tb2",
UpdateExpression="SET body.#nested.#data = :tb",
ExpressionAttributeNames={"#nested": "nested", "#data": "data",},
ExpressionAttributeValues={":tb": "new_value"},
)
# Running this against AWS DDB gives an exception so make sure it also fails.:
with assert_raises(client.exceptions.ClientError):
# botocore.exceptions.ClientError: An error occurred (ValidationException) when calling the UpdateItem
# operation: The document path provided in the update expression is invalid for update
table.update_item(
Key={"forum_name": "the-key", "subject": "123"},
UpdateExpression="SET body.#nested.#nonexistentnested.#data = :tb2",
ExpressionAttributeNames={
"#nested": "nested",
"#nonexistentnested": "nonexistentnested",
"#data": "data",
},
ExpressionAttributeValues={":tb2": "other_value"},
)
table.update_item(
Key={"forum_name": "the-key", "subject": "123"},
UpdateExpression="SET body.#nested.#nonexistentnested = :tb2",
ExpressionAttributeNames={
"#nested": "nested",
"#nonexistentnested": "nonexistentnested",
"#data": "data",
},
ExpressionAttributeValues={":tb": "new_value", ":tb2": "other_value"},
ExpressionAttributeValues={":tb2": {"data": "other_value"}},
)
resp = table.scan()
@ -2160,8 +2188,8 @@ def test_update_item_on_map():
{"nested": {"data": "new_value", "nonexistentnested": {"data": "other_value"}}}
)
# Test nested value for a nonexistent attribute.
with assert_raises(client.exceptions.ConditionalCheckFailedException):
# Test nested value for a nonexistent attribute throws a ClientError.
with assert_raises(client.exceptions.ClientError):
table.update_item(
Key={"forum_name": "the-key", "subject": "123"},
UpdateExpression="SET nonexistent.#nested = :tb",
@ -2764,7 +2792,7 @@ def test_query_gsi_with_range_key():
res = dynamodb.query(
TableName="test",
IndexName="test_gsi",
KeyConditionExpression="gsi_hash_key = :gsi_hash_key AND gsi_range_key = :gsi_range_key",
KeyConditionExpression="gsi_hash_key = :gsi_hash_key and gsi_range_key = :gsi_range_key",
ExpressionAttributeValues={
":gsi_hash_key": {"S": "key1"},
":gsi_range_key": {"S": "range1"},
@ -3183,7 +3211,10 @@ def test_remove_top_level_attribute():
TableName=table_name, Item={"id": {"S": "foo"}, "item": {"S": "bar"}}
)
client.update_item(
TableName=table_name, Key={"id": {"S": "foo"}}, UpdateExpression="REMOVE item"
TableName=table_name,
Key={"id": {"S": "foo"}},
UpdateExpression="REMOVE #i",
ExpressionAttributeNames={"#i": "item"},
)
#
result = client.get_item(TableName=table_name, Key={"id": {"S": "foo"}})["Item"]
@ -3358,21 +3389,21 @@ def test_item_size_is_under_400KB():
assert_failure_due_to_item_size(
func=client.put_item,
TableName="moto-test",
Item={"id": {"S": "foo"}, "item": {"S": large_item}},
Item={"id": {"S": "foo"}, "cont": {"S": large_item}},
)
assert_failure_due_to_item_size(
func=table.put_item, Item={"id": "bar", "item": large_item}
func=table.put_item, Item={"id": "bar", "cont": large_item}
)
assert_failure_due_to_item_size(
assert_failure_due_to_item_size_to_update(
func=client.update_item,
TableName="moto-test",
Key={"id": {"S": "foo2"}},
UpdateExpression="set item=:Item",
UpdateExpression="set cont=:Item",
ExpressionAttributeValues={":Item": {"S": large_item}},
)
# Assert op fails when updating a nested item
assert_failure_due_to_item_size(
func=table.put_item, Item={"id": "bar", "itemlist": [{"item": large_item}]}
func=table.put_item, Item={"id": "bar", "itemlist": [{"cont": large_item}]}
)
assert_failure_due_to_item_size(
func=client.put_item,
@ -3393,6 +3424,15 @@ def assert_failure_due_to_item_size(func, **kwargs):
)
def assert_failure_due_to_item_size_to_update(func, **kwargs):
with assert_raises(ClientError) as ex:
func(**kwargs)
ex.exception.response["Error"]["Code"].should.equal("ValidationException")
ex.exception.response["Error"]["Message"].should.equal(
"Item size to update has exceeded the maximum allowed size"
)
@mock_dynamodb2
# https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_Query.html#DDB-Query-request-KeyConditionExpression
def test_hash_key_cannot_use_begins_with_operations():
@ -4529,3 +4569,117 @@ def test_transact_write_items_update_with_failed_condition_expression():
items = dynamodb.scan(TableName="test-table")["Items"]
items.should.have.length_of(1)
items[0].should.equal({"email_address": {"S": "test@moto.com"}, "id": {"S": "foo"}})
@mock_dynamodb2
def test_dynamodb_max_1mb_limit():
ddb = boto3.resource("dynamodb", region_name="eu-west-1")
table_name = "populated-mock-table"
table = ddb.create_table(
TableName=table_name,
KeySchema=[
{"AttributeName": "partition_key", "KeyType": "HASH"},
{"AttributeName": "sort_key", "KeyType": "RANGE"},
],
AttributeDefinitions=[
{"AttributeName": "partition_key", "AttributeType": "S"},
{"AttributeName": "sort_key", "AttributeType": "S"},
],
BillingMode="PAY_PER_REQUEST",
)
# Populate the table
items = [
{
"partition_key": "partition_key_val", # size=30
"sort_key": "sort_key_value____" + str(i), # size=30
}
for i in range(10000, 29999)
]
with table.batch_writer() as batch:
for item in items:
batch.put_item(Item=item)
response = table.query(
KeyConditionExpression=Key("partition_key").eq("partition_key_val")
)
# We shouldn't get everything back - the total result set is well over 1MB
len(items).should.be.greater_than(response["Count"])
response["LastEvaluatedKey"].shouldnt.be(None)
def assert_raise_syntax_error(client_error, token, near):
"""
Assert whether a client_error is as expected Syntax error. Syntax error looks like: `syntax_error_template`
Args:
client_error(ClientError): The ClientError exception that was raised
token(str): The token that ws unexpected
near(str): The part in the expression that shows where the error occurs it generally has the preceding token the
optional separation and the problematic token.
"""
syntax_error_template = (
'Invalid UpdateExpression: Syntax error; token: "{token}", near: "{near}"'
)
expected_syntax_error = syntax_error_template.format(token=token, near=near)
assert client_error.response["Error"]["Code"] == "ValidationException"
assert expected_syntax_error == client_error.response["Error"]["Message"]
@mock_dynamodb2
def test_update_expression_with_numeric_literal_instead_of_value():
"""
DynamoDB requires literals to be passed in as values. If they are put literally in the expression a token error will
be raised
"""
dynamodb = boto3.client("dynamodb", region_name="eu-west-1")
dynamodb.create_table(
TableName="moto-test",
KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],
)
try:
dynamodb.update_item(
TableName="moto-test",
Key={"id": {"S": "1"}},
UpdateExpression="SET MyStr = myNum + 1",
)
assert False, "Validation exception not thrown"
except dynamodb.exceptions.ClientError as e:
assert_raise_syntax_error(e, "1", "+ 1")
@mock_dynamodb2
def test_update_expression_with_multiple_set_clauses_must_be_comma_separated():
"""
An UpdateExpression can have multiple set clauses but if they are passed in without the separating comma.
"""
dynamodb = boto3.client("dynamodb", region_name="eu-west-1")
dynamodb.create_table(
TableName="moto-test",
KeySchema=[{"AttributeName": "id", "KeyType": "HASH"}],
AttributeDefinitions=[{"AttributeName": "id", "AttributeType": "S"}],
)
try:
dynamodb.update_item(
TableName="moto-test",
Key={"id": {"S": "1"}},
UpdateExpression="SET MyStr = myNum Mystr2 myNum2",
)
assert False, "Validation exception not thrown"
except dynamodb.exceptions.ClientError as e:
assert_raise_syntax_error(e, "Mystr2", "myNum Mystr2 myNum2")
@mock_dynamodb2
def test_list_tables_exclusive_start_table_name_empty():
client = boto3.client("dynamodb", region_name="us-east-1")
resp = client.list_tables(Limit=1, ExclusiveStartTableName="whatever")
len(resp["TableNames"]).should.equal(0)

259
tests/test_dynamodb2/test_dynamodb_expression_tokenizer.py Normal file
View file

@ -0,0 +1,259 @@
from moto.dynamodb2.exceptions import (
InvalidTokenException,
InvalidExpressionAttributeNameKey,
)
from moto.dynamodb2.parsing.tokens import ExpressionTokenizer, Token
def test_expression_tokenizer_single_set_action():
set_action = "SET attrName = :attrValue"
token_list = ExpressionTokenizer.make_list(set_action)
assert token_list == [
Token(Token.ATTRIBUTE, "SET"),
Token(Token.WHITESPACE, " "),
Token(Token.ATTRIBUTE, "attrName"),
Token(Token.WHITESPACE, " "),
Token(Token.EQUAL_SIGN, "="),
Token(Token.WHITESPACE, " "),
Token(Token.ATTRIBUTE_VALUE, ":attrValue"),
]
def test_expression_tokenizer_single_set_action_leading_space():
set_action = "Set attrName = :attrValue"
token_list = ExpressionTokenizer.make_list(set_action)
assert token_list == [
Token(Token.ATTRIBUTE, "Set"),
Token(Token.WHITESPACE, " "),
Token(Token.ATTRIBUTE, "attrName"),
Token(Token.WHITESPACE, " "),
Token(Token.EQUAL_SIGN, "="),
Token(Token.WHITESPACE, " "),
Token(Token.ATTRIBUTE_VALUE, ":attrValue"),
]
def test_expression_tokenizer_single_set_action_attribute_name_leading_space():
set_action = "SET #a = :attrValue"
token_list = ExpressionTokenizer.make_list(set_action)
assert token_list == [
Token(Token.ATTRIBUTE, "SET"),
Token(Token.WHITESPACE, " "),
Token(Token.ATTRIBUTE_NAME, "#a"),
Token(Token.WHITESPACE, " "),
Token(Token.EQUAL_SIGN, "="),
Token(Token.WHITESPACE, " "),
Token(Token.ATTRIBUTE_VALUE, ":attrValue"),
]
def test_expression_tokenizer_single_set_action_trailing_space():
set_action = "SET attrName = :attrValue "
token_list = ExpressionTokenizer.make_list(set_action)
assert token_list == [
Token(Token.ATTRIBUTE, "SET"),
Token(Token.WHITESPACE, " "),
Token(Token.ATTRIBUTE, "attrName"),
Token(Token.WHITESPACE, " "),
Token(Token.EQUAL_SIGN, "="),
Token(Token.WHITESPACE, " "),
Token(Token.ATTRIBUTE_VALUE, ":attrValue"),
Token(Token.WHITESPACE, " "),
]
def test_expression_tokenizer_single_set_action_multi_spaces():
set_action = "SET attrName = :attrValue "
token_list = ExpressionTokenizer.make_list(set_action)
assert token_list == [
Token(Token.ATTRIBUTE, "SET"),
Token(Token.WHITESPACE, " "),
Token(Token.ATTRIBUTE, "attrName"),
Token(Token.WHITESPACE, " "),
Token(Token.EQUAL_SIGN, "="),
Token(Token.WHITESPACE, " "),
Token(Token.ATTRIBUTE_VALUE, ":attrValue"),
Token(Token.WHITESPACE, " "),
]
def test_expression_tokenizer_single_set_action_with_numbers_in_identifiers():
set_action = "SET attrName3 = :attr3Value"
token_list = ExpressionTokenizer.make_list(set_action)
assert token_list == [
Token(Token.ATTRIBUTE, "SET"),
Token(Token.WHITESPACE, " "),
Token(Token.ATTRIBUTE, "attrName3"),
Token(Token.WHITESPACE, " "),
Token(Token.EQUAL_SIGN, "="),
Token(Token.WHITESPACE, " "),
Token(Token.ATTRIBUTE_VALUE, ":attr3Value"),
]
def test_expression_tokenizer_single_set_action_with_underscore_in_identifier():
set_action = "SET attr_Name = :attr_Value"
token_list = ExpressionTokenizer.make_list(set_action)
assert token_list == [
Token(Token.ATTRIBUTE, "SET"),
Token(Token.WHITESPACE, " "),
Token(Token.ATTRIBUTE, "attr_Name"),
Token(Token.WHITESPACE, " "),
Token(Token.EQUAL_SIGN, "="),
Token(Token.WHITESPACE, " "),
Token(Token.ATTRIBUTE_VALUE, ":attr_Value"),
]
def test_expression_tokenizer_leading_underscore_in_attribute_name_expression():
"""Leading underscore is not allowed for an attribute name"""
set_action = "SET attrName = _idid"
try:
ExpressionTokenizer.make_list(set_action)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "_"
assert te.near == "= _idid"
def test_expression_tokenizer_leading_underscore_in_attribute_value_expression():
"""Leading underscore is allowed in an attribute value"""
set_action = "SET attrName = :_attrValue"
token_list = ExpressionTokenizer.make_list(set_action)
assert token_list == [
Token(Token.ATTRIBUTE, "SET"),
Token(Token.WHITESPACE, " "),
Token(Token.ATTRIBUTE, "attrName"),
Token(Token.WHITESPACE, " "),
Token(Token.EQUAL_SIGN, "="),
Token(Token.WHITESPACE, " "),
Token(Token.ATTRIBUTE_VALUE, ":_attrValue"),
]
def test_expression_tokenizer_single_set_action_nested_attribute():
set_action = "SET attrName.elem = :attrValue"
token_list = ExpressionTokenizer.make_list(set_action)
assert token_list == [
Token(Token.ATTRIBUTE, "SET"),
Token(Token.WHITESPACE, " "),
Token(Token.ATTRIBUTE, "attrName"),
Token(Token.DOT, "."),
Token(Token.ATTRIBUTE, "elem"),
Token(Token.WHITESPACE, " "),
Token(Token.EQUAL_SIGN, "="),
Token(Token.WHITESPACE, " "),
Token(Token.ATTRIBUTE_VALUE, ":attrValue"),
]
def test_expression_tokenizer_list_index_with_sub_attribute():
set_action = "SET itemmap.itemlist[1].foos=:Item"
token_list = ExpressionTokenizer.make_list(set_action)
assert token_list == [
Token(Token.ATTRIBUTE, "SET"),
Token(Token.WHITESPACE, " "),
Token(Token.ATTRIBUTE, "itemmap"),
Token(Token.DOT, "."),
Token(Token.ATTRIBUTE, "itemlist"),
Token(Token.OPEN_SQUARE_BRACKET, "["),
Token(Token.NUMBER, "1"),
Token(Token.CLOSE_SQUARE_BRACKET, "]"),
Token(Token.DOT, "."),
Token(Token.ATTRIBUTE, "foos"),
Token(Token.EQUAL_SIGN, "="),
Token(Token.ATTRIBUTE_VALUE, ":Item"),
]
def test_expression_tokenizer_list_index_surrounded_with_whitespace():
set_action = "SET itemlist[ 1 ]=:Item"
token_list = ExpressionTokenizer.make_list(set_action)
assert token_list == [
Token(Token.ATTRIBUTE, "SET"),
Token(Token.WHITESPACE, " "),
Token(Token.ATTRIBUTE, "itemlist"),
Token(Token.OPEN_SQUARE_BRACKET, "["),
Token(Token.WHITESPACE, " "),
Token(Token.NUMBER, "1"),
Token(Token.WHITESPACE, " "),
Token(Token.CLOSE_SQUARE_BRACKET, "]"),
Token(Token.EQUAL_SIGN, "="),
Token(Token.ATTRIBUTE_VALUE, ":Item"),
]
def test_expression_tokenizer_single_set_action_attribute_name_invalid_key():
"""
ExpressionAttributeNames contains invalid key: Syntax error; key: "#va#l2"
"""
set_action = "SET #va#l2 = 3"
try:
ExpressionTokenizer.make_list(set_action)
assert False, "Exception not raised correctly"
except InvalidExpressionAttributeNameKey as e:
assert e.key == "#va#l2"
def test_expression_tokenizer_single_set_action_attribute_name_invalid_key_double_hash():
"""
ExpressionAttributeNames contains invalid key: Syntax error; key: "#va#l"
"""
set_action = "SET #va#l = 3"
try:
ExpressionTokenizer.make_list(set_action)
assert False, "Exception not raised correctly"
except InvalidExpressionAttributeNameKey as e:
assert e.key == "#va#l"
def test_expression_tokenizer_single_set_action_attribute_name_valid_key():
set_action = "SET attr=#val2"
token_list = ExpressionTokenizer.make_list(set_action)
assert token_list == [
Token(Token.ATTRIBUTE, "SET"),
Token(Token.WHITESPACE, " "),
Token(Token.ATTRIBUTE, "attr"),
Token(Token.EQUAL_SIGN, "="),
Token(Token.ATTRIBUTE_NAME, "#val2"),
]
def test_expression_tokenizer_just_a_pipe():
set_action = "|"
try:
ExpressionTokenizer.make_list(set_action)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "|"
assert te.near == "|"
def test_expression_tokenizer_just_a_pipe_with_leading_white_spaces():
set_action = " |"
try:
ExpressionTokenizer.make_list(set_action)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "|"
assert te.near == " |"
def test_expression_tokenizer_just_a_pipe_for_set_expression():
set_action = "SET|"
try:
ExpressionTokenizer.make_list(set_action)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "|"
assert te.near == "SET|"
def test_expression_tokenizer_just_an_attribute_and_a_pipe_for_set_expression():
set_action = "SET a|"
try:
ExpressionTokenizer.make_list(set_action)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "|"
assert te.near == "a|"

405
tests/test_dynamodb2/test_dynamodb_expressions.py Normal file
View file

@ -0,0 +1,405 @@
from moto.dynamodb2.exceptions import InvalidTokenException
from moto.dynamodb2.parsing.expressions import UpdateExpressionParser
from moto.dynamodb2.parsing.reserved_keywords import ReservedKeywords
def test_get_reserved_keywords():
reserved_keywords = ReservedKeywords.get_reserved_keywords()
assert "SET" in reserved_keywords
assert "DELETE" in reserved_keywords
assert "ADD" in reserved_keywords
# REMOVE is not part of the list of reserved keywords.
assert "REMOVE" not in reserved_keywords
def test_update_expression_numeric_literal_in_expression():
set_action = "SET attrName = 3"
try:
UpdateExpressionParser.make(set_action)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "3"
assert te.near == "= 3"
def test_expression_tokenizer_multi_number_numeric_literal_in_expression():
set_action = "SET attrName = 34"
try:
UpdateExpressionParser.make(set_action)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "34"
assert te.near == "= 34"
def test_expression_tokenizer_numeric_literal_unclosed_square_bracket():
set_action = "SET MyStr[ 3"
try:
UpdateExpressionParser.make(set_action)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "<EOF>"
assert te.near == "3"
def test_expression_tokenizer_wrong_closing_bracket_with_space():
set_action = "SET MyStr[3 )"
try:
UpdateExpressionParser.make(set_action)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == ")"
assert te.near == "3 )"
def test_expression_tokenizer_wrong_closing_bracket():
set_action = "SET MyStr[3)"
try:
UpdateExpressionParser.make(set_action)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == ")"
assert te.near == "3)"
def test_expression_tokenizer_only_numeric_literal_for_set():
set_action = "SET 2"
try:
UpdateExpressionParser.make(set_action)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "2"
assert te.near == "SET 2"
def test_expression_tokenizer_only_numeric_literal():
set_action = "2"
try:
UpdateExpressionParser.make(set_action)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "2"
assert te.near == "2"
def test_expression_tokenizer_set_closing_round_bracket():
set_action = "SET )"
try:
UpdateExpressionParser.make(set_action)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == ")"
assert te.near == "SET )"
def test_expression_tokenizer_set_closing_followed_by_numeric_literal():
set_action = "SET ) 3"
try:
UpdateExpressionParser.make(set_action)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == ")"
assert te.near == "SET ) 3"
def test_expression_tokenizer_numeric_literal_unclosed_square_bracket_trailing_space():
set_action = "SET MyStr[ 3 "
try:
UpdateExpressionParser.make(set_action)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "<EOF>"
assert te.near == "3 "
def test_expression_tokenizer_unbalanced_round_brackets_only_opening():
set_action = "SET MyStr = (:_val"
try:
UpdateExpressionParser.make(set_action)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "<EOF>"
assert te.near == ":_val"
def test_expression_tokenizer_unbalanced_round_brackets_only_opening_trailing_space():
set_action = "SET MyStr = (:_val "
try:
UpdateExpressionParser.make(set_action)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "<EOF>"
assert te.near == ":_val "
def test_expression_tokenizer_unbalanced_square_brackets_only_opening():
set_action = "SET MyStr = [:_val"
try:
UpdateExpressionParser.make(set_action)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "["
assert te.near == "= [:_val"
def test_expression_tokenizer_unbalanced_square_brackets_only_opening_trailing_spaces():
set_action = "SET MyStr = [:_val "
try:
UpdateExpressionParser.make(set_action)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "["
assert te.near == "= [:_val"
def test_expression_tokenizer_unbalanced_round_brackets_multiple_opening():
set_action = "SET MyStr = (:_val + (:val2"
try:
UpdateExpressionParser.make(set_action)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "<EOF>"
assert te.near == ":val2"
def test_expression_tokenizer_unbalanced_round_brackets_only_closing():
set_action = "SET MyStr = ):_val"
try:
UpdateExpressionParser.make(set_action)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == ")"
assert te.near == "= ):_val"
def test_expression_tokenizer_unbalanced_square_brackets_only_closing():
set_action = "SET MyStr = ]:_val"
try:
UpdateExpressionParser.make(set_action)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "]"
assert te.near == "= ]:_val"
def test_expression_tokenizer_unbalanced_round_brackets_only_closing_followed_by_other_parts():
set_action = "SET MyStr = ):_val + :val2"
try:
UpdateExpressionParser.make(set_action)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == ")"
assert te.near == "= ):_val"
def test_update_expression_starts_with_keyword_reset_followed_by_identifier():
update_expression = "RESET NonExistent"
try:
UpdateExpressionParser.make(update_expression)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "RESET"
assert te.near == "RESET NonExistent"
def test_update_expression_starts_with_keyword_reset_followed_by_identifier_and_value():
update_expression = "RESET NonExistent value"
try:
UpdateExpressionParser.make(update_expression)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "RESET"
assert te.near == "RESET NonExistent"
def test_update_expression_starts_with_leading_spaces_and_keyword_reset_followed_by_identifier_and_value():
update_expression = " RESET NonExistent value"
try:
UpdateExpressionParser.make(update_expression)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "RESET"
assert te.near == " RESET NonExistent"
def test_update_expression_with_only_keyword_reset():
update_expression = "RESET"
try:
UpdateExpressionParser.make(update_expression)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "RESET"
assert te.near == "RESET"
def test_update_nested_expression_with_selector_just_should_fail_parsing_at_numeric_literal_value():
update_expression = "SET a[0].b = 5"
try:
UpdateExpressionParser.make(update_expression)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "5"
assert te.near == "= 5"
def test_update_nested_expression_with_selector_and_spaces_should_only_fail_parsing_at_numeric_literal_value():
update_expression = "SET a [ 2 ]. b = 5"
try:
UpdateExpressionParser.make(update_expression)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "5"
assert te.near == "= 5"
def test_update_nested_expression_with_double_selector_and_spaces_should_only_fail_parsing_at_numeric_literal_value():
update_expression = "SET a [2][ 3 ]. b = 5"
try:
UpdateExpressionParser.make(update_expression)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "5"
assert te.near == "= 5"
def test_update_nested_expression_should_only_fail_parsing_at_numeric_literal_value():
update_expression = "SET a . b = 5"
try:
UpdateExpressionParser.make(update_expression)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "5"
assert te.near == "= 5"
def test_nested_selectors_in_update_expression_should_fail_at_nesting():
update_expression = "SET a [ [2] ]. b = 5"
try:
UpdateExpressionParser.make(update_expression)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "["
assert te.near == "[ [2"
def test_update_expression_number_in_selector_cannot_be_splite():
update_expression = "SET a [2 1]. b = 5"
try:
UpdateExpressionParser.make(update_expression)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "1"
assert te.near == "2 1]"
def test_update_expression_cannot_have_successive_attributes():
update_expression = "SET #a a = 5"
try:
UpdateExpressionParser.make(update_expression)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "a"
assert te.near == "#a a ="
def test_update_expression_path_with_both_attribute_and_attribute_name_should_only_fail_at_numeric_value():
update_expression = "SET #a.a = 5"
try:
UpdateExpressionParser.make(update_expression)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "5"
assert te.near == "= 5"
def test_expression_tokenizer_2_same_operators_back_to_back():
set_action = "SET MyStr = NoExist + + :_val "
try:
UpdateExpressionParser.make(set_action)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "+"
assert te.near == "+ + :_val"
def test_expression_tokenizer_2_different_operators_back_to_back():
set_action = "SET MyStr = NoExist + - :_val "
try:
UpdateExpressionParser.make(set_action)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "-"
assert te.near == "+ - :_val"
def test_update_expression_remove_does_not_allow_operations():
remove_action = "REMOVE NoExist + "
try:
UpdateExpressionParser.make(remove_action)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "+"
assert te.near == "NoExist + "
def test_update_expression_add_does_not_allow_attribute_after_path():
"""value here is not really a value since a value starts with a colon (:)"""
add_expr = "ADD attr val foobar"
try:
UpdateExpressionParser.make(add_expr)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "val"
assert te.near == "attr val foobar"
def test_update_expression_add_does_not_allow_attribute_foobar_after_value():
add_expr = "ADD attr :val foobar"
try:
UpdateExpressionParser.make(add_expr)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "foobar"
assert te.near == ":val foobar"
def test_update_expression_delete_does_not_allow_attribute_after_path():
"""value here is not really a value since a value starts with a colon (:)"""
delete_expr = "DELETE attr val"
try:
UpdateExpressionParser.make(delete_expr)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "val"
assert te.near == "attr val"
def test_update_expression_delete_does_not_allow_attribute_foobar_after_value():
delete_expr = "DELETE attr :val foobar"
try:
UpdateExpressionParser.make(delete_expr)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "foobar"
assert te.near == ":val foobar"
def test_update_expression_parsing_is_not_keyword_aware():
"""path and VALUE are keywords. Yet a token error will be thrown for the numeric literal 1."""
delete_expr = "SET path = VALUE 1"
try:
UpdateExpressionParser.make(delete_expr)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "1"
assert te.near == "VALUE 1"
def test_expression_if_not_exists_is_not_valid_in_remove_statement():
set_action = "REMOVE if_not_exists(a,b)"
try:
UpdateExpressionParser.make(set_action)
assert False, "Exception not raised correctly"
except InvalidTokenException as te:
assert te.token == "("
assert te.near == "if_not_exists(a"

16
tests/test_dynamodb2/test_dynamodb_table_with_range_key.py
View file

@ -1254,14 +1254,22 @@ def test_update_item_with_expression():
item_key = {"forum_name": "the-key", "subject": "123"}
table.update_item(Key=item_key, UpdateExpression="SET field=2")
table.update_item(
Key=item_key,
UpdateExpression="SET field = :field_value",
ExpressionAttributeValues={":field_value": 2},
)
dict(table.get_item(Key=item_key)["Item"]).should.equal(
{"field": "2", "forum_name": "the-key", "subject": "123"}
{"field": Decimal("2"), "forum_name": "the-key", "subject": "123"}
)
table.update_item(Key=item_key, UpdateExpression="SET field = 3")
table.update_item(
Key=item_key,
UpdateExpression="SET field = :field_value",
ExpressionAttributeValues={":field_value": 3},
)
dict(table.get_item(Key=item_key)["Item"]).should.equal(
{"field": "3", "forum_name": "the-key", "subject": "123"}
{"field": Decimal("3"), "forum_name": "the-key", "subject": "123"}
)

63
tests/test_dynamodb2/test_dynamodb_table_without_range_key.py
View file

@ -443,23 +443,40 @@ def test_update_item_nested_remove():
dict(returned_item).should.equal({"username": "steve", "Meta": {}})
@mock_dynamodb2_deprecated
@mock_dynamodb2
def test_update_item_double_nested_remove():
conn = boto.dynamodb2.connect_to_region("us-east-1")
table = Table.create("messages", schema=[HashKey("username")])
conn = boto3.client("dynamodb", region_name="us-east-1")
conn.create_table(
TableName="messages",
KeySchema=[{"AttributeName": "username", "KeyType": "HASH"}],
AttributeDefinitions=[{"AttributeName": "username", "AttributeType": "S"}],
BillingMode="PAY_PER_REQUEST",
)
data = {"username": "steve", "Meta": {"Name": {"First": "Steve", "Last": "Urkel"}}}
table.put_item(data=data)
item = {
"username": {"S": "steve"},
"Meta": {
"M": {"Name": {"M": {"First": {"S": "Steve"}, "Last": {"S": "Urkel"}}}}
},
}
conn.put_item(TableName="messages", Item=item)
key_map = {"username": {"S": "steve"}}
# Then remove the Meta.FullName field
conn.update_item("messages", key_map, update_expression="REMOVE Meta.Name.First")
returned_item = table.get_item(username="steve")
dict(returned_item).should.equal(
{"username": "steve", "Meta": {"Name": {"Last": "Urkel"}}}
conn.update_item(
TableName="messages",
Key=key_map,
UpdateExpression="REMOVE Meta.#N.#F",
ExpressionAttributeNames={"#N": "Name", "#F": "First"},
)
returned_item = conn.get_item(TableName="messages", Key=key_map)
expected_item = {
"username": {"S": "steve"},
"Meta": {"M": {"Name": {"M": {"Last": {"S": "Urkel"}}}}},
}
dict(returned_item["Item"]).should.equal(expected_item)
@mock_dynamodb2_deprecated
def test_update_item_set():
@ -471,7 +488,10 @@ def test_update_item_set():
key_map = {"username": {"S": "steve"}}
conn.update_item(
"messages", key_map, update_expression="SET foo=bar, blah=baz REMOVE SentBy"
"messages",
key_map,
update_expression="SET foo=:bar, blah=:baz REMOVE SentBy",
expression_attribute_values={":bar": {"S": "bar"}, ":baz": {"S": "baz"}},
)
returned_item = table.get_item(username="steve")
@ -616,8 +636,9 @@ def test_boto3_update_item_conditions_fail():
table.put_item(Item={"username": "johndoe", "foo": "baz"})
table.update_item.when.called_with(
Key={"username": "johndoe"},
UpdateExpression="SET foo=bar",
UpdateExpression="SET foo=:bar",
Expected={"foo": {"Value": "bar"}},
ExpressionAttributeValues={":bar": "bar"},
).should.throw(botocore.client.ClientError)
@ -627,8 +648,9 @@ def test_boto3_update_item_conditions_fail_because_expect_not_exists():
table.put_item(Item={"username": "johndoe", "foo": "baz"})
table.update_item.when.called_with(
Key={"username": "johndoe"},
UpdateExpression="SET foo=bar",
UpdateExpression="SET foo=:bar",
Expected={"foo": {"Exists": False}},
ExpressionAttributeValues={":bar": "bar"},
).should.throw(botocore.client.ClientError)
@ -638,8 +660,9 @@ def test_boto3_update_item_conditions_fail_because_expect_not_exists_by_compare_
table.put_item(Item={"username": "johndoe", "foo": "baz"})
table.update_item.when.called_with(
Key={"username": "johndoe"},
UpdateExpression="SET foo=bar",
UpdateExpression="SET foo=:bar",
Expected={"foo": {"ComparisonOperator": "NULL"}},
ExpressionAttributeValues={":bar": "bar"},
).should.throw(botocore.client.ClientError)
@ -649,8 +672,9 @@ def test_boto3_update_item_conditions_pass():
table.put_item(Item={"username": "johndoe", "foo": "bar"})
table.update_item(
Key={"username": "johndoe"},
UpdateExpression="SET foo=baz",
UpdateExpression="SET foo=:baz",
Expected={"foo": {"Value": "bar"}},
ExpressionAttributeValues={":baz": "baz"},
)
returned_item = table.get_item(Key={"username": "johndoe"})
assert dict(returned_item)["Item"]["foo"].should.equal("baz")
@ -662,8 +686,9 @@ def test_boto3_update_item_conditions_pass_because_expect_not_exists():
table.put_item(Item={"username": "johndoe", "foo": "bar"})
table.update_item(
Key={"username": "johndoe"},
UpdateExpression="SET foo=baz",
UpdateExpression="SET foo=:baz",
Expected={"whatever": {"Exists": False}},
ExpressionAttributeValues={":baz": "baz"},
)
returned_item = table.get_item(Key={"username": "johndoe"})
assert dict(returned_item)["Item"]["foo"].should.equal("baz")
@ -675,8 +700,9 @@ def test_boto3_update_item_conditions_pass_because_expect_not_exists_by_compare_
table.put_item(Item={"username": "johndoe", "foo": "bar"})
table.update_item(
Key={"username": "johndoe"},
UpdateExpression="SET foo=baz",
UpdateExpression="SET foo=:baz",
Expected={"whatever": {"ComparisonOperator": "NULL"}},
ExpressionAttributeValues={":baz": "baz"},
)
returned_item = table.get_item(Key={"username": "johndoe"})
assert dict(returned_item)["Item"]["foo"].should.equal("baz")
@ -688,8 +714,9 @@ def test_boto3_update_item_conditions_pass_because_expect_exists_by_compare_to_n
table.put_item(Item={"username": "johndoe", "foo": "bar"})
table.update_item(
Key={"username": "johndoe"},
UpdateExpression="SET foo=baz",
UpdateExpression="SET foo=:baz",
Expected={"foo": {"ComparisonOperator": "NOT_NULL"}},
ExpressionAttributeValues={":baz": "baz"},
)
returned_item = table.get_item(Key={"username": "johndoe"})
assert dict(returned_item)["Item"]["foo"].should.equal("baz")

464
tests/test_dynamodb2/test_dynamodb_validation.py Normal file
View file

@ -0,0 +1,464 @@
from moto.dynamodb2.exceptions import (
AttributeIsReservedKeyword,
ExpressionAttributeValueNotDefined,
AttributeDoesNotExist,
ExpressionAttributeNameNotDefined,
IncorrectOperandType,
InvalidUpdateExpressionInvalidDocumentPath,
)
from moto.dynamodb2.models import Item, DynamoType
from moto.dynamodb2.parsing.ast_nodes import (
NodeDepthLeftTypeFetcher,
UpdateExpressionSetAction,
UpdateExpressionValue,
DDBTypedValue,
)
from moto.dynamodb2.parsing.expressions import UpdateExpressionParser
from moto.dynamodb2.parsing.validators import UpdateExpressionValidator
from parameterized import parameterized
def test_validation_of_update_expression_with_keyword():
try:
update_expression = "SET myNum = path + :val"
update_expression_values = {":val": {"N": "3"}}
update_expression_ast = UpdateExpressionParser.make(update_expression)
item = Item(
hash_key=DynamoType({"S": "id"}),
hash_key_type="TYPE",
range_key=None,
range_key_type=None,
attrs={"id": {"S": "1"}, "path": {"N": "3"}},
)
UpdateExpressionValidator(
update_expression_ast,
expression_attribute_names=None,
expression_attribute_values=update_expression_values,
item=item,
).validate()
assert False, "No exception raised"
except AttributeIsReservedKeyword as e:
assert e.keyword == "path"
@parameterized(
["SET a = #b + :val2", "SET a = :val2 + #b",]
)
def test_validation_of_a_set_statement_with_incorrect_passed_value(update_expression):
"""
By running permutations it shows that values are replaced prior to resolving attributes.
An error occurred (ValidationException) when calling the UpdateItem operation: Invalid UpdateExpression:
An expression attribute value used in expression is not defined; attribute value: :val2
"""
update_expression_ast = UpdateExpressionParser.make(update_expression)
item = Item(
hash_key=DynamoType({"S": "id"}),
hash_key_type="TYPE",
range_key=None,
range_key_type=None,
attrs={"id": {"S": "1"}, "b": {"N": "3"}},
)
try:
UpdateExpressionValidator(
update_expression_ast,
expression_attribute_names={"#b": "ok"},
expression_attribute_values={":val": {"N": "3"}},
item=item,
).validate()
except ExpressionAttributeValueNotDefined as e:
assert e.attribute_value == ":val2"
def test_validation_of_update_expression_with_attribute_that_does_not_exist_in_item():
"""
When an update expression tries to get an attribute that does not exist it must throw the appropriate exception.
An error occurred (ValidationException) when calling the UpdateItem operation:
The provided expression refers to an attribute that does not exist in the item
"""
try:
update_expression = "SET a = nonexistent"
update_expression_ast = UpdateExpressionParser.make(update_expression)
item = Item(
hash_key=DynamoType({"S": "id"}),
hash_key_type="TYPE",
range_key=None,
range_key_type=None,
attrs={"id": {"S": "1"}, "path": {"N": "3"}},
)
UpdateExpressionValidator(
update_expression_ast,
expression_attribute_names=None,
expression_attribute_values=None,
item=item,
).validate()
assert False, "No exception raised"
except AttributeDoesNotExist:
assert True
@parameterized(
["SET a = #c", "SET a = #c + #d",]
)
def test_validation_of_update_expression_with_attribute_name_that_is_not_defined(
update_expression,
):
"""
When an update expression tries to get an attribute name that is not provided it must throw an exception.
An error occurred (ValidationException) when calling the UpdateItem operation: Invalid UpdateExpression:
An expression attribute name used in the document path is not defined; attribute name: #c
"""
try:
update_expression_ast = UpdateExpressionParser.make(update_expression)
item = Item(
hash_key=DynamoType({"S": "id"}),
hash_key_type="TYPE",
range_key=None,
range_key_type=None,
attrs={"id": {"S": "1"}, "path": {"N": "3"}},
)
UpdateExpressionValidator(
update_expression_ast,
expression_attribute_names={"#b": "ok"},
expression_attribute_values=None,
item=item,
).validate()
assert False, "No exception raised"
except ExpressionAttributeNameNotDefined as e:
assert e.not_defined_attribute_name == "#c"
def test_validation_of_if_not_exists_not_existing_invalid_replace_value():
try:
update_expression = "SET a = if_not_exists(b, a.c)"
update_expression_ast = UpdateExpressionParser.make(update_expression)
item = Item(
hash_key=DynamoType({"S": "id"}),
hash_key_type="TYPE",
range_key=None,
range_key_type=None,
attrs={"id": {"S": "1"}, "a": {"S": "A"}},
)
UpdateExpressionValidator(
update_expression_ast,
expression_attribute_names=None,
expression_attribute_values=None,
item=item,
).validate()
assert False, "No exception raised"
except AttributeDoesNotExist:
assert True
def get_first_node_of_type(ast, node_type):
return next(NodeDepthLeftTypeFetcher(node_type, ast))
def get_set_action_value(ast):
"""
Helper that takes an AST and gets the first UpdateExpressionSetAction and retrieves the value of that action.
This should only be called on validated expressions.
Args:
ast(Node):
Returns:
DynamoType: The DynamoType object representing the Dynamo value.
"""
set_action = get_first_node_of_type(ast, UpdateExpressionSetAction)
typed_value = set_action.children[1]
assert isinstance(typed_value, DDBTypedValue)
dynamo_value = typed_value.children[0]
assert isinstance(dynamo_value, DynamoType)
return dynamo_value
def test_validation_of_if_not_exists_not_existing_value():
update_expression = "SET a = if_not_exists(b, a)"
update_expression_ast = UpdateExpressionParser.make(update_expression)
item = Item(
hash_key=DynamoType({"S": "id"}),
hash_key_type="TYPE",
range_key=None,
range_key_type=None,
attrs={"id": {"S": "1"}, "a": {"S": "A"}},
)
validated_ast = UpdateExpressionValidator(
update_expression_ast,
expression_attribute_names=None,
expression_attribute_values=None,
item=item,
).validate()
dynamo_value = get_set_action_value(validated_ast)
assert dynamo_value == DynamoType({"S": "A"})
def test_validation_of_if_not_exists_with_existing_attribute_should_return_attribute():
update_expression = "SET a = if_not_exists(b, a)"
update_expression_ast = UpdateExpressionParser.make(update_expression)
item = Item(
hash_key=DynamoType({"S": "id"}),
hash_key_type="TYPE",
range_key=None,
range_key_type=None,
attrs={"id": {"S": "1"}, "a": {"S": "A"}, "b": {"S": "B"}},
)
validated_ast = UpdateExpressionValidator(
update_expression_ast,
expression_attribute_names=None,
expression_attribute_values=None,
item=item,
).validate()
dynamo_value = get_set_action_value(validated_ast)
assert dynamo_value == DynamoType({"S": "B"})
def test_validation_of_if_not_exists_with_existing_attribute_should_return_value():
update_expression = "SET a = if_not_exists(b, :val)"
update_expression_values = {":val": {"N": "4"}}
update_expression_ast = UpdateExpressionParser.make(update_expression)
item = Item(
hash_key=DynamoType({"S": "id"}),
hash_key_type="TYPE",
range_key=None,
range_key_type=None,
attrs={"id": {"S": "1"}, "b": {"N": "3"}},
)
validated_ast = UpdateExpressionValidator(
update_expression_ast,
expression_attribute_names=None,
expression_attribute_values=update_expression_values,
item=item,
).validate()
dynamo_value = get_set_action_value(validated_ast)
assert dynamo_value == DynamoType({"N": "3"})
def test_validation_of_if_not_exists_with_non_existing_attribute_should_return_value():
update_expression = "SET a = if_not_exists(b, :val)"
update_expression_values = {":val": {"N": "4"}}
update_expression_ast = UpdateExpressionParser.make(update_expression)
item = Item(
hash_key=DynamoType({"S": "id"}),
hash_key_type="TYPE",
range_key=None,
range_key_type=None,
attrs={"id": {"S": "1"}},
)
validated_ast = UpdateExpressionValidator(
update_expression_ast,
expression_attribute_names=None,
expression_attribute_values=update_expression_values,
item=item,
).validate()
dynamo_value = get_set_action_value(validated_ast)
assert dynamo_value == DynamoType({"N": "4"})
def test_validation_of_sum_operation():
update_expression = "SET a = a + b"
update_expression_ast = UpdateExpressionParser.make(update_expression)
item = Item(
hash_key=DynamoType({"S": "id"}),
hash_key_type="TYPE",
range_key=None,
range_key_type=None,
attrs={"id": {"S": "1"}, "a": {"N": "3"}, "b": {"N": "4"}},
)
validated_ast = UpdateExpressionValidator(
update_expression_ast,
expression_attribute_names=None,
expression_attribute_values=None,
item=item,
).validate()
dynamo_value = get_set_action_value(validated_ast)
assert dynamo_value == DynamoType({"N": "7"})
def test_validation_homogeneous_list_append_function():
update_expression = "SET ri = list_append(ri, :vals)"
update_expression_ast = UpdateExpressionParser.make(update_expression)
item = Item(
hash_key=DynamoType({"S": "id"}),
hash_key_type="TYPE",
range_key=None,
range_key_type=None,
attrs={"id": {"S": "1"}, "ri": {"L": [{"S": "i1"}, {"S": "i2"}]}},
)
validated_ast = UpdateExpressionValidator(
update_expression_ast,
expression_attribute_names=None,
expression_attribute_values={":vals": {"L": [{"S": "i3"}, {"S": "i4"}]}},
item=item,
).validate()
dynamo_value = get_set_action_value(validated_ast)
assert dynamo_value == DynamoType(
{"L": [{"S": "i1"}, {"S": "i2"}, {"S": "i3"}, {"S": "i4"}]}
)
def test_validation_hetereogenous_list_append_function():
update_expression = "SET ri = list_append(ri, :vals)"
update_expression_ast = UpdateExpressionParser.make(update_expression)
item = Item(
hash_key=DynamoType({"S": "id"}),
hash_key_type="TYPE",
range_key=None,
range_key_type=None,
attrs={"id": {"S": "1"}, "ri": {"L": [{"S": "i1"}, {"S": "i2"}]}},
)
validated_ast = UpdateExpressionValidator(
update_expression_ast,
expression_attribute_names=None,
expression_attribute_values={":vals": {"L": [{"N": "3"}]}},
item=item,
).validate()
dynamo_value = get_set_action_value(validated_ast)
assert dynamo_value == DynamoType({"L": [{"S": "i1"}, {"S": "i2"}, {"N": "3"}]})
def test_validation_list_append_function_with_non_list_arg():
"""
Must error out:
Invalid UpdateExpression: Incorrect operand type for operator or function;
operator or function: list_append, operand type: S'
Returns:
"""
try:
update_expression = "SET ri = list_append(ri, :vals)"
update_expression_ast = UpdateExpressionParser.make(update_expression)
item = Item(
hash_key=DynamoType({"S": "id"}),
hash_key_type="TYPE",
range_key=None,
range_key_type=None,
attrs={"id": {"S": "1"}, "ri": {"L": [{"S": "i1"}, {"S": "i2"}]}},
)
UpdateExpressionValidator(
update_expression_ast,
expression_attribute_names=None,
expression_attribute_values={":vals": {"S": "N"}},
item=item,
).validate()
except IncorrectOperandType as e:
assert e.operand_type == "S"
assert e.operator_or_function == "list_append"
def test_sum_with_incompatible_types():
"""
Must error out:
Invalid UpdateExpression: Incorrect operand type for operator or function; operator or function: +, operand type: S'
Returns:
"""
try:
update_expression = "SET ri = :val + :val2"
update_expression_ast = UpdateExpressionParser.make(update_expression)
item = Item(
hash_key=DynamoType({"S": "id"}),
hash_key_type="TYPE",
range_key=None,
range_key_type=None,
attrs={"id": {"S": "1"}, "ri": {"L": [{"S": "i1"}, {"S": "i2"}]}},
)
UpdateExpressionValidator(
update_expression_ast,
expression_attribute_names=None,
expression_attribute_values={":val": {"S": "N"}, ":val2": {"N": "3"}},
item=item,
).validate()
except IncorrectOperandType as e:
assert e.operand_type == "S"
assert e.operator_or_function == "+"
def test_validation_of_subraction_operation():
update_expression = "SET ri = :val - :val2"
update_expression_ast = UpdateExpressionParser.make(update_expression)
item = Item(
hash_key=DynamoType({"S": "id"}),
hash_key_type="TYPE",
range_key=None,
range_key_type=None,
attrs={"id": {"S": "1"}, "a": {"N": "3"}, "b": {"N": "4"}},
)
validated_ast = UpdateExpressionValidator(
update_expression_ast,
expression_attribute_names=None,
expression_attribute_values={":val": {"N": "1"}, ":val2": {"N": "3"}},
item=item,
).validate()
dynamo_value = get_set_action_value(validated_ast)
assert dynamo_value == DynamoType({"N": "-2"})
def test_cannot_index_into_a_string():
"""
Must error out:
The document path provided in the update expression is invalid for update'
"""
try:
update_expression = "set itemstr[1]=:Item"
update_expression_ast = UpdateExpressionParser.make(update_expression)
item = Item(
hash_key=DynamoType({"S": "id"}),
hash_key_type="TYPE",
range_key=None,
range_key_type=None,
attrs={"id": {"S": "foo2"}, "itemstr": {"S": "somestring"}},
)
UpdateExpressionValidator(
update_expression_ast,
expression_attribute_names=None,
expression_attribute_values={":Item": {"S": "string_update"}},
item=item,
).validate()
assert False, "Must raise exception"
except InvalidUpdateExpressionInvalidDocumentPath:
assert True
def test_validation_set_path_does_not_need_to_be_resolvable_when_setting_a_new_attribute():
"""If this step just passes we are happy enough"""
update_expression = "set d=a"
update_expression_ast = UpdateExpressionParser.make(update_expression)
item = Item(
hash_key=DynamoType({"S": "id"}),
hash_key_type="TYPE",
range_key=None,
range_key_type=None,
attrs={"id": {"S": "foo2"}, "a": {"N": "3"}},
)
validated_ast = UpdateExpressionValidator(
update_expression_ast,
expression_attribute_names=None,
expression_attribute_values=None,
item=item,
).validate()
dynamo_value = get_set_action_value(validated_ast)
assert dynamo_value == DynamoType({"N": "3"})
def test_validation_set_path_does_not_need_to_be_resolvable_but_must_be_creatable_when_setting_a_new_attribute():
try:
update_expression = "set d.e=a"
update_expression_ast = UpdateExpressionParser.make(update_expression)
item = Item(
hash_key=DynamoType({"S": "id"}),
hash_key_type="TYPE",
range_key=None,
range_key_type=None,
attrs={"id": {"S": "foo2"}, "a": {"N": "3"}},
)
UpdateExpressionValidator(
update_expression_ast,
expression_attribute_names=None,
expression_attribute_values=None,
item=item,
).validate()
assert False, "Must raise exception"
except InvalidUpdateExpressionInvalidDocumentPath:
assert True

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"}})

130
tests/test_eb/test_eb.py Normal file
View file

@ -0,0 +1,130 @@
import boto3
import sure # noqa
from botocore.exceptions import ClientError
from moto import mock_elasticbeanstalk
@mock_elasticbeanstalk
def test_create_application():
# Create Elastic Beanstalk Application
conn = boto3.client("elasticbeanstalk", region_name="us-east-1")
app = conn.create_application(ApplicationName="myapp",)
app["Application"]["ApplicationName"].should.equal("myapp")
@mock_elasticbeanstalk
def test_create_application_dup():
conn = boto3.client("elasticbeanstalk", region_name="us-east-1")
conn.create_application(ApplicationName="myapp",)
conn.create_application.when.called_with(ApplicationName="myapp",).should.throw(
ClientError
)
@mock_elasticbeanstalk
def test_describe_applications():
# Create Elastic Beanstalk Application
conn = boto3.client("elasticbeanstalk", region_name="us-east-1")
conn.create_application(ApplicationName="myapp",)
apps = conn.describe_applications()
len(apps["Applications"]).should.equal(1)
apps["Applications"][0]["ApplicationName"].should.equal("myapp")
@mock_elasticbeanstalk
def test_create_environment():
# Create Elastic Beanstalk Environment
conn = boto3.client("elasticbeanstalk", region_name="us-east-1")
app = conn.create_application(ApplicationName="myapp",)
env = conn.create_environment(ApplicationName="myapp", EnvironmentName="myenv",)
env["EnvironmentName"].should.equal("myenv")
@mock_elasticbeanstalk
def test_describe_environments():
# List Elastic Beanstalk Envs
conn = boto3.client("elasticbeanstalk", region_name="us-east-1")
conn.create_application(ApplicationName="myapp",)
conn.create_environment(
ApplicationName="myapp", EnvironmentName="myenv",
)
envs = conn.describe_environments()
envs = envs["Environments"]
len(envs).should.equal(1)
envs[0]["ApplicationName"].should.equal("myapp")
envs[0]["EnvironmentName"].should.equal("myenv")
def tags_dict_to_list(tag_dict):
tag_list = []
for key, value in tag_dict.items():
tag_list.append({"Key": key, "Value": value})
return tag_list
def tags_list_to_dict(tag_list):
tag_dict = {}
for tag in tag_list:
tag_dict[tag["Key"]] = tag["Value"]
return tag_dict
@mock_elasticbeanstalk
def test_create_environment_tags():
conn = boto3.client("elasticbeanstalk", region_name="us-east-1")
conn.create_application(ApplicationName="myapp",)
env_tags = {"initial key": "initial value"}
env = conn.create_environment(
ApplicationName="myapp",
EnvironmentName="myenv",
Tags=tags_dict_to_list(env_tags),
)
tags = conn.list_tags_for_resource(ResourceArn=env["EnvironmentArn"],)
tags["ResourceArn"].should.equal(env["EnvironmentArn"])
tags_list_to_dict(tags["ResourceTags"]).should.equal(env_tags)
@mock_elasticbeanstalk
def test_update_tags():
conn = boto3.client("elasticbeanstalk", region_name="us-east-1")
conn.create_application(ApplicationName="myapp",)
env_tags = {
"initial key": "initial value",
"to remove": "delete me",
"to update": "original",
}
env = conn.create_environment(
ApplicationName="myapp",
EnvironmentName="myenv",
Tags=tags_dict_to_list(env_tags),
)
extra_env_tags = {
"to update": "new",
"extra key": "extra value",
}
conn.update_tags_for_resource(
ResourceArn=env["EnvironmentArn"],
TagsToAdd=tags_dict_to_list(extra_env_tags),
TagsToRemove=["to remove"],
)
total_env_tags = env_tags.copy()
total_env_tags.update(extra_env_tags)
del total_env_tags["to remove"]
tags = conn.list_tags_for_resource(ResourceArn=env["EnvironmentArn"],)
tags["ResourceArn"].should.equal(env["EnvironmentArn"])
tags_list_to_dict(tags["ResourceTags"]).should.equal(total_env_tags)
@mock_elasticbeanstalk
def test_list_available_solution_stacks():
conn = boto3.client("elasticbeanstalk", region_name="us-east-1")
stacks = conn.list_available_solution_stacks()
len(stacks["SolutionStacks"]).should.be.greater_than(0)
len(stacks["SolutionStacks"]).should.be.equal(len(stacks["SolutionStackDetails"]))

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")

63
tests/test_ec2/test_instances.py
View file

@ -9,6 +9,7 @@ from nose.tools import assert_raises
import base64
import datetime
import ipaddress
import json
import six
import boto
@ -18,7 +19,7 @@ from boto.exception import EC2ResponseError, EC2ResponseError
from freezegun import freeze_time
import sure # noqa
from moto import mock_ec2_deprecated, mock_ec2
from moto import mock_ec2_deprecated, mock_ec2, mock_cloudformation
from tests.helpers import requires_boto_gte
@ -71,7 +72,7 @@ def test_instance_launch_and_terminate():
instance.id.should.equal(instance.id)
instance.state.should.equal("running")
instance.launch_time.should.equal("2014-01-01T05:00:00.000Z")
instance.vpc_id.should.equal(None)
instance.vpc_id.shouldnt.equal(None)
instance.placement.should.equal("us-east-1a")
root_device_name = instance.root_device_name
@ -1166,6 +1167,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")
@ -1319,6 +1335,12 @@ def test_create_instance_ebs_optimized():
instance.load()
instance.ebs_optimized.should.be(False)
instance = ec2_resource.create_instances(
ImageId="ami-12345678", MaxCount=1, MinCount=1,
)[0]
instance.load()
instance.ebs_optimized.should.be(False)
@mock_ec2
def test_run_multiple_instances_in_same_command():
@ -1399,3 +1421,40 @@ def test_describe_instance_attribute():
invalid_instance_attribute=invalid_instance_attribute
)
ex.exception.response["Error"]["Message"].should.equal(message)
@mock_ec2
@mock_cloudformation
def test_volume_size_through_cloudformation():
ec2 = boto3.client("ec2", region_name="us-east-1")
cf = boto3.client("cloudformation", region_name="us-east-1")
volume_template = {
"AWSTemplateFormatVersion": "2010-09-09",
"Resources": {
"testInstance": {
"Type": "AWS::EC2::Instance",
"Properties": {
"ImageId": "ami-d3adb33f",
"KeyName": "dummy",
"InstanceType": "t2.micro",
"BlockDeviceMappings": [
{"DeviceName": "/dev/sda2", "Ebs": {"VolumeSize": "50"}}
],
"Tags": [
{"Key": "foo", "Value": "bar"},
{"Key": "blah", "Value": "baz"},
],
},
}
},
}
template_json = json.dumps(volume_template)
cf.create_stack(StackName="test_stack", TemplateBody=template_json)
instances = ec2.describe_instances()
volume = instances["Reservations"][0]["Instances"][0]["BlockDeviceMappings"][0][
"Ebs"
]
volumes = ec2.describe_volumes(VolumeIds=[volume["VolumeId"]])
volumes["Volumes"][0]["Size"].should.equal(50)

17
tests/test_ec2/test_subnets.py
View file

@ -599,3 +599,20 @@ def validate_subnet_details_after_creating_eni(
for eni in enis_created:
client.delete_network_interface(NetworkInterfaceId=eni["NetworkInterfaceId"])
client.delete_subnet(SubnetId=subnet["SubnetId"])
@mock_ec2
def test_run_instances_should_attach_to_default_subnet():
# https://github.com/spulec/moto/issues/2877
ec2 = boto3.resource("ec2", region_name="us-west-1")
client = boto3.client("ec2", region_name="us-west-1")
ec2.create_security_group(GroupName="sg01", Description="Test security group sg01")
# run_instances
instances = client.run_instances(MinCount=1, MaxCount=1, SecurityGroups=["sg01"],)
# Assert subnet is created appropriately
subnets = client.describe_subnets()["Subnets"]
default_subnet_id = subnets[0]["SubnetId"]
instances["Instances"][0]["NetworkInterfaces"][0]["SubnetId"].should.equal(
default_subnet_id
)
subnets[0]["AvailableIpAddressCount"].should.equal(4090)

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():

16
tests/test_events/test_events.py
View file

@ -79,13 +79,23 @@ def generate_environment():
@mock_events
def test_put_rule():
client = boto3.client("events", "us-west-2")
client.list_rules()["Rules"].should.have.length_of(0)
rule_data = get_random_rule()
rule_data = {
"Name": "my-event",
"ScheduleExpression": "rate(5 minutes)",
"EventPattern": '{"source": ["test-source"]}',
}
client.put_rule(**rule_data)
client.list_rules()["Rules"].should.have.length_of(1)
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

23
tests/test_glue/test_datacatalog.py
View file

@ -52,6 +52,29 @@ def test_get_database_not_exits():
)
@mock_glue
def test_get_databases_empty():
client = boto3.client("glue", region_name="us-east-1")
response = client.get_databases()
response["DatabaseList"].should.have.length_of(0)
@mock_glue
def test_get_databases_several_items():
client = boto3.client("glue", region_name="us-east-1")
database_name_1, database_name_2 = "firstdatabase", "seconddatabase"
helpers.create_database(client, database_name_1)
helpers.create_database(client, database_name_2)
database_list = sorted(
client.get_databases()["DatabaseList"], key=lambda x: x["Name"]
)
database_list.should.have.length_of(2)
database_list[0].should.equal({"Name": database_name_1})
database_list[1].should.equal({"Name": database_name_2})
@mock_glue
def test_create_table():
client = boto3.client("glue", region_name="us-east-1")

8
tests/test_iot/test_iot.py
View file

@ -728,6 +728,14 @@ def test_principal_thing():
res = client.list_thing_principals(thingName=thing_name)
res.should.have.key("principals").which.should.have.length_of(0)
with assert_raises(ClientError) as e:
client.list_thing_principals(thingName="xxx")
e.exception.response["Error"]["Code"].should.equal("ResourceNotFoundException")
e.exception.response["Error"]["Message"].should.equal(
"Failed to list principals for thing xxx because the thing does not exist in your account"
)
@mock_iot
def test_delete_principal_thing():

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

31
tests/test_rds2/test_rds2.py
View file

@ -183,12 +183,12 @@ def test_start_database():
@mock_rds2
def test_fail_to_stop_multi_az():
def test_fail_to_stop_multi_az_and_sqlserver():
conn = boto3.client("rds", region_name="us-west-2")
database = conn.create_db_instance(
DBInstanceIdentifier="db-master-1",
AllocatedStorage=10,
Engine="postgres",
Engine="sqlserver-ee",
DBName="staging-postgres",
DBInstanceClass="db.m1.small",
LicenseModel="license-included",
@ -213,6 +213,33 @@ def test_fail_to_stop_multi_az():
).should.throw(ClientError)
@mock_rds2
def test_stop_multi_az_postgres():
conn = boto3.client("rds", region_name="us-west-2")
database = conn.create_db_instance(
DBInstanceIdentifier="db-master-1",
AllocatedStorage=10,
Engine="postgres",
DBName="staging-postgres",
DBInstanceClass="db.m1.small",
LicenseModel="license-included",
MasterUsername="root",
MasterUserPassword="hunter2",
Port=1234,
DBSecurityGroups=["my_sg"],
MultiAZ=True,
)
mydb = conn.describe_db_instances(
DBInstanceIdentifier=database["DBInstance"]["DBInstanceIdentifier"]
)["DBInstances"][0]
mydb["DBInstanceStatus"].should.equal("available")
response = conn.stop_db_instance(DBInstanceIdentifier=mydb["DBInstanceIdentifier"])
response["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
response["DBInstance"]["DBInstanceStatus"].should.equal("stopped")
@mock_rds2
def test_fail_to_stop_readreplica():
conn = boto3.client("rds", region_name="us-west-2")

81
tests/test_s3/test_s3.py
View file

@ -14,6 +14,7 @@ from io import BytesIO
import mimetypes
import zlib
import pickle
import uuid
import json
import boto
@ -3255,7 +3256,8 @@ def test_boto3_put_object_tagging_on_earliest_version():
# Older version has tags while the most recent does not
resp = s3.get_object_tagging(Bucket=bucket_name, Key=key, VersionId=first_object.id)
resp["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
resp["TagSet"].should.equal(
sorted_tagset = sorted(resp["TagSet"], key=lambda t: t["Key"])
sorted_tagset.should.equal(
[{"Key": "item1", "Value": "foo"}, {"Key": "item2", "Value": "bar"}]
)
@ -3333,7 +3335,8 @@ def test_boto3_put_object_tagging_on_both_version():
resp = s3.get_object_tagging(Bucket=bucket_name, Key=key, VersionId=first_object.id)
resp["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
resp["TagSet"].should.equal(
sorted_tagset = sorted(resp["TagSet"], key=lambda t: t["Key"])
sorted_tagset.should.equal(
[{"Key": "item1", "Value": "foo"}, {"Key": "item2", "Value": "bar"}]
)
@ -3341,7 +3344,8 @@ def test_boto3_put_object_tagging_on_both_version():
Bucket=bucket_name, Key=key, VersionId=second_object.id
)
resp["ResponseMetadata"]["HTTPStatusCode"].should.equal(200)
resp["TagSet"].should.equal(
sorted_tagset = sorted(resp["TagSet"], key=lambda t: t["Key"])
sorted_tagset.should.equal(
[{"Key": "item1", "Value": "baz"}, {"Key": "item2", "Value": "bin"}]
)
@ -3743,6 +3747,28 @@ def test_root_dir_with_empty_name_works():
store_and_read_back_a_key("/")
@parameterized(["mybucket", "my.bucket"])
@mock_s3
def test_leading_slashes_not_removed(bucket_name):
"""Make sure that leading slashes are not removed internally."""
s3 = boto3.client("s3", region_name=DEFAULT_REGION_NAME)
s3.create_bucket(Bucket=bucket_name)
uploaded_key = "/key"
invalid_key_1 = "key"
invalid_key_2 = "//key"
s3.put_object(Bucket=bucket_name, Key=uploaded_key, Body=b"Some body")
with assert_raises(ClientError) as e:
s3.get_object(Bucket=bucket_name, Key=invalid_key_1)
e.exception.response["Error"]["Code"].should.equal("NoSuchKey")
with assert_raises(ClientError) as e:
s3.get_object(Bucket=bucket_name, Key=invalid_key_2)
e.exception.response["Error"]["Code"].should.equal("NoSuchKey")
@parameterized(
[("foo/bar/baz",), ("foo",), ("foo/run_dt%3D2019-01-01%252012%253A30%253A00",)]
)
@ -4292,24 +4318,17 @@ def test_s3_config_dict():
FakeAcl,
FakeGrant,
FakeGrantee,
FakeTag,
FakeTagging,
FakeTagSet,
OWNER,
)
# Without any buckets:
assert not s3_config_query.get_config_resource("some_bucket")
tags = FakeTagging(
FakeTagSet(
[FakeTag("someTag", "someValue"), FakeTag("someOtherTag", "someOtherValue")]
)
)
tags = {"someTag": "someValue", "someOtherTag": "someOtherValue"}
# With 1 bucket in us-west-2:
s3_config_query.backends["global"].create_bucket("bucket1", "us-west-2")
s3_config_query.backends["global"].put_bucket_tagging("bucket1", tags)
s3_config_query.backends["global"].put_bucket_tags("bucket1", tags)
# With a log bucket:
s3_config_query.backends["global"].create_bucket("logbucket", "us-west-2")
@ -4428,3 +4447,41 @@ def test_s3_config_dict():
assert not logging_bucket["supplementaryConfiguration"].get(
"BucketTaggingConfiguration"
)
@mock_s3
def test_creating_presigned_post():
bucket = "presigned-test"
s3 = boto3.client("s3", region_name="us-east-1")
s3.create_bucket(Bucket=bucket)
success_url = "http://localhost/completed"
fdata = b"test data\n"
file_uid = uuid.uuid4()
conditions = [
{"Content-Type": "text/plain"},
{"x-amz-server-side-encryption": "AES256"},
{"success_action_redirect": success_url},
]
conditions.append(["content-length-range", 1, 30])
data = s3.generate_presigned_post(
Bucket=bucket,
Key="{file_uid}.txt".format(file_uid=file_uid),
Fields={
"content-type": "text/plain",
"success_action_redirect": success_url,
"x-amz-server-side-encryption": "AES256",
},
Conditions=conditions,
ExpiresIn=1000,
)
resp = requests.post(
data["url"], data=data["fields"], files={"file": fdata}, allow_redirects=False
)
assert resp.headers["Location"] == success_url
assert resp.status_code == 303
assert (
s3.get_object(Bucket=bucket, Key="{file_uid}.txt".format(file_uid=file_uid))[
"Body"
].read()
== fdata
)

76
tests/test_secretsmanager/test_secretsmanager.py
View file

@ -711,3 +711,79 @@ def test_can_list_secret_version_ids():
returned_version_ids = [v["VersionId"] for v in versions_list["Versions"]]
assert [first_version_id, second_version_id].sort() == returned_version_ids.sort()
@mock_secretsmanager
def test_update_secret():
conn = boto3.client("secretsmanager", region_name="us-west-2")
created_secret = conn.create_secret(Name="test-secret", SecretString="foosecret")
assert created_secret["ARN"]
assert created_secret["Name"] == "test-secret"
assert created_secret["VersionId"] != ""
secret = conn.get_secret_value(SecretId="test-secret")
assert secret["SecretString"] == "foosecret"
updated_secret = conn.update_secret(
SecretId="test-secret", SecretString="barsecret"
)
assert updated_secret["ARN"]
assert updated_secret["Name"] == "test-secret"
assert updated_secret["VersionId"] != ""
secret = conn.get_secret_value(SecretId="test-secret")
assert secret["SecretString"] == "barsecret"
assert created_secret["VersionId"] != updated_secret["VersionId"]
@mock_secretsmanager
def test_update_secret_which_does_not_exit():
conn = boto3.client("secretsmanager", region_name="us-west-2")
with assert_raises(ClientError) as cm:
updated_secret = conn.update_secret(
SecretId="test-secret", SecretString="barsecret"
)
assert_equal(
"Secrets Manager can't find the specified secret.",
cm.exception.response["Error"]["Message"],
)
@mock_secretsmanager
def test_update_secret_marked_as_deleted():
conn = boto3.client("secretsmanager", region_name="us-west-2")
created_secret = conn.create_secret(Name="test-secret", SecretString="foosecret")
deleted_secret = conn.delete_secret(SecretId="test-secret")
with assert_raises(ClientError) as cm:
updated_secret = conn.update_secret(
SecretId="test-secret", SecretString="barsecret"
)
assert (
"because it was marked for deletion."
in cm.exception.response["Error"]["Message"]
)
@mock_secretsmanager
def test_update_secret_marked_as_deleted_after_restoring():
conn = boto3.client("secretsmanager", region_name="us-west-2")
created_secret = conn.create_secret(Name="test-secret", SecretString="foosecret")
deleted_secret = conn.delete_secret(SecretId="test-secret")
restored_secret = conn.restore_secret(SecretId="test-secret")
updated_secret = conn.update_secret(
SecretId="test-secret", SecretString="barsecret"
)
assert updated_secret["ARN"]
assert updated_secret["Name"] == "test-secret"
assert updated_secret["VersionId"] != ""

31
tests/test_utilities/test_tagging_service.py
View file

@ -77,3 +77,34 @@ def test_extract_tag_names():
expected = ["key1", "key2"]
expected.should.be.equal(actual)
def test_copy_non_existing_arn():
svc = TaggingService()
tags = [{"Key": "key1", "Value": "value1"}, {"Key": "key2", "Value": "value2"}]
svc.tag_resource("new_arn", tags)
#
svc.copy_tags("non_existing_arn", "new_arn")
# Copying from a non-existing ARN should a NOOP
# Assert the old tags still exist
actual = sorted(
svc.list_tags_for_resource("new_arn")["Tags"], key=lambda t: t["Key"]
)
actual.should.equal(tags)
def test_copy_existing_arn():
svc = TaggingService()
tags_old_arn = [{"Key": "key1", "Value": "value1"}]
tags_new_arn = [{"Key": "key2", "Value": "value2"}]
svc.tag_resource("old_arn", tags_old_arn)
svc.tag_resource("new_arn", tags_new_arn)
#
svc.copy_tags("old_arn", "new_arn")
# Assert the old tags still exist
actual = sorted(
svc.list_tags_for_resource("new_arn")["Tags"], key=lambda t: t["Key"]
)
actual.should.equal(
[{"Key": "key1", "Value": "value1"}, {"Key": "key2", "Value": "value2"}]
)