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

Merge branch 'master' into feature/2546

Browse source
This commit is contained in:
Bert Blommers 2019-11-19 08:00:59 +00:00
commit 0ea98f22ee
44 changed files with 2383 additions and 218 deletions
Download patch file Download diff file Expand all files Collapse all files

10
tests/test_awslambda/test_lambda.py
View file

@ -164,7 +164,7 @@ if settings.TEST_SERVER_MODE:
conn = boto3.client("lambda", "us-west-2")
conn.create_function(
FunctionName="testFunction",
Runtime="python2.7",
Runtime="python3.7",
Role=get_role_name(),
Handler="lambda_function.lambda_handler",
Code={"ZipFile": get_test_zip_file2()},
@ -186,18 +186,20 @@ if settings.TEST_SERVER_MODE:
vol.id,
vol.state,
vol.size,
json.dumps(in_data),
json.dumps(in_data).replace(
" ", ""
), # Makes the tests pass as the result is missing the whitespace
)
log_result = base64.b64decode(result["LogResult"]).decode("utf-8")
# fix for running under travis (TODO: investigate why it has an extra newline)
# The Docker lambda invocation will return an additional '\n', so need to replace it:
log_result = log_result.replace("\n\n", "\n")
log_result.should.equal(msg)
payload = result["Payload"].read().decode("utf-8")
# fix for running under travis (TODO: investigate why it has an extra newline)
# The Docker lambda invocation will return an additional '\n', so need to replace it:
payload = payload.replace("\n\n", "\n")
payload.should.equal(msg)

23
tests/test_core/test_auth.py
View file

