Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #1378 from william-richard/start-running-ami-tests

Browse Source 
Make test_amis not executable, so nose runs it
This commit is contained in:
Jack Danger 2018-01-24 18:28:40 -08:00 committed by GitHub
commit 4520cd930f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 120 additions and 93 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
moto/core/responses.py
View File

@ -192,7 +192,7 @@ class BaseResponse(_TemplateEnvironmentMixin):
return self.querystring.get('AWSAccessKeyId')
else:
# Should we raise an unauthorized exception instead?
return None
return '111122223333'
def _dispatch(self, request, full_url, headers):
self.setup_class(request, full_url, headers)

72
moto/ec2/models.py
View File

@ -48,7 +48,6 @@ from .exceptions import (
InvalidRouteError,
InvalidInstanceIdError,
InvalidAMIIdError,
MalformedAMIIdError,
InvalidAMIAttributeItemValueError,
InvalidSnapshotIdError,
InvalidVolumeIdError,
@ -68,8 +67,8 @@ from .exceptions import (
InvalidCustomerGatewayIdError,
RulesPerSecurityGroupLimitExceededError,
MotoNotImplementedError,
FilterNotImplementedError
)
FilterNotImplementedError,
MalformedAMIIdError)
from .utils import (
EC2_RESOURCE_TO_PREFIX,
EC2_PREFIX_TO_RESOURCE,
@ -1032,11 +1031,11 @@ class TagBackend(object):
class Ami(TaggedEC2Resource):
def __init__(self, ec2_backend, ami_id, instance=None, source_ami=None,
name=None, description=None, owner_id=None,
name=None, description=None, owner_id=111122223333,
public=False, virtualization_type=None, architecture=None,
state='available', creation_date=None, platform=None,
image_type='machine', image_location=None, hypervisor=None,
root_device_type=None, root_device_name=None, sriov='simple',
root_device_type='standard', root_device_name='/dev/sda1', sriov='simple',
region_name='us-east-1a'
):
self.ec2_backend = ec2_backend
@ -1137,14 +1136,14 @@ class AmiBackend(object):
ami_id = ami['ami_id']
self.amis[ami_id] = Ami(self, **ami)
def create_image(self, instance_id, name=None, description=None,
context=None):
def create_image(self, instance_id, name=None, description=None, context=None):
# TODO: check that instance exists and pull info from it.
ami_id = random_ami_id()
instance = self.get_instance(instance_id)
ami = Ami(self, ami_id, instance=instance, source_ami=None,
name=name, description=description,
owner_id=context.get_current_user() if context else None)
owner_id=context.get_current_user() if context else '111122223333')
self.amis[ami_id] = ami
return ami
@ -1161,36 +1160,39 @@ class AmiBackend(object):
context=None):
images = self.amis.values()
# Limit images by launch permissions
if exec_users:
tmp_images = []
for ami in images:
for user_id in exec_users:
if user_id in ami.launch_permission_users:
tmp_images.append(ami)
images = tmp_images
if len(ami_ids):
# boto3 seems to default to just searching based on ami ids if that parameter is passed
# and if no images are found, it raises an errors
malformed_ami_ids = [ami_id for ami_id in ami_ids if not ami_id.startswith('ami-')]
if malformed_ami_ids:
raise MalformedAMIIdError(malformed_ami_ids)
# Limit by owner ids
if owners:
# support filtering by Owners=['self']
owners = list(map(
lambda o: context.get_current_user()
if context and o == 'self' else o,
owners))
images = [ami for ami in images if ami.owner_id in owners]
if ami_ids:
images = [ami for ami in images if ami.id in ami_ids]
if len(ami_ids) > len(images):
unknown_ids = set(ami_ids) - set(images)
for id in unknown_ids:
if not self.AMI_REGEX.match(id):
raise MalformedAMIIdError(id)
raise InvalidAMIIdError(unknown_ids)
if len(images) == 0:
raise InvalidAMIIdError(ami_ids)
else:
# Limit images by launch permissions
if exec_users:
tmp_images = []
for ami in images:
for user_id in exec_users:
if user_id in ami.launch_permission_users:
tmp_images.append(ami)
images = tmp_images
# Limit by owner ids
if owners:
# support filtering by Owners=['self']
owners = list(map(
lambda o: context.get_current_user()
if context and o == 'self' else o,
owners))
images = [ami for ami in images if ami.owner_id in owners]
# Generic filters
if filters:
return generic_filter(filters, images)
# Generic filters
if filters:
return generic_filter(filters, images)
return images
def deregister_image(self, ami_id):

