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

Merge remote-tracking branch 'upstream/master'

Browse source
This commit is contained in:
Stephan Huber 2019-08-30 14:39:29 +02:00
commit 5a9c921d97
132 changed files with 33915 additions and 9581 deletions
Download patch file Download diff file Expand all files Collapse all files

25
tests/test_ec2/test_amis.py
View file

@ -10,7 +10,7 @@ from nose.tools import assert_raises
import sure # noqa
from moto import mock_ec2_deprecated, mock_ec2
from moto.ec2.models import AMIS
from moto.ec2.models import AMIS, OWNER_ID
from tests.helpers import requires_boto_gte
@ -152,6 +152,29 @@ def test_ami_copy():
cm.exception.request_id.should_not.be.none
@mock_ec2
def test_copy_image_changes_owner_id():
conn = boto3.client('ec2', region_name='us-east-1')
# this source AMI ID is from moto/ec2/resources/amis.json
source_ami_id = "ami-03cf127a"
# confirm the source ami owner id is different from the default owner id.
# if they're ever the same it means this test is invalid.
check_resp = conn.describe_images(ImageIds=[source_ami_id])
check_resp["Images"][0]["OwnerId"].should_not.equal(OWNER_ID)
copy_resp = conn.copy_image(
SourceImageId=source_ami_id,
Name="new-image",
Description="a copy of an image",
SourceRegion="us-east-1")
describe_resp = conn.describe_images(Owners=["self"])
describe_resp["Images"][0]["OwnerId"].should.equal(OWNER_ID)
describe_resp["Images"][0]["ImageId"].should.equal(copy_resp["ImageId"])
@mock_ec2_deprecated
def test_ami_tagging():
conn = boto.connect_vpc('the_key', 'the_secret')

3
tests/test_ec2/test_elastic_block_store.py
View file

@ -12,6 +12,7 @@ from freezegun import freeze_time
import sure # noqa
from moto import mock_ec2_deprecated, mock_ec2
from moto.ec2.models import OWNER_ID
@mock_ec2_deprecated
@ -395,7 +396,7 @@ def test_snapshot_filters():
).should.equal({snapshot3.id})
snapshots_by_owner_id = conn.get_all_snapshots(
filters={'owner-id': '123456789012'})
filters={'owner-id': OWNER_ID})
set([snap.id for snap in snapshots_by_owner_id]
).should.equal({snapshot1.id, snapshot2.id, snapshot3.id})

108
tests/test_ec2/test_elastic_network_interfaces.py
View file