@ -11,6 +11,7 @@ from nose.tools import assert_raises
from moto import mock_iam, mock_ec2, mock_s3, mock_sts, mock_elbv2, mock_rds2
from moto.core import set_initial_no_auth_action_count
from moto.iam.models import ACCOUNT_ID
from uuid import uuid4
@mock_iam
@ -71,8 +72,10 @@ def create_user_with_access_key_and_multiple_policies(
def create_group_with_attached_policy_and_add_user(
user_name, policy_document, group_name="test-group", policy_name="policy1"
user_name, policy_document, group_name="test-group", policy_name=None
):
if not policy_name:
policy_name = str(uuid4())
client = boto3.client("iam", region_name="us-east-1")
client.create_group(GroupName=group_name)
policy_arn = client.create_policy(
@ -101,8 +104,10 @@ def create_group_with_multiple_policies_and_add_user(
attached_policy_document,
group_name="test-group",
inline_policy_name="policy1",
attached_policy_name="policy1",
attached_policy_name=None,
):
if not attached_policy_name:
attached_policy_name = str(uuid4())
client = boto3.client("iam", region_name="us-east-1")
client.create_group(GroupName=group_name)
client.put_group_policy(
@ -402,10 +407,10 @@ def test_s3_access_denied_with_denying_attached_group_policy():
"Statement": [{"Effect": "Deny", "Action": "s3:List*", "Resource": "*"}],
}
access_key = create_user_with_access_key_and_attached_policy(
user_name, attached_policy_document
user_name, attached_policy_document, policy_name="policy1"
)
create_group_with_attached_policy_and_add_user(
user_name, group_attached_policy_document
user_name, group_attached_policy_document, policy_name="policy2"
)
client = boto3.client(
"s3",
@ -476,10 +481,16 @@ def test_access_denied_with_many_irrelevant_policies():
"Statement": [{"Effect": "Deny", "Action": "lambda:*", "Resource": "*"}],
}
access_key = create_user_with_access_key_and_multiple_policies(
user_name, inline_policy_document, attached_policy_document
user_name,
inline_policy_document,
attached_policy_document,
attached_policy_name="policy1",
)
create_group_with_multiple_policies_and_add_user(
user_name, group_inline_policy_document, group_attached_policy_document
user_name,
group_inline_policy_document,
group_attached_policy_document,
attached_policy_name="policy2",
)
client = boto3.client(
"ec2",

98
tests/test_datasync/test_datasync.py
View file

@ -127,6 +127,22 @@ def test_list_locations():
assert response["Locations"][2]["LocationUri"] == "s3://my_bucket/dir"
@mock_datasync
def test_delete_location():
client = boto3.client("datasync", region_name="us-east-1")
locations = create_locations(client, create_smb=True)
response = client.list_locations()
assert len(response["Locations"]) == 1
location_arn = locations["smb_arn"]
response = client.delete_location(LocationArn=location_arn)
response = client.list_locations()
assert len(response["Locations"]) == 0
with assert_raises(ClientError) as e:
response = client.delete_location(LocationArn=location_arn)
@mock_datasync
def test_create_task():
client = boto3.client("datasync", region_name="us-east-1")
@ -208,6 +224,72 @@ def test_describe_task_not_exist():
client.describe_task(TaskArn="abc")
@mock_datasync
def test_update_task():
client = boto3.client("datasync", region_name="us-east-1")
locations = create_locations(client, create_s3=True, create_smb=True)
initial_name = "Initial_Name"
updated_name = "Updated_Name"
initial_options = {
"VerifyMode": "NONE",
"Atime": "BEST_EFFORT",
"Mtime": "PRESERVE",
}
updated_options = {
"VerifyMode": "POINT_IN_TIME_CONSISTENT",
"Atime": "BEST_EFFORT",
"Mtime": "PRESERVE",
}
response = client.create_task(
SourceLocationArn=locations["smb_arn"],
DestinationLocationArn=locations["s3_arn"],
Name=initial_name,
Options=initial_options,
)
task_arn = response["TaskArn"]
response = client.describe_task(TaskArn=task_arn)
assert response["TaskArn"] == task_arn
assert response["Name"] == initial_name
assert response["Options"] == initial_options
response = client.update_task(
TaskArn=task_arn, Name=updated_name, Options=updated_options
)
response = client.describe_task(TaskArn=task_arn)
assert response["TaskArn"] == task_arn
assert response["Name"] == updated_name
assert response["Options"] == updated_options
with assert_raises(ClientError) as e:
client.update_task(TaskArn="doesnt_exist")
@mock_datasync
def test_delete_task():
client = boto3.client("datasync", region_name="us-east-1")
locations = create_locations(client, create_s3=True, create_smb=True)
response = client.create_task(
SourceLocationArn=locations["smb_arn"],
DestinationLocationArn=locations["s3_arn"],
Name="task_name",
)
response = client.list_tasks()
assert len(response["Tasks"]) == 1
task_arn = response["Tasks"][0]["TaskArn"]
assert task_arn is not None
response = client.delete_task(TaskArn=task_arn)
response = client.list_tasks()
assert len(response["Tasks"]) == 0
with assert_raises(ClientError) as e:
response = client.delete_task(TaskArn=task_arn)
@mock_datasync
def test_start_task_execution():
client = boto3.client("datasync", region_name="us-east-1")
@ -261,6 +343,8 @@ def test_describe_task_execution():
Name="task_name",
)
task_arn = response["TaskArn"]
response = client.describe_task(TaskArn=task_arn)
assert response["Status"] == "AVAILABLE"
response = client.start_task_execution(TaskArn=task_arn)
task_execution_arn = response["TaskExecutionArn"]
@ -270,26 +354,38 @@ def test_describe_task_execution():
response = client.describe_task_execution(TaskExecutionArn=task_execution_arn)
assert response["TaskExecutionArn"] == task_execution_arn
assert response["Status"] == "INITIALIZING"
response = client.describe_task(TaskArn=task_arn)
assert response["Status"] == "RUNNING"
response = client.describe_task_execution(TaskExecutionArn=task_execution_arn)
assert response["TaskExecutionArn"] == task_execution_arn
assert response["Status"] == "PREPARING"
response = client.describe_task(TaskArn=task_arn)
assert response["Status"] == "RUNNING"
response = client.describe_task_execution(TaskExecutionArn=task_execution_arn)
assert response["TaskExecutionArn"] == task_execution_arn
assert response["Status"] == "TRANSFERRING"
response = client.describe_task(TaskArn=task_arn)
assert response["Status"] == "RUNNING"
response = client.describe_task_execution(TaskExecutionArn=task_execution_arn)
assert response["TaskExecutionArn"] == task_execution_arn
assert response["Status"] == "VERIFYING"
response = client.describe_task(TaskArn=task_arn)
assert response["Status"] == "RUNNING"
response = client.describe_task_execution(TaskExecutionArn=task_execution_arn)
assert response["TaskExecutionArn"] == task_execution_arn
assert response["Status"] == "SUCCESS"
response = client.describe_task(TaskArn=task_arn)
assert response["Status"] == "AVAILABLE"
response = client.describe_task_execution(TaskExecutionArn=task_execution_arn)
assert response["TaskExecutionArn"] == task_execution_arn
assert response["Status"] == "SUCCESS"
response = client.describe_task(TaskArn=task_arn)
assert response["Status"] == "AVAILABLE"
@mock_datasync
@ -317,11 +413,13 @@ def test_cancel_task_execution():
response = client.describe_task(TaskArn=task_arn)
assert response["CurrentTaskExecutionArn"] == task_execution_arn
assert response["Status"] == "RUNNING"
response = client.cancel_task_execution(TaskExecutionArn=task_execution_arn)
response = client.describe_task(TaskArn=task_arn)
assert "CurrentTaskExecutionArn" not in response
assert response["Status"] == "AVAILABLE"
response = client.describe_task_execution(TaskExecutionArn=task_execution_arn)
assert response["Status"] == "ERROR"

63
tests/test_dynamodb2/test_dynamodb.py
View file

@ -3319,3 +3319,66 @@ def _create_user_table():
TableName="users", Item={"username": {"S": "user3"}, "foo": {"S": "bar"}}
)
return client
@mock_dynamodb2
def test_update_item_if_original_value_is_none():
dynamo = boto3.resource("dynamodb", region_name="eu-central-1")
dynamo.create_table(
AttributeDefinitions=[{"AttributeName": "job_id", "AttributeType": "S"}],
TableName="origin-rbu-dev",
KeySchema=[{"AttributeName": "job_id", "KeyType": "HASH"}],
ProvisionedThroughput={"ReadCapacityUnits": 1, "WriteCapacityUnits": 1},
)
table = dynamo.Table("origin-rbu-dev")
table.put_item(Item={"job_id": "a", "job_name": None})
table.update_item(
Key={"job_id": "a"},
UpdateExpression="SET job_name = :output",
ExpressionAttributeValues={":output": "updated",},
)
table.scan()["Items"][0]["job_name"].should.equal("updated")
@mock_dynamodb2
def test_update_nested_item_if_original_value_is_none():
dynamo = boto3.resource("dynamodb", region_name="eu-central-1")
dynamo.create_table(
AttributeDefinitions=[{"AttributeName": "job_id", "AttributeType": "S"}],
TableName="origin-rbu-dev",
KeySchema=[{"AttributeName": "job_id", "KeyType": "HASH"}],
ProvisionedThroughput={"ReadCapacityUnits": 1, "WriteCapacityUnits": 1},
)
table = dynamo.Table("origin-rbu-dev")
table.put_item(Item={"job_id": "a", "job_details": {"job_name": None}})
table.update_item(
Key={"job_id": "a"},
UpdateExpression="SET job_details.job_name = :output",
ExpressionAttributeValues={":output": "updated",},
)
table.scan()["Items"][0]["job_details"]["job_name"].should.equal("updated")
@mock_dynamodb2
def test_allow_update_to_item_with_different_type():
dynamo = boto3.resource("dynamodb", region_name="eu-central-1")
dynamo.create_table(
AttributeDefinitions=[{"AttributeName": "job_id", "AttributeType": "S"}],
TableName="origin-rbu-dev",
KeySchema=[{"AttributeName": "job_id", "KeyType": "HASH"}],
ProvisionedThroughput={"ReadCapacityUnits": 1, "WriteCapacityUnits": 1},
)
table = dynamo.Table("origin-rbu-dev")
table.put_item(Item={"job_id": "a", "job_details": {"job_name": {"nested": "yes"}}})
table.put_item(Item={"job_id": "b", "job_details": {"job_name": {"nested": "yes"}}})
table.update_item(
Key={"job_id": "a"},
UpdateExpression="SET job_details.job_name = :output",
ExpressionAttributeValues={":output": "updated"},
)
table.get_item(Key={"job_id": "a"})["Item"]["job_details"][
"job_name"
].should.be.equal("updated")
table.get_item(Key={"job_id": "b"})["Item"]["job_details"][
"job_name"
].should.be.equal({"nested": "yes"})

125
tests/test_ec2/test_subnets.py
View file

@ -11,6 +11,7 @@ from boto.exception import EC2ResponseError
from botocore.exceptions import ParamValidationError, ClientError
import json
import sure # noqa
import random
from moto import mock_cloudformation_deprecated, mock_ec2, mock_ec2_deprecated
@ -474,3 +475,127 @@ def test_create_subnets_with_overlapping_cidr_blocks():
subnet_cidr_block
)
)
@mock_ec2
def test_available_ip_addresses_in_subnet():
ec2 = boto3.resource("ec2", region_name="us-west-1")
client = boto3.client("ec2", region_name="us-west-1")
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
cidr_range_addresses = [
("10.0.0.0/16", 65531),
("10.0.0.0/17", 32763),
("10.0.0.0/18", 16379),
("10.0.0.0/19", 8187),
("10.0.0.0/20", 4091),
("10.0.0.0/21", 2043),
("10.0.0.0/22", 1019),
("10.0.0.0/23", 507),
("10.0.0.0/24", 251),
("10.0.0.0/25", 123),
("10.0.0.0/26", 59),
("10.0.0.0/27", 27),
("10.0.0.0/28", 11),
]
for (cidr, expected_count) in cidr_range_addresses:
validate_subnet_details(client, vpc, cidr, expected_count)
@mock_ec2
def test_available_ip_addresses_in_subnet_with_enis():
ec2 = boto3.resource("ec2", region_name="us-west-1")
client = boto3.client("ec2", region_name="us-west-1")
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
# Verify behaviour for various CIDR ranges (...)
# Don't try to assign ENIs to /27 and /28, as there are not a lot of IP addresses to go around
cidr_range_addresses = [
("10.0.0.0/16", 65531),
("10.0.0.0/17", 32763),
("10.0.0.0/18", 16379),
("10.0.0.0/19", 8187),
("10.0.0.0/20", 4091),
("10.0.0.0/21", 2043),
("10.0.0.0/22", 1019),
("10.0.0.0/23", 507),
("10.0.0.0/24", 251),
("10.0.0.0/25", 123),
("10.0.0.0/26", 59),
]
for (cidr, expected_count) in cidr_range_addresses:
validate_subnet_details_after_creating_eni(client, vpc, cidr, expected_count)
def validate_subnet_details(client, vpc, cidr, expected_ip_address_count):
subnet = client.create_subnet(
VpcId=vpc.id, CidrBlock=cidr, AvailabilityZone="us-west-1b"
)["Subnet"]
subnet["AvailableIpAddressCount"].should.equal(expected_ip_address_count)
client.delete_subnet(SubnetId=subnet["SubnetId"])
def validate_subnet_details_after_creating_eni(
client, vpc, cidr, expected_ip_address_count
):
subnet = client.create_subnet(
VpcId=vpc.id, CidrBlock=cidr, AvailabilityZone="us-west-1b"
)["Subnet"]
# Create a random number of Elastic Network Interfaces
nr_of_eni_to_create = random.randint(0, 5)
ip_addresses_assigned = 0
enis_created = []
for i in range(0, nr_of_eni_to_create):
# Create a random number of IP addresses per ENI
nr_of_ip_addresses = random.randint(1, 5)
if nr_of_ip_addresses == 1:
# Pick the first available IP address (First 4 are reserved by AWS)
private_address = "10.0.0." + str(ip_addresses_assigned + 4)
eni = client.create_network_interface(
SubnetId=subnet["SubnetId"], PrivateIpAddress=private_address
)["NetworkInterface"]
enis_created.append(eni)
ip_addresses_assigned = ip_addresses_assigned + 1
else:
# Assign a list of IP addresses
private_addresses = [
"10.0.0." + str(4 + ip_addresses_assigned + i)
for i in range(0, nr_of_ip_addresses)
]
eni = client.create_network_interface(
SubnetId=subnet["SubnetId"],
PrivateIpAddresses=[
{"PrivateIpAddress": address} for address in private_addresses
],
)["NetworkInterface"]
enis_created.append(eni)
ip_addresses_assigned = ip_addresses_assigned + nr_of_ip_addresses + 1 #
# Verify that the nr of available IP addresses takes these ENIs into account
updated_subnet = client.describe_subnets(SubnetIds=[subnet["SubnetId"]])["Subnets"][
0
]
private_addresses = [
eni["PrivateIpAddress"] for eni in enis_created if eni["PrivateIpAddress"]
]
for eni in enis_created:
private_addresses.extend(
[address["PrivateIpAddress"] for address in eni["PrivateIpAddresses"]]
)
error_msg = (
"Nr of IP addresses for Subnet with CIDR {0} is incorrect. Expected: {1}, Actual: {2}. "
"Addresses: {3}"
)
with sure.ensure(
error_msg,
cidr,
str(expected_ip_address_count),
updated_subnet["AvailableIpAddressCount"],
str(private_addresses),
):
updated_subnet["AvailableIpAddressCount"].should.equal(
expected_ip_address_count - ip_addresses_assigned
)
# Clean up, as we have to create a few more subnets that shouldn't interfere with each other
for eni in enis_created:
client.delete_network_interface(NetworkInterfaceId=eni["NetworkInterfaceId"])
client.delete_subnet(SubnetId=subnet["SubnetId"])