4
moto/ec2/responses/amis.py
View File

@ -113,12 +113,12 @@ DESCRIBE_IMAGES_RESPONSE = """<DescribeImagesResponse xmlns="http://ec2.amazonaw
<rootDeviceName>{{ image.root_device_name }}</rootDeviceName>
<blockDeviceMapping>
<item>
<deviceName>/dev/sda1</deviceName>
<deviceName>{{ image.root_device_name }}</deviceName>
<ebs>
<snapshotId>{{ image.ebs_snapshot.id }}</snapshotId>
<volumeSize>15</volumeSize>
<deleteOnTermination>false</deleteOnTermination>
<volumeType>standard</volumeType>
<volumeType>{{ image.root_device_type }}</volumeType>
</ebs>
</item>
</blockDeviceMapping>

135
tests/test_ec2/test_amis.py Executable file → Normal file
View File

@ -1,15 +1,12 @@
from __future__ import unicode_literals
# Ensure 'assert_raises' context manager support for Python 2.6
import tests.backport_assert_raises # noqa
from nose.tools import assert_raises
import boto
import boto3
import boto.ec2
import boto3
from boto.exception import EC2ResponseError, EC2ResponseError
from boto.exception import EC2ResponseError
from botocore.exceptions import ClientError
# Ensure 'assert_raises' context manager support for Python 2.6
from nose.tools import assert_raises
import sure # noqa
from moto import mock_ec2_deprecated, mock_ec2
@ -19,6 +16,11 @@ from tests.helpers import requires_boto_gte
@mock_ec2_deprecated
def test_ami_create_and_delete():
conn = boto.connect_ec2('the_key', 'the_secret')
initial_volume_count = 34
conn.get_all_volumes().should.have.length_of(initial_volume_count)
conn.get_all_snapshots().should.have.length_of(initial_volume_count)
reservation = conn.run_instances('ami-1234abcd')
instance = reservation.instances[0]
@ -33,33 +35,34 @@ def test_ami_create_and_delete():
image_id = conn.create_image(instance.id, "test-ami", "this is a test ami")
all_images = conn.get_all_images()
image = all_images[0]
set([i.id for i in all_images]).should.contain(image_id)
image.id.should.equal(image_id)
image.virtualization_type.should.equal(instance.virtualization_type)
image.architecture.should.equal(instance.architecture)
image.kernel_id.should.equal(instance.kernel)
image.platform.should.equal(instance.platform)
image.creationDate.should_not.be.none
retrieved_image = [i for i in all_images if i.id == image_id][0]
retrieved_image.id.should.equal(image_id)
retrieved_image.virtualization_type.should.equal(instance.virtualization_type)
retrieved_image.architecture.should.equal(instance.architecture)
retrieved_image.kernel_id.should.equal(instance.kernel)
retrieved_image.platform.should.equal(instance.platform)
retrieved_image.creationDate.should_not.be.none
instance.terminate()
# Validate auto-created volume and snapshot
volumes = conn.get_all_volumes()
volumes.should.have.length_of(1)
volume = volumes[0]
volumes.should.have.length_of(initial_volume_count + 1)
snapshots = conn.get_all_snapshots()
snapshots.should.have.length_of(1)
snapshot = snapshots[0]
snapshots.should.have.length_of(initial_volume_count + 1)
image.block_device_mapping.current_value.snapshot_id.should.equal(
snapshot.id)
retrieved_image_snapshot_id = retrieved_image.block_device_mapping.current_value.snapshot_id
[s.id for s in snapshots].should.contain(retrieved_image_snapshot_id)
snapshot = [s for s in snapshots if s.id == retrieved_image_snapshot_id][0]
snapshot.description.should.equal(
"Auto-created snapshot for AMI {0}".format(image.id))
snapshot.volume_id.should.equal(volume.id)
"Auto-created snapshot for AMI {0}".format(retrieved_image.id))
[v.id for v in volumes].should.contain(snapshot.volume_id)
# root device should be in AMI's block device mappings
root_mapping = image.block_device_mapping.get(image.root_device_name)
root_mapping = retrieved_image.block_device_mapping.get(retrieved_image.root_device_name)
root_mapping.should_not.be.none
# Deregister
@ -84,6 +87,11 @@ def test_ami_create_and_delete():
@mock_ec2_deprecated
def test_ami_copy():
conn = boto.ec2.connect_to_region("us-west-1")
initial_volume_count = 34
conn.get_all_volumes().should.have.length_of(initial_volume_count)
conn.get_all_snapshots().should.have.length_of(initial_volume_count)
reservation = conn.run_instances('ami-1234abcd')
instance = reservation.instances[0]
@ -96,7 +104,8 @@ def test_ami_copy():
# the image_id to fetch the full info.
with assert_raises(EC2ResponseError) as ex:
copy_image_ref = conn.copy_image(
source_image.region.name, source_image.id, "test-copy-ami", "this is a test copy ami", dry_run=True)
source_image.region.name, source_image.id, "test-copy-ami", "this is a test copy ami",
dry_run=True)
ex.exception.error_code.should.equal('DryRunOperation')
ex.exception.status.should.equal(400)
ex.exception.message.should.equal(
@ -115,8 +124,8 @@ def test_ami_copy():
copy_image.platform.should.equal(source_image.platform)
# Validate auto-created volume and snapshot
conn.get_all_volumes().should.have.length_of(2)
conn.get_all_snapshots().should.have.length_of(2)
conn.get_all_volumes().should.have.length_of(initial_volume_count + 2)
conn.get_all_snapshots().should.have.length_of(initial_volume_count + 2)
copy_image.block_device_mapping.current_value.snapshot_id.should_not.equal(
source_image.block_device_mapping.current_value.snapshot_id)
@ -218,7 +227,8 @@ def test_ami_filters():
amis_by_architecture = conn.get_all_images(
filters={'architecture': 'x86_64'})
set([ami.id for ami in amis_by_architecture]).should.equal(set([imageB.id]))
set([ami.id for ami in amis_by_architecture]).should.contain(imageB.id)
len(amis_by_architecture).should.equal(35)
amis_by_kernel = conn.get_all_images(filters={'kernel-id': 'k-abcd1234'})
set([ami.id for ami in amis_by_kernel]).should.equal(set([imageB.id]))
@ -226,26 +236,32 @@ def test_ami_filters():
amis_by_virtualization = conn.get_all_images(
filters={'virtualization-type': 'paravirtual'})
set([ami.id for ami in amis_by_virtualization]
).should.equal(set([imageB.id]))
).should.contain(imageB.id)
len(amis_by_virtualization).should.equal(3)
amis_by_platform = conn.get_all_images(filters={'platform': 'windows'})
set([ami.id for ami in amis_by_platform]).should.equal(set([imageA.id]))
set([ami.id for ami in amis_by_platform]).should.contain(imageA.id)
len(amis_by_platform).should.equal(24)
amis_by_id = conn.get_all_images(filters={'image-id': imageA.id})
set([ami.id for ami in amis_by_id]).should.equal(set([imageA.id]))
amis_by_state = conn.get_all_images(filters={'state': 'available'})
set([ami.id for ami in amis_by_state]).should.equal(
set([imageA.id, imageB.id]))
ami_ids_by_state = [ami.id for ami in amis_by_state]
ami_ids_by_state.should.contain(imageA.id)
ami_ids_by_state.should.contain(imageB.id)
len(amis_by_state).should.equal(36)
amis_by_name = conn.get_all_images(filters={'name': imageA.name})
set([ami.id for ami in amis_by_name]).should.equal(set([imageA.id]))
amis_by_public = conn.get_all_images(filters={'is-public': True})
set([ami.id for ami in amis_by_public]).should.equal(set([imageB.id]))
set([ami.id for ami in amis_by_public]).should.contain(imageB.id)
len(amis_by_public).should.equal(35)
amis_by_nonpublic = conn.get_all_images(filters={'is-public': False})
set([ami.id for ami in amis_by_nonpublic]).should.equal(set([imageA.id]))
set([ami.id for ami in amis_by_nonpublic]).should.contain(imageA.id)
len(amis_by_nonpublic).should.equal(1)
@mock_ec2_deprecated
@ -428,18 +444,17 @@ def test_ami_attribute_user_permissions():
**REMOVE_USERS_ARGS).should_not.throw(EC2ResponseError)
@mock_ec2_deprecated
@mock_ec2
def test_ami_describe_executable_users():
conn = boto3.client('ec2', region_name='us-east-1')
ec2 = boto3.resource('ec2', 'us-east-1')
ec2.create_instances(ImageId='',
MinCount=1,
MaxCount=1)
response = conn.describe_instances(Filters=[{'Name': 'instance-state-name','Values': ['running']}])
response = conn.describe_instances(Filters=[{'Name': 'instance-state-name', 'Values': ['running']}])
instance_id = response['Reservations'][0]['Instances'][0]['InstanceId']
image_id = conn.create_image(InstanceId=instance_id,
Name='TestImage',)['ImageId']
Name='TestImage', )['ImageId']
USER1 = '123456789011'
@ -461,19 +476,18 @@ def test_ami_describe_executable_users():
images[0]['ImageId'].should.equal(image_id)
@mock_ec2_deprecated
@mock_ec2
def test_ami_describe_executable_users_negative():
conn = boto3.client('ec2', region_name='us-east-1')
ec2 = boto3.resource('ec2', 'us-east-1')
ec2.create_instances(ImageId='',
MinCount=1,
MaxCount=1)
response = conn.describe_instances(Filters=[{'Name': 'instance-state-name','Values': ['running']}])
response = conn.describe_instances(Filters=[{'Name': 'instance-state-name', 'Values': ['running']}])
instance_id = response['Reservations'][0]['Instances'][0]['InstanceId']
image_id = conn.create_image(InstanceId=instance_id,
Name='TestImage')['ImageId']
USER1 = '123456789011'
USER2 = '113355789012'
@ -482,6 +496,7 @@ def test_ami_describe_executable_users_negative():
'OperationType': 'add',
'UserIds': [USER1]}
# Add users and get no images
# Add users and get no images
conn.modify_image_attribute(**ADD_USER_ARGS)
@ -494,18 +509,17 @@ def test_ami_describe_executable_users_negative():
images.should.have.length_of(0)
@mock_ec2_deprecated
@mock_ec2
def test_ami_describe_executable_users_and_filter():
conn = boto3.client('ec2', region_name='us-east-1')
ec2 = boto3.resource('ec2', 'us-east-1')
ec2.create_instances(ImageId='',
MinCount=1,
MaxCount=1)
response = conn.describe_instances(Filters=[{'Name': 'instance-state-name','Values': ['running']}])
response = conn.describe_instances(Filters=[{'Name': 'instance-state-name', 'Values': ['running']}])
instance_id = response['Reservations'][0]['Instances'][0]['InstanceId']
image_id = conn.create_image(InstanceId=instance_id,
Name='ImageToDelete',)['ImageId']
Name='ImageToDelete', )['ImageId']
USER1 = '123456789011'
@ -682,11 +696,20 @@ def test_ami_describe_non_existent():
@mock_ec2
def test_ami_filter_wildcard():
ec2 = boto3.resource('ec2', region_name='us-west-1')
instance = ec2.create_instances(ImageId='ami-1234abcd', MinCount=1, MaxCount=1)[0]
image = instance.create_image(Name='test-image')
filter_result = list(ec2.images.filter(Owners=['111122223333'], Filters=[{'Name':'name', 'Values':['test*']}]))
assert filter_result == [image]
ec2_resource = boto3.resource('ec2', region_name='us-west-1')
ec2_client = boto3.client('ec2', region_name='us-west-1')
instance = ec2_resource.create_instances(ImageId='ami-1234abcd', MinCount=1, MaxCount=1)[0]
instance.create_image(Name='test-image')
# create an image with the same owner but will not match the filter
instance.create_image(Name='not-matching-image')
my_images = ec2_client.describe_images(
Owners=['111122223333'],
Filters=[{'Name': 'name', 'Values': ['test*']}]
)['Images']
my_images.should.have.length_of(1)
@mock_ec2
@ -706,16 +729,18 @@ def test_ami_filter_by_owner_id():
# Check we actually have a subset of images
assert len(ubuntu_ids) < len(all_ids)
@mock_ec2
def test_ami_filter_by_self():
client = boto3.client('ec2', region_name='us-east-1')
ec2_resource = boto3.resource('ec2', region_name='us-west-1')
ec2_client = boto3.client('ec2', region_name='us-west-1')
my_images = client.describe_images(Owners=['self'])
assert len(my_images) == 0
my_images = ec2_client.describe_images(Owners=['self'])['Images']
my_images.should.have.length_of(0)
# Create a new image
instance = ec2.create_instances(ImageId='ami-1234abcd', MinCount=1, MaxCount=1)[0]
image = instance.create_image(Name='test-image')
instance = ec2_resource.create_instances(ImageId='ami-1234abcd', MinCount=1, MaxCount=1)[0]
instance.create_image(Name='test-image')
my_images = client.describe_images(Owners=['self'])
assert len(my_images) == 1
my_images = ec2_client.describe_images(Owners=['self'])['Images']
my_images.should.have.length_of(1)