@ -161,7 +161,7 @@ def test_elastic_network_interfaces_filtering():
subnet.id, groups=[security_group1.id, security_group2.id])
eni2 = conn.create_network_interface(
subnet.id, groups=[security_group1.id])
eni3 = conn.create_network_interface(subnet.id)
eni3 = conn.create_network_interface(subnet.id, description='test description')
all_enis = conn.get_all_network_interfaces()
all_enis.should.have.length_of(3)
@ -189,6 +189,12 @@ def test_elastic_network_interfaces_filtering():
enis_by_group.should.have.length_of(1)
set([eni.id for eni in enis_by_group]).should.equal(set([eni1.id]))
# Filter by Description
enis_by_description = conn.get_all_network_interfaces(
filters={'description': eni3.description })
enis_by_description.should.have.length_of(1)
enis_by_description[0].description.should.equal(eni3.description)
# Unsupported filter
conn.get_all_network_interfaces.when.called_with(
filters={'not-implemented-filter': 'foobar'}).should.throw(NotImplementedError)
@ -343,6 +349,106 @@ def test_elastic_network_interfaces_get_by_subnet_id():
enis.should.have.length_of(0)
@mock_ec2
def test_elastic_network_interfaces_get_by_description():
ec2 = boto3.resource('ec2', region_name='us-west-2')
ec2_client = boto3.client('ec2', region_name='us-west-2')
vpc = ec2.create_vpc(CidrBlock='10.0.0.0/16')
subnet = ec2.create_subnet(
VpcId=vpc.id, CidrBlock='10.0.0.0/24', AvailabilityZone='us-west-2a')
eni1 = ec2.create_network_interface(
SubnetId=subnet.id, PrivateIpAddress='10.0.10.5', Description='test interface')
# The status of the new interface should be 'available'
waiter = ec2_client.get_waiter('network_interface_available')
waiter.wait(NetworkInterfaceIds=[eni1.id])
filters = [{'Name': 'description', 'Values': [eni1.description]}]
enis = list(ec2.network_interfaces.filter(Filters=filters))
enis.should.have.length_of(1)
filters = [{'Name': 'description', 'Values': ['bad description']}]
enis = list(ec2.network_interfaces.filter(Filters=filters))
enis.should.have.length_of(0)
@mock_ec2
def test_elastic_network_interfaces_describe_network_interfaces_with_filter():
ec2 = boto3.resource('ec2', region_name='us-west-2')
ec2_client = boto3.client('ec2', region_name='us-west-2')
vpc = ec2.create_vpc(CidrBlock='10.0.0.0/16')
subnet = ec2.create_subnet(
VpcId=vpc.id, CidrBlock='10.0.0.0/24', AvailabilityZone='us-west-2a')
eni1 = ec2.create_network_interface(
SubnetId=subnet.id, PrivateIpAddress='10.0.10.5', Description='test interface')
# The status of the new interface should be 'available'
waiter = ec2_client.get_waiter('network_interface_available')
waiter.wait(NetworkInterfaceIds=[eni1.id])
# Filter by network-interface-id
response = ec2_client.describe_network_interfaces(
Filters=[{'Name': 'network-interface-id', 'Values': [eni1.id]}])
response['NetworkInterfaces'].should.have.length_of(1)
response['NetworkInterfaces'][0]['NetworkInterfaceId'].should.equal(eni1.id)
response['NetworkInterfaces'][0]['PrivateIpAddress'].should.equal(eni1.private_ip_address)
response['NetworkInterfaces'][0]['Description'].should.equal(eni1.description)
response = ec2_client.describe_network_interfaces(
Filters=[{'Name': 'network-interface-id', 'Values': ['bad-id']}])
response['NetworkInterfaces'].should.have.length_of(0)
# Filter by private-ip-address
response = ec2_client.describe_network_interfaces(
Filters=[{'Name': 'private-ip-address', 'Values': [eni1.private_ip_address]}])
response['NetworkInterfaces'].should.have.length_of(1)
response['NetworkInterfaces'][0]['NetworkInterfaceId'].should.equal(eni1.id)
response['NetworkInterfaces'][0]['PrivateIpAddress'].should.equal(eni1.private_ip_address)
response['NetworkInterfaces'][0]['Description'].should.equal(eni1.description)
response = ec2_client.describe_network_interfaces(
Filters=[{'Name': 'private-ip-address', 'Values': ['11.11.11.11']}])
response['NetworkInterfaces'].should.have.length_of(0)
# Filter by sunet-id
response = ec2_client.describe_network_interfaces(
Filters=[{'Name': 'subnet-id', 'Values': [eni1.subnet.id]}])
response['NetworkInterfaces'].should.have.length_of(1)
response['NetworkInterfaces'][0]['NetworkInterfaceId'].should.equal(eni1.id)
response['NetworkInterfaces'][0]['PrivateIpAddress'].should.equal(eni1.private_ip_address)
response['NetworkInterfaces'][0]['Description'].should.equal(eni1.description)
response = ec2_client.describe_network_interfaces(
Filters=[{'Name': 'subnet-id', 'Values': ['sn-bad-id']}])
response['NetworkInterfaces'].should.have.length_of(0)
# Filter by description
response = ec2_client.describe_network_interfaces(
Filters=[{'Name': 'description', 'Values': [eni1.description]}])
response['NetworkInterfaces'].should.have.length_of(1)
response['NetworkInterfaces'][0]['NetworkInterfaceId'].should.equal(eni1.id)
response['NetworkInterfaces'][0]['PrivateIpAddress'].should.equal(eni1.private_ip_address)
response['NetworkInterfaces'][0]['Description'].should.equal(eni1.description)
response = ec2_client.describe_network_interfaces(
Filters=[{'Name': 'description', 'Values': ['bad description']}])
response['NetworkInterfaces'].should.have.length_of(0)
# Filter by multiple filters
response = ec2_client.describe_network_interfaces(
Filters=[{'Name': 'private-ip-address', 'Values': [eni1.private_ip_address]},
{'Name': 'network-interface-id', 'Values': [eni1.id]},
{'Name': 'subnet-id', 'Values': [eni1.subnet.id]}])
response['NetworkInterfaces'].should.have.length_of(1)
response['NetworkInterfaces'][0]['NetworkInterfaceId'].should.equal(eni1.id)
response['NetworkInterfaces'][0]['PrivateIpAddress'].should.equal(eni1.private_ip_address)
response['NetworkInterfaces'][0]['Description'].should.equal(eni1.description)
@mock_ec2_deprecated
@mock_cloudformation_deprecated
def test_elastic_network_interfaces_cloudformation():

41
tests/test_ec2/test_instances.py
View file