147
tests/test_ec2/test_vpcs.py
View file

@ -678,3 +678,150 @@ def test_create_vpc_with_invalid_cidr_range():
"An error occurred (InvalidVpc.Range) when calling the CreateVpc "
"operation: The CIDR '{}' is invalid.".format(vpc_cidr_block)
)
@mock_ec2
def test_enable_vpc_classic_link():
ec2 = boto3.resource("ec2", region_name="us-west-1")
# Create VPC
vpc = ec2.create_vpc(CidrBlock="10.1.0.0/16")
response = ec2.meta.client.enable_vpc_classic_link(VpcId=vpc.id)
assert response.get("Return").should.be.true
@mock_ec2
def test_enable_vpc_classic_link_failure():
ec2 = boto3.resource("ec2", region_name="us-west-1")
# Create VPC
vpc = ec2.create_vpc(CidrBlock="10.90.0.0/16")
response = ec2.meta.client.enable_vpc_classic_link(VpcId=vpc.id)
assert response.get("Return").should.be.false
@mock_ec2
def test_disable_vpc_classic_link():
ec2 = boto3.resource("ec2", region_name="us-west-1")
# Create VPC
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
ec2.meta.client.enable_vpc_classic_link(VpcId=vpc.id)
response = ec2.meta.client.disable_vpc_classic_link(VpcId=vpc.id)
assert response.get("Return").should.be.false
@mock_ec2
def test_describe_classic_link_enabled():
ec2 = boto3.resource("ec2", region_name="us-west-1")
# Create VPC
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
ec2.meta.client.enable_vpc_classic_link(VpcId=vpc.id)
response = ec2.meta.client.describe_vpc_classic_link(VpcIds=[vpc.id])
assert response.get("Vpcs")[0].get("ClassicLinkEnabled").should.be.true
@mock_ec2
def test_describe_classic_link_disabled():
ec2 = boto3.resource("ec2", region_name="us-west-1")
# Create VPC
vpc = ec2.create_vpc(CidrBlock="10.90.0.0/16")
response = ec2.meta.client.describe_vpc_classic_link(VpcIds=[vpc.id])
assert response.get("Vpcs")[0].get("ClassicLinkEnabled").should.be.false
@mock_ec2
def test_describe_classic_link_multiple():
ec2 = boto3.resource("ec2", region_name="us-west-1")
# Create VPC
vpc1 = ec2.create_vpc(CidrBlock="10.90.0.0/16")
vpc2 = ec2.create_vpc(CidrBlock="10.0.0.0/16")
ec2.meta.client.enable_vpc_classic_link(VpcId=vpc2.id)
response = ec2.meta.client.describe_vpc_classic_link(VpcIds=[vpc1.id, vpc2.id])
expected = [
{"VpcId": vpc1.id, "ClassicLinkDnsSupported": False},
{"VpcId": vpc2.id, "ClassicLinkDnsSupported": True},
]
# Ensure response is sorted, because they can come in random order
assert response.get("Vpcs").sort(key=lambda x: x["VpcId"]) == expected.sort(
key=lambda x: x["VpcId"]
)
@mock_ec2
def test_enable_vpc_classic_link_dns_support():
ec2 = boto3.resource("ec2", region_name="us-west-1")
# Create VPC
vpc = ec2.create_vpc(CidrBlock="10.1.0.0/16")
response = ec2.meta.client.enable_vpc_classic_link_dns_support(VpcId=vpc.id)
assert response.get("Return").should.be.true
@mock_ec2
def test_disable_vpc_classic_link_dns_support():
ec2 = boto3.resource("ec2", region_name="us-west-1")
# Create VPC
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
ec2.meta.client.enable_vpc_classic_link_dns_support(VpcId=vpc.id)
response = ec2.meta.client.disable_vpc_classic_link_dns_support(VpcId=vpc.id)
assert response.get("Return").should.be.false
@mock_ec2
def test_describe_classic_link_dns_support_enabled():
ec2 = boto3.resource("ec2", region_name="us-west-1")
# Create VPC
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
ec2.meta.client.enable_vpc_classic_link_dns_support(VpcId=vpc.id)
response = ec2.meta.client.describe_vpc_classic_link_dns_support(VpcIds=[vpc.id])
assert response.get("Vpcs")[0].get("ClassicLinkDnsSupported").should.be.true
@mock_ec2
def test_describe_classic_link_dns_support_disabled():
ec2 = boto3.resource("ec2", region_name="us-west-1")
# Create VPC
vpc = ec2.create_vpc(CidrBlock="10.90.0.0/16")
response = ec2.meta.client.describe_vpc_classic_link_dns_support(VpcIds=[vpc.id])
assert response.get("Vpcs")[0].get("ClassicLinkDnsSupported").should.be.false
@mock_ec2
def test_describe_classic_link_dns_support_multiple():
ec2 = boto3.resource("ec2", region_name="us-west-1")
# Create VPC
vpc1 = ec2.create_vpc(CidrBlock="10.90.0.0/16")
vpc2 = ec2.create_vpc(CidrBlock="10.0.0.0/16")
ec2.meta.client.enable_vpc_classic_link_dns_support(VpcId=vpc2.id)
response = ec2.meta.client.describe_vpc_classic_link_dns_support(
VpcIds=[vpc1.id, vpc2.id]
)
expected = [
{"VpcId": vpc1.id, "ClassicLinkDnsSupported": False},
{"VpcId": vpc2.id, "ClassicLinkDnsSupported": True},
]
# Ensure response is sorted, because they can come in random order
assert response.get("Vpcs").sort(key=lambda x: x["VpcId"]) == expected.sort(
key=lambda x: x["VpcId"]
)

222
tests/test_events/test_events.py
View file

@ -1,6 +1,7 @@
import random
import boto3
import json
import sure # noqa
from moto.events import mock_events
from botocore.exceptions import ClientError
@ -204,6 +205,53 @@ def test_permissions():
assert resp_policy["Statement"][0]["Sid"] == "Account1"
@mock_events
def test_put_permission_errors():
client = boto3.client("events", "us-east-1")
client.create_event_bus(Name="test-bus")
client.put_permission.when.called_with(
EventBusName="non-existing",
Action="events:PutEvents",
Principal="111111111111",
StatementId="test",
).should.throw(ClientError, "Event bus non-existing does not exist.")
client.put_permission.when.called_with(
EventBusName="test-bus",
Action="events:PutPermission",
Principal="111111111111",
StatementId="test",
).should.throw(
ClientError, "Provided value in parameter 'action' is not supported."
)
@mock_events
def test_remove_permission_errors():
client = boto3.client("events", "us-east-1")
client.create_event_bus(Name="test-bus")
client.remove_permission.when.called_with(
EventBusName="non-existing", StatementId="test"
).should.throw(ClientError, "Event bus non-existing does not exist.")
client.remove_permission.when.called_with(
EventBusName="test-bus", StatementId="test"
).should.throw(ClientError, "EventBus does not have a policy.")
client.put_permission(
EventBusName="test-bus",
Action="events:PutEvents",
Principal="111111111111",
StatementId="test",
)
client.remove_permission.when.called_with(
EventBusName="test-bus", StatementId="non-existing"
).should.throw(ClientError, "Statement with the provided id does not exist.")
@mock_events
def test_put_events():
client = boto3.client("events", "eu-central-1")
@ -220,3 +268,177 @@ def test_put_events():
with assert_raises(ClientError):
client.put_events(Entries=[event] * 20)
@mock_events
def test_create_event_bus():
client = boto3.client("events", "us-east-1")
response = client.create_event_bus(Name="test-bus")
response["EventBusArn"].should.equal(
"arn:aws:events:us-east-1:123456789012:event-bus/test-bus"
)
@mock_events
def test_create_event_bus_errors():
client = boto3.client("events", "us-east-1")
client.create_event_bus(Name="test-bus")
client.create_event_bus.when.called_with(Name="test-bus").should.throw(
ClientError, "Event bus test-bus already exists."
)
# the 'default' name is already used for the account's default event bus.
client.create_event_bus.when.called_with(Name="default").should.throw(
ClientError, "Event bus default already exists."
)
# non partner event buses can't contain the '/' character
client.create_event_bus.when.called_with(Name="test/test-bus").should.throw(
ClientError, "Event bus name must not contain '/'."
)
client.create_event_bus.when.called_with(
Name="aws.partner/test/test-bus", EventSourceName="aws.partner/test/test-bus"
).should.throw(
ClientError, "Event source aws.partner/test/test-bus does not exist."
)
@mock_events
def test_describe_event_bus():
client = boto3.client("events", "us-east-1")
response = client.describe_event_bus()
response["Name"].should.equal("default")
response["Arn"].should.equal(
"arn:aws:events:us-east-1:123456789012:event-bus/default"
)
response.should_not.have.key("Policy")
client.create_event_bus(Name="test-bus")
client.put_permission(
EventBusName="test-bus",
Action="events:PutEvents",
Principal="111111111111",
StatementId="test",
)
response = client.describe_event_bus(Name="test-bus")
response["Name"].should.equal("test-bus")
response["Arn"].should.equal(
"arn:aws:events:us-east-1:123456789012:event-bus/test-bus"
)
json.loads(response["Policy"]).should.equal(
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "test",
"Effect": "Allow",
"Principal": {"AWS": "arn:aws:iam::111111111111:root"},
"Action": "events:PutEvents",
"Resource": "arn:aws:events:us-east-1:123456789012:event-bus/test-bus",
}
],
}
)
@mock_events
def test_describe_event_bus_errors():
client = boto3.client("events", "us-east-1")
client.describe_event_bus.when.called_with(Name="non-existing").should.throw(
ClientError, "Event bus non-existing does not exist."
)
@mock_events
def test_list_event_buses():
client = boto3.client("events", "us-east-1")
client.create_event_bus(Name="test-bus-1")
client.create_event_bus(Name="test-bus-2")
client.create_event_bus(Name="other-bus-1")
client.create_event_bus(Name="other-bus-2")
response = client.list_event_buses()
response["EventBuses"].should.have.length_of(5)
sorted(response["EventBuses"], key=lambda i: i["Name"]).should.equal(
[
{
"Name": "default",
"Arn": "arn:aws:events:us-east-1:123456789012:event-bus/default",
},
{
"Name": "other-bus-1",
"Arn": "arn:aws:events:us-east-1:123456789012:event-bus/other-bus-1",
},
{
"Name": "other-bus-2",
"Arn": "arn:aws:events:us-east-1:123456789012:event-bus/other-bus-2",
},
{
"Name": "test-bus-1",
"Arn": "arn:aws:events:us-east-1:123456789012:event-bus/test-bus-1",
},
{
"Name": "test-bus-2",
"Arn": "arn:aws:events:us-east-1:123456789012:event-bus/test-bus-2",
},
]
)
response = client.list_event_buses(NamePrefix="other-bus")
response["EventBuses"].should.have.length_of(2)
sorted(response["EventBuses"], key=lambda i: i["Name"]).should.equal(
[
{
"Name": "other-bus-1",
"Arn": "arn:aws:events:us-east-1:123456789012:event-bus/other-bus-1",
},
{
"Name": "other-bus-2",
"Arn": "arn:aws:events:us-east-1:123456789012:event-bus/other-bus-2",
},
]
)
@mock_events
def test_delete_event_bus():
client = boto3.client("events", "us-east-1")
client.create_event_bus(Name="test-bus")
response = client.list_event_buses()
response["EventBuses"].should.have.length_of(2)
client.delete_event_bus(Name="test-bus")
response = client.list_event_buses()
response["EventBuses"].should.have.length_of(1)
response["EventBuses"].should.equal(
[
{
"Name": "default",
"Arn": "arn:aws:events:us-east-1:123456789012:event-bus/default",
}
]
)
# deleting non existing event bus should be successful
client.delete_event_bus(Name="non-existing")
@mock_events
def test_delete_event_bus_errors():
client = boto3.client("events", "us-east-1")
client.delete_event_bus.when.called_with(Name="default").should.throw(
ClientError, "Cannot delete event bus default."
)