@ -1,5 +1,7 @@
from __future__ import unicode_literals
# Ensure 'assert_raises' context manager support for Python 2.6
from botocore.exceptions import ClientError
import tests.backport_assert_raises
from nose.tools import assert_raises
@ -679,8 +681,8 @@ def test_modify_instance_attribute_security_groups():
reservation = conn.run_instances('ami-1234abcd')
instance = reservation.instances[0]
sg_id = 'sg-1234abcd'
sg_id2 = 'sg-abcd4321'
sg_id = conn.create_security_group('test security group', 'this is a test security group').id
sg_id2 = conn.create_security_group('test security group 2', 'this is a test security group 2').id
with assert_raises(EC2ResponseError) as ex:
instance.modify_attribute("groupSet", [sg_id, sg_id2], dry_run=True)
@ -1255,6 +1257,7 @@ def test_create_instance_ebs_optimized():
instance.load()
instance.ebs_optimized.should.be(False)
@mock_ec2
def test_run_multiple_instances_in_same_command():
instance_count = 4
@ -1269,3 +1272,37 @@ def test_run_multiple_instances_in_same_command():
instances = reservations[0]['Instances']
for i in range(0, instance_count):
instances[i]['AmiLaunchIndex'].should.be(i)
@mock_ec2
def test_describe_instance_attribute():
client = boto3.client('ec2', region_name='us-east-1')
security_group_id = client.create_security_group(
GroupName='test security group', Description='this is a test security group')['GroupId']
client.run_instances(ImageId='ami-1234abcd',
MinCount=1,
MaxCount=1,
SecurityGroupIds=[security_group_id])
instance_id = client.describe_instances()['Reservations'][0]['Instances'][0]['InstanceId']
valid_instance_attributes = ['instanceType', 'kernel', 'ramdisk', 'userData', 'disableApiTermination', 'instanceInitiatedShutdownBehavior', 'rootDeviceName', 'blockDeviceMapping', 'productCodes', 'sourceDestCheck', 'groupSet', 'ebsOptimized', 'sriovNetSupport']
for valid_instance_attribute in valid_instance_attributes:
response = client.describe_instance_attribute(InstanceId=instance_id, Attribute=valid_instance_attribute)
if valid_instance_attribute == "groupSet":
response.should.have.key("Groups")
response["Groups"].should.have.length_of(1)
response["Groups"][0]["GroupId"].should.equal(security_group_id)
elif valid_instance_attribute == "userData":
response.should.have.key("UserData")
response["UserData"].should.be.empty
invalid_instance_attributes = ['abc', 'Kernel', 'RamDisk', 'userdata', 'iNsTaNcEtYpE']
for invalid_instance_attribute in invalid_instance_attributes:
with assert_raises(ClientError) as ex:
client.describe_instance_attribute(InstanceId=instance_id, Attribute=invalid_instance_attribute)
ex.exception.response['Error']['Code'].should.equal('InvalidParameterValue')
ex.exception.response['ResponseMetadata']['HTTPStatusCode'].should.equal(400)
message = 'Value ({invalid_instance_attribute}) for parameter attribute is invalid. Unknown attribute.'.format(invalid_instance_attribute=invalid_instance_attribute)
ex.exception.response['Error']['Message'].should.equal(message)

4
tests/test_ec2/test_network_acls.py
View file

@ -30,12 +30,12 @@ def test_new_subnet_associates_with_default_network_acl():
conn = boto.connect_vpc('the_key', 'the secret')
vpc = conn.get_all_vpcs()[0]
subnet = conn.create_subnet(vpc.id, "172.31.48.0/20")
subnet = conn.create_subnet(vpc.id, "172.31.112.0/20")
all_network_acls = conn.get_all_network_acls()
all_network_acls.should.have.length_of(1)
acl = all_network_acls[0]
acl.associations.should.have.length_of(4)
acl.associations.should.have.length_of(7)
[a.subnet_id for a in acl.associations].should.contain(subnet.id)

108
tests/test_ec2/test_subnets.py
View file