296
tests/test_iam/test_iam.py
View file

@ -18,6 +18,7 @@ from nose.tools import raises
from datetime import datetime
from tests.helpers import requires_boto_gte
from uuid import uuid4
MOCK_CERT = """-----BEGIN CERTIFICATE-----
@ -169,6 +170,14 @@ def test_create_role_and_instance_profile():
profile.path.should.equal("/")
@mock_iam
def test_create_instance_profile_should_throw_when_name_is_not_unique():
conn = boto3.client("iam", region_name="us-east-1")
conn.create_instance_profile(InstanceProfileName="unique-instance-profile")
with assert_raises(ClientError):
conn.create_instance_profile(InstanceProfileName="unique-instance-profile")
@mock_iam_deprecated()
def test_remove_role_from_instance_profile():
conn = boto.connect_iam()
@ -400,6 +409,21 @@ def test_create_policy():
)
@mock_iam
def test_create_policy_already_exists():
conn = boto3.client("iam", region_name="us-east-1")
response = conn.create_policy(
PolicyName="TestCreatePolicy", PolicyDocument=MOCK_POLICY
)
with assert_raises(conn.exceptions.EntityAlreadyExistsException) as ex:
response = conn.create_policy(
PolicyName="TestCreatePolicy", PolicyDocument=MOCK_POLICY
)
ex.exception.response["Error"]["Code"].should.equal("EntityAlreadyExists")
ex.exception.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(409)
ex.exception.response["Error"]["Message"].should.contain("TestCreatePolicy")
@mock_iam
def test_delete_policy():
conn = boto3.client("iam", region_name="us-east-1")
@ -1292,6 +1316,122 @@ def test_get_access_key_last_used():
resp["UserName"].should.equal(create_key_response["UserName"])
@mock_iam
def test_upload_ssh_public_key():
iam = boto3.resource("iam", region_name="us-east-1")
client = iam.meta.client
username = "test-user"
iam.create_user(UserName=username)
public_key = MOCK_CERT
resp = client.upload_ssh_public_key(UserName=username, SSHPublicKeyBody=public_key)
pubkey = resp["SSHPublicKey"]
pubkey["SSHPublicKeyBody"].should.equal(public_key)
pubkey["UserName"].should.equal(username)
pubkey["SSHPublicKeyId"].should.have.length_of(20)
assert pubkey["SSHPublicKeyId"].startswith("APKA")
pubkey.should.have.key("Fingerprint")
pubkey["Status"].should.equal("Active")
(
datetime.utcnow() - pubkey["UploadDate"].replace(tzinfo=None)
).seconds.should.be.within(0, 10)
@mock_iam
def test_get_ssh_public_key():
iam = boto3.resource("iam", region_name="us-east-1")
client = iam.meta.client
username = "test-user"
iam.create_user(UserName=username)
public_key = MOCK_CERT
with assert_raises(ClientError):
client.get_ssh_public_key(
UserName=username, SSHPublicKeyId="xxnon-existent-keyxx", Encoding="SSH"
)
resp = client.upload_ssh_public_key(UserName=username, SSHPublicKeyBody=public_key)
ssh_public_key_id = resp["SSHPublicKey"]["SSHPublicKeyId"]
resp = client.get_ssh_public_key(
UserName=username, SSHPublicKeyId=ssh_public_key_id, Encoding="SSH"
)
resp["SSHPublicKey"]["SSHPublicKeyBody"].should.equal(public_key)
@mock_iam
def test_list_ssh_public_keys():
iam = boto3.resource("iam", region_name="us-east-1")
client = iam.meta.client
username = "test-user"
iam.create_user(UserName=username)
public_key = MOCK_CERT
resp = client.list_ssh_public_keys(UserName=username)
resp["SSHPublicKeys"].should.have.length_of(0)
resp = client.upload_ssh_public_key(UserName=username, SSHPublicKeyBody=public_key)
ssh_public_key_id = resp["SSHPublicKey"]["SSHPublicKeyId"]
resp = client.list_ssh_public_keys(UserName=username)
resp["SSHPublicKeys"].should.have.length_of(1)
resp["SSHPublicKeys"][0]["SSHPublicKeyId"].should.equal(ssh_public_key_id)
@mock_iam
def test_update_ssh_public_key():
iam = boto3.resource("iam", region_name="us-east-1")
client = iam.meta.client
username = "test-user"
iam.create_user(UserName=username)
public_key = MOCK_CERT
with assert_raises(ClientError):
client.update_ssh_public_key(
UserName=username, SSHPublicKeyId="xxnon-existent-keyxx", Status="Inactive"
)
resp = client.upload_ssh_public_key(UserName=username, SSHPublicKeyBody=public_key)
ssh_public_key_id = resp["SSHPublicKey"]["SSHPublicKeyId"]
resp["SSHPublicKey"]["Status"].should.equal("Active")
resp = client.update_ssh_public_key(
UserName=username, SSHPublicKeyId=ssh_public_key_id, Status="Inactive"
)
resp = client.get_ssh_public_key(
UserName=username, SSHPublicKeyId=ssh_public_key_id, Encoding="SSH"
)
resp["SSHPublicKey"]["Status"].should.equal("Inactive")
@mock_iam
def test_delete_ssh_public_key():
iam = boto3.resource("iam", region_name="us-east-1")
client = iam.meta.client
username = "test-user"
iam.create_user(UserName=username)
public_key = MOCK_CERT
with assert_raises(ClientError):
client.delete_ssh_public_key(
UserName=username, SSHPublicKeyId="xxnon-existent-keyxx"
)
resp = client.upload_ssh_public_key(UserName=username, SSHPublicKeyBody=public_key)
ssh_public_key_id = resp["SSHPublicKey"]["SSHPublicKeyId"]
resp = client.list_ssh_public_keys(UserName=username)
resp["SSHPublicKeys"].should.have.length_of(1)
resp = client.delete_ssh_public_key(
UserName=username, SSHPublicKeyId=ssh_public_key_id
)
resp = client.list_ssh_public_keys(UserName=username)
resp["SSHPublicKeys"].should.have.length_of(0)
@mock_iam
def test_get_account_authorization_details():
test_policy = json.dumps(
@ -2027,6 +2167,42 @@ def test_create_role_with_permissions_boundary():
conn.list_roles().get("Roles")[0].get("PermissionsBoundary").should.equal(expected)
@mock_iam
def test_create_role_with_same_name_should_fail():
iam = boto3.client("iam", region_name="us-east-1")
test_role_name = str(uuid4())
iam.create_role(
RoleName=test_role_name, AssumeRolePolicyDocument="policy", Description="test"
)
# Create the role again, and verify that it fails
with assert_raises(ClientError) as err:
iam.create_role(
RoleName=test_role_name,
AssumeRolePolicyDocument="policy",
Description="test",
)
err.exception.response["Error"]["Code"].should.equal("EntityAlreadyExists")
err.exception.response["Error"]["Message"].should.equal(
"Role with name {0} already exists.".format(test_role_name)
)
@mock_iam
def test_create_policy_with_same_name_should_fail():
iam = boto3.client("iam", region_name="us-east-1")
test_policy_name = str(uuid4())
policy = iam.create_policy(PolicyName=test_policy_name, PolicyDocument=MOCK_POLICY)
# Create the role again, and verify that it fails
with assert_raises(ClientError) as err:
iam.create_policy(PolicyName=test_policy_name, PolicyDocument=MOCK_POLICY)
err.exception.response["Error"]["Code"].should.equal("EntityAlreadyExists")
err.exception.response["Error"]["Message"].should.equal(
"A policy called {0} already exists. Duplicate names are not allowed.".format(
test_policy_name
)
)
@mock_iam
def test_create_open_id_connect_provider():
client = boto3.client("iam", region_name="us-east-1")
@ -2302,3 +2478,123 @@ def test_delete_account_password_policy_errors():
client.delete_account_password_policy.when.called_with().should.throw(
ClientError, "The account policy with name PasswordPolicy cannot be found."
)
@mock_iam
def test_get_account_summary():
client = boto3.client("iam", region_name="us-east-1")
iam = boto3.resource("iam", region_name="us-east-1")
account_summary = iam.AccountSummary()
account_summary.summary_map.should.equal(
{
"GroupPolicySizeQuota": 5120,
"InstanceProfilesQuota": 1000,
"Policies": 0,
"GroupsPerUserQuota": 10,
"InstanceProfiles": 0,
"AttachedPoliciesPerUserQuota": 10,
"Users": 0,
"PoliciesQuota": 1500,
"Providers": 0,
"AccountMFAEnabled": 0,
"AccessKeysPerUserQuota": 2,
"AssumeRolePolicySizeQuota": 2048,
"PolicyVersionsInUseQuota": 10000,
"GlobalEndpointTokenVersion": 1,
"VersionsPerPolicyQuota": 5,
"AttachedPoliciesPerGroupQuota": 10,
"PolicySizeQuota": 6144,
"Groups": 0,
"AccountSigningCertificatesPresent": 0,
"UsersQuota": 5000,
"ServerCertificatesQuota": 20,
"MFADevices": 0,
"UserPolicySizeQuota": 2048,
"PolicyVersionsInUse": 0,
"ServerCertificates": 0,
"Roles": 0,
"RolesQuota": 1000,
"SigningCertificatesPerUserQuota": 2,
"MFADevicesInUse": 0,
"RolePolicySizeQuota": 10240,
"AttachedPoliciesPerRoleQuota": 10,
"AccountAccessKeysPresent": 0,
"GroupsQuota": 300,
}
)
client.create_instance_profile(InstanceProfileName="test-profile")
client.create_open_id_connect_provider(
Url="https://example.com", ThumbprintList=[],
)
response_policy = client.create_policy(
PolicyName="test-policy", PolicyDocument=MOCK_POLICY
)
client.create_role(RoleName="test-role", AssumeRolePolicyDocument="test policy")
client.attach_role_policy(
RoleName="test-role", PolicyArn=response_policy["Policy"]["Arn"]
)
client.create_saml_provider(
Name="TestSAMLProvider", SAMLMetadataDocument="a" * 1024
)
client.create_group(GroupName="test-group")
client.attach_group_policy(
GroupName="test-group", PolicyArn=response_policy["Policy"]["Arn"]
)
client.create_user(UserName="test-user")
client.attach_user_policy(
UserName="test-user", PolicyArn=response_policy["Policy"]["Arn"]
)
client.enable_mfa_device(
UserName="test-user",
SerialNumber="123456789",
AuthenticationCode1="234567",
AuthenticationCode2="987654",
)
client.create_virtual_mfa_device(VirtualMFADeviceName="test-device")
client.upload_server_certificate(
ServerCertificateName="test-cert",
CertificateBody="cert-body",
PrivateKey="private-key",
)
account_summary.load()
account_summary.summary_map.should.equal(
{
"GroupPolicySizeQuota": 5120,
"InstanceProfilesQuota": 1000,
"Policies": 1,
"GroupsPerUserQuota": 10,
"InstanceProfiles": 1,
"AttachedPoliciesPerUserQuota": 10,
"Users": 1,
"PoliciesQuota": 1500,
"Providers": 2,
"AccountMFAEnabled": 0,
"AccessKeysPerUserQuota": 2,
"AssumeRolePolicySizeQuota": 2048,
"PolicyVersionsInUseQuota": 10000,
"GlobalEndpointTokenVersion": 1,
"VersionsPerPolicyQuota": 5,
"AttachedPoliciesPerGroupQuota": 10,
"PolicySizeQuota": 6144,
"Groups": 1,
"AccountSigningCertificatesPresent": 0,
"UsersQuota": 5000,
"ServerCertificatesQuota": 20,
"MFADevices": 1,
"UserPolicySizeQuota": 2048,
"PolicyVersionsInUse": 3,
"ServerCertificates": 1,
"Roles": 1,
"RolesQuota": 1000,
"SigningCertificatesPerUserQuota": 2,
"MFADevicesInUse": 1,
"RolePolicySizeQuota": 10240,
"AttachedPoliciesPerRoleQuota": 10,
"AccountAccessKeysPresent": 0,
"GroupsQuota": 300,
}
)

23
tests/test_iam/test_iam_groups.py
View file

@ -8,6 +8,7 @@ import sure # noqa
from nose.tools import assert_raises
from boto.exception import BotoServerError
from botocore.exceptions import ClientError
from moto import mock_iam, mock_iam_deprecated
MOCK_POLICY = """
@ -182,3 +183,25 @@ def test_list_group_policies():
conn.list_group_policies(GroupName="my-group")["PolicyNames"].should.equal(
["my-policy"]
)
@mock_iam
def test_delete_group():
conn = boto3.client("iam", region_name="us-east-1")
conn.create_group(GroupName="my-group")
groups = conn.list_groups()
assert groups["Groups"][0]["GroupName"] == "my-group"
assert len(groups["Groups"]) == 1
conn.delete_group(GroupName="my-group")
conn.list_groups()["Groups"].should.be.empty
@mock_iam
def test_delete_unknown_group():
conn = boto3.client("iam", region_name="us-east-1")
with assert_raises(ClientError) as err:
conn.delete_group(GroupName="unknown-group")
err.exception.response["Error"]["Code"].should.equal("NoSuchEntity")
err.exception.response["Error"]["Message"].should.equal(
"The group with name unknown-group cannot be found."
)

11
tests/test_organizations/test_organizations_boto3.py
View file

@ -159,6 +159,17 @@ def test_create_account():
create_status["AccountName"].should.equal(mockname)
@mock_organizations
def test_describe_create_account_status():
client = boto3.client("organizations", region_name="us-east-1")
client.create_organization(FeatureSet="ALL")["Organization"]
request_id = client.create_account(AccountName=mockname, Email=mockemail)[
"CreateAccountStatus"
]["Id"]
response = client.describe_create_account_status(CreateAccountRequestId=request_id)
validate_create_account_status(response["CreateAccountStatus"])
@mock_organizations
def test_describe_account():
client = boto3.client("organizations", region_name="us-east-1")

6
tests/test_secretsmanager/test_secretsmanager.py
View file

@ -45,7 +45,7 @@ def test_get_secret_that_does_not_exist():
result = conn.get_secret_value(SecretId="i-dont-exist")
assert_equal(
"Secrets Manager can\u2019t find the specified secret.",
"Secrets Manager can't find the specified secret.",
cm.exception.response["Error"]["Message"],
)
@ -61,7 +61,7 @@ def test_get_secret_that_does_not_match():
result = conn.get_secret_value(SecretId="i-dont-match")
assert_equal(
"Secrets Manager can\u2019t find the specified secret.",
"Secrets Manager can't find the specified secret.",
cm.exception.response["Error"]["Message"],
)
@ -88,7 +88,7 @@ def test_get_secret_that_has_no_value():
result = conn.get_secret_value(SecretId="java-util-test-password")
assert_equal(
"Secrets Manager can\u2019t find the specified secret value for staging label: AWSCURRENT",
"Secrets Manager can't find the specified secret value for staging label: AWSCURRENT",
cm.exception.response["Error"]["Message"],
)