@ -118,7 +118,7 @@ def test_boto3_non_default_subnet():
@mock_ec2
def test_modify_subnet_attribute():
def test_modify_subnet_attribute_public_ip_on_launch():
ec2 = boto3.resource('ec2', region_name='us-west-1')
client = boto3.client('ec2', region_name='us-west-1')
@ -145,6 +145,34 @@ def test_modify_subnet_attribute():
subnet.map_public_ip_on_launch.should.be.ok
@mock_ec2
def test_modify_subnet_attribute_assign_ipv6_address_on_creation():
ec2 = boto3.resource('ec2', region_name='us-west-1')
client = boto3.client('ec2', region_name='us-west-1')
# Get the default VPC
vpc = list(ec2.vpcs.all())[0]
subnet = ec2.create_subnet(
VpcId=vpc.id, CidrBlock='172.31.112.0/20', AvailabilityZone='us-west-1a')
# 'map_public_ip_on_launch' is set when calling 'DescribeSubnets' action
subnet.reload()
# For non default subnet, attribute value should be 'False'
subnet.assign_ipv6_address_on_creation.shouldnt.be.ok
client.modify_subnet_attribute(
SubnetId=subnet.id, AssignIpv6AddressOnCreation={'Value': False})
subnet.reload()
subnet.assign_ipv6_address_on_creation.shouldnt.be.ok
client.modify_subnet_attribute(
SubnetId=subnet.id, AssignIpv6AddressOnCreation={'Value': True})
subnet.reload()
subnet.assign_ipv6_address_on_creation.should.be.ok
@mock_ec2
def test_modify_subnet_attribute_validation():
ec2 = boto3.resource('ec2', region_name='us-west-1')
@ -291,6 +319,84 @@ def test_subnet_tags_through_cloudformation():
subnet.tags["blah"].should.equal("baz")
@mock_ec2
def test_create_subnet_response_fields():
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')
subnet = client.create_subnet(
VpcId=vpc.id, CidrBlock='10.0.0.0/24', AvailabilityZone='us-west-1a')['Subnet']
subnet.should.have.key('AvailabilityZone')
subnet.should.have.key('AvailabilityZoneId')
subnet.should.have.key('AvailableIpAddressCount')
subnet.should.have.key('CidrBlock')
subnet.should.have.key('State')
subnet.should.have.key('SubnetId')
subnet.should.have.key('VpcId')
subnet.shouldnt.have.key('Tags')
subnet.should.have.key('DefaultForAz').which.should.equal(False)
subnet.should.have.key('MapPublicIpOnLaunch').which.should.equal(False)
subnet.should.have.key('OwnerId')
subnet.should.have.key('AssignIpv6AddressOnCreation').which.should.equal(False)
subnet_arn = "arn:aws:ec2:{region}:{owner_id}:subnet/{subnet_id}".format(region=subnet['AvailabilityZone'][0:-1],
owner_id=subnet['OwnerId'],
subnet_id=subnet['SubnetId'])
subnet.should.have.key('SubnetArn').which.should.equal(subnet_arn)
subnet.should.have.key('Ipv6CidrBlockAssociationSet').which.should.equal([])
@mock_ec2
def test_describe_subnet_response_fields():
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')
subnet_object = ec2.create_subnet(
VpcId=vpc.id, CidrBlock='10.0.0.0/24', AvailabilityZone='us-west-1a')
subnets = client.describe_subnets(SubnetIds=[subnet_object.id])['Subnets']
subnets.should.have.length_of(1)
subnet = subnets[0]
subnet.should.have.key('AvailabilityZone')
subnet.should.have.key('AvailabilityZoneId')
subnet.should.have.key('AvailableIpAddressCount')
subnet.should.have.key('CidrBlock')
subnet.should.have.key('State')
subnet.should.have.key('SubnetId')
subnet.should.have.key('VpcId')
subnet.shouldnt.have.key('Tags')
subnet.should.have.key('DefaultForAz').which.should.equal(False)
subnet.should.have.key('MapPublicIpOnLaunch').which.should.equal(False)
subnet.should.have.key('OwnerId')
subnet.should.have.key('AssignIpv6AddressOnCreation').which.should.equal(False)
subnet_arn = "arn:aws:ec2:{region}:{owner_id}:subnet/{subnet_id}".format(region=subnet['AvailabilityZone'][0:-1],
owner_id=subnet['OwnerId'],
subnet_id=subnet['SubnetId'])
subnet.should.have.key('SubnetArn').which.should.equal(subnet_arn)
subnet.should.have.key('Ipv6CidrBlockAssociationSet').which.should.equal([])
@mock_ec2
def test_create_subnet_with_invalid_availability_zone():
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')
subnet_availability_zone = 'asfasfas'
with assert_raises(ClientError) as ex:
subnet = client.create_subnet(
VpcId=vpc.id, CidrBlock='10.0.0.0/24', AvailabilityZone=subnet_availability_zone)
assert str(ex.exception).startswith(
"An error occurred (InvalidParameterValue) when calling the CreateSubnet "
"operation: Value ({}) for parameter availabilityZone is invalid. Subnets can currently only be created in the following availability zones: ".format(subnet_availability_zone))
@mock_ec2
def test_create_subnet_with_invalid_cidr_range():
ec2 = boto3.resource('ec2', region_name='us-west-1')