26
tests/test_secretsmanager/test_server.py
View file

@ -48,9 +48,7 @@ def test_get_secret_that_does_not_exist():
headers={"X-Amz-Target": "secretsmanager.GetSecretValue"},
)
json_data = json.loads(get_secret.data.decode("utf-8"))
assert (
json_data["message"] == "Secrets Manager can\u2019t find the specified secret."
)
assert json_data["message"] == "Secrets Manager can't find the specified secret."
assert json_data["__type"] == "ResourceNotFoundException"
@ -70,9 +68,7 @@ def test_get_secret_that_does_not_match():
headers={"X-Amz-Target": "secretsmanager.GetSecretValue"},
)
json_data = json.loads(get_secret.data.decode("utf-8"))
assert (
json_data["message"] == "Secrets Manager can\u2019t find the specified secret."
)
assert json_data["message"] == "Secrets Manager can't find the specified secret."
assert json_data["__type"] == "ResourceNotFoundException"
@ -95,7 +91,7 @@ def test_get_secret_that_has_no_value():
json_data = json.loads(get_secret.data.decode("utf-8"))
assert (
json_data["message"]
== "Secrets Manager can\u2019t find the specified secret value for staging label: AWSCURRENT"
== "Secrets Manager can't find the specified secret value for staging label: AWSCURRENT"
)
assert json_data["__type"] == "ResourceNotFoundException"
@ -178,9 +174,7 @@ def test_describe_secret_that_does_not_exist():
)
json_data = json.loads(describe_secret.data.decode("utf-8"))
assert (
json_data["message"] == "Secrets Manager can\u2019t find the specified secret."
)
assert json_data["message"] == "Secrets Manager can't find the specified secret."
assert json_data["__type"] == "ResourceNotFoundException"
@ -202,9 +196,7 @@ def test_describe_secret_that_does_not_match():
)
json_data = json.loads(describe_secret.data.decode("utf-8"))
assert (
json_data["message"] == "Secrets Manager can\u2019t find the specified secret."
)
assert json_data["message"] == "Secrets Manager can't find the specified secret."
assert json_data["__type"] == "ResourceNotFoundException"
@ -306,9 +298,7 @@ def test_rotate_secret_that_does_not_exist():
)
json_data = json.loads(rotate_secret.data.decode("utf-8"))
assert (
json_data["message"] == "Secrets Manager can\u2019t find the specified secret."
)
assert json_data["message"] == "Secrets Manager can't find the specified secret."
assert json_data["__type"] == "ResourceNotFoundException"
@ -330,9 +320,7 @@ def test_rotate_secret_that_does_not_match():
)
json_data = json.loads(rotate_secret.data.decode("utf-8"))
assert (
json_data["message"] == "Secrets Manager can\u2019t find the specified secret."
)
assert json_data["message"] == "Secrets Manager can't find the specified secret."
assert json_data["__type"] == "ResourceNotFoundException"

21
tests/test_sns/test_publishing_boto3.py
View file

@ -173,6 +173,27 @@ def test_publish_to_sqs_msg_attr_byte_value():
)
@mock_sqs
@mock_sns
def test_publish_to_sqs_msg_attr_number_type():
sns = boto3.resource("sns", region_name="us-east-1")
topic = sns.create_topic(Name="test-topic")
sqs = boto3.resource("sqs", region_name="us-east-1")
queue = sqs.create_queue(QueueName="test-queue")
topic.subscribe(Protocol="sqs", Endpoint=queue.attributes["QueueArn"])
topic.publish(
Message="test message",
MessageAttributes={"retries": {"DataType": "Number", "StringValue": "0"}},
)
message = json.loads(queue.receive_messages()[0].body)
message["Message"].should.equal("test message")
message["MessageAttributes"].should.equal(
{"retries": {"Type": "Number", "Value": 0}}
)
@mock_sns
def test_publish_sms():
client = boto3.client("sns", region_name="us-east-1")