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

Error handling: Model-level validations, proper error responses.

Browse source
This commit is contained in:
Shawn Falkner-Horine 2014-08-25 10:54:47 -07:00
commit 071c03e216
25 changed files with 616 additions and 287 deletions
Download patch file Download diff file Expand all files Collapse all files

22
tests/test_ec2/test_amis.py
View file

@ -2,6 +2,7 @@ import boto
from boto.exception import EC2ResponseError
import sure # noqa
from nose.tools import assert_raises
from moto import mock_ec2
@ -19,7 +20,11 @@ def test_ami_create_and_delete():
success = conn.deregister_image(image)
success.should.be.true
success = conn.deregister_image.when.called_with(image).should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.deregister_image(image)
cm.exception.code.should.equal('InvalidAMIID.NotFound')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2
@ -46,7 +51,12 @@ def test_ami_tagging():
def test_ami_create_from_missing_instance():
conn = boto.connect_ec2('the_key', 'the_secret')
args = ["i-abcdefg", "test-ami", "this is a test ami"]
conn.create_image.when.called_with(*args).should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.create_image(*args)
cm.exception.code.should.equal('InvalidInstanceID.NotFound')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2
@ -64,4 +74,10 @@ def test_ami_pulls_attributes_from_instance():
@mock_ec2
def test_getting_missing_ami():
conn = boto.connect_ec2('the_key', 'the_secret')
conn.get_image.when.called_with('ami-missing').should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.get_image('ami-missing')
cm.exception.code.should.equal('InvalidAMIID.NotFound')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none

74
tests/test_ec2/test_dhcp_options.py
View file

@ -2,6 +2,7 @@ import boto
from boto.exception import EC2ResponseError
import sure # noqa
from nose.tools import assert_raises
from moto import mock_ec2
@ -26,7 +27,11 @@ def test_dhcp_options_associate_invalid_dhcp_id():
conn = boto.connect_vpc('the_key', 'the_secret')
vpc = conn.create_vpc("10.0.0.0/16")
conn.associate_dhcp_options.when.called_with("foo", vpc.id).should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.associate_dhcp_options("foo", vpc.id)
cm.exception.code.should.equal('InvalidDhcpOptionID.NotFound')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2
@ -35,7 +40,11 @@ def test_dhcp_options_associate_invalid_vpc_id():
conn = boto.connect_vpc('the_key', 'the_secret')
dhcp_options = conn.create_dhcp_options(SAMPLE_DOMAIN_NAME, SAMPLE_NAME_SERVERS)
conn.associate_dhcp_options.when.called_with(dhcp_options.id, "foo").should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.associate_dhcp_options(dhcp_options.id, "foo")
cm.exception.code.should.equal('InvalidVpcID.NotFound')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2
@ -49,11 +58,19 @@ def test_dhcp_options_delete_with_vpc():
rval = conn.associate_dhcp_options(dhcp_options_id, vpc.id)
rval.should.be.equal(True)
#conn.delete_dhcp_options(dhcp_options_id)
conn.delete_dhcp_options.when.called_with(dhcp_options_id).should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.delete_dhcp_options(dhcp_options_id)
cm.exception.code.should.equal('DependencyViolation')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
vpc.delete()
conn.get_all_dhcp_options.when.called_with([dhcp_options_id]).should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.get_all_dhcp_options([dhcp_options_id])
cm.exception.code.should.equal('InvalidDhcpOptionID.NotFound')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2
@ -72,8 +89,18 @@ def test_create_dhcp_options_invalid_options():
"""Create invalid dhcp options"""
conn = boto.connect_vpc('the_key', 'the_secret')
servers = ["f", "f", "f", "f", "f"]
conn.create_dhcp_options.when.called_with(ntp_servers=servers).should.throw(EC2ResponseError)
conn.create_dhcp_options.when.called_with(netbios_node_type="0").should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.create_dhcp_options(ntp_servers=servers)
cm.exception.code.should.equal('InvalidParameterValue')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
with assert_raises(EC2ResponseError) as cm:
conn.create_dhcp_options(netbios_node_type="0")
cm.exception.code.should.equal('InvalidParameterValue')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2
@ -94,7 +121,11 @@ def test_describe_dhcp_options_invalid_id():
"""get error on invalid dhcp_option_id lookup"""
conn = boto.connect_vpc('the_key', 'the_secret')
conn.get_all_dhcp_options.when.called_with(["1"]).should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.get_all_dhcp_options(["1"])
cm.exception.code.should.equal('InvalidDhcpOptionID.NotFound')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2
@ -107,7 +138,12 @@ def test_delete_dhcp_options():
dhcp_options.should.be.length_of(1)
conn.delete_dhcp_options(dhcp_option.id) # .should.be.equal(True)
conn.get_all_dhcp_options.when.called_with([dhcp_option.id]).should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.get_all_dhcp_options([dhcp_option.id])
cm.exception.code.should.equal('InvalidDhcpOptionID.NotFound')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2
@ -115,7 +151,25 @@ def test_delete_dhcp_options_invalid_id():
conn = boto.connect_vpc('the_key', 'the_secret')
conn.create_dhcp_options()
conn.delete_dhcp_options.when.called_with("1").should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.delete_dhcp_options("dopt-abcd1234")
cm.exception.code.should.equal('InvalidDhcpOptionID.NotFound')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2
def test_delete_dhcp_options_malformed_id():
conn = boto.connect_vpc('the_key', 'the_secret')
conn.create_dhcp_options()
with assert_raises(EC2ResponseError) as cm:
conn.delete_dhcp_options("foo-abcd1234")
cm.exception.code.should.equal('InvalidDhcpOptionsId.Malformed')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2

34
tests/test_ec2/test_elastic_block_store.py
View file

@ -1,6 +1,7 @@
import boto
from boto.exception import EC2ResponseError
import sure # noqa
from nose.tools import assert_raises
from moto import mock_ec2
from moto.ec2.models import ec2_backend
@ -22,7 +23,11 @@ def test_create_and_delete_volume():
conn.get_all_volumes().should.have.length_of(0)
# Deleting something that was already deleted should throw an error
volume.delete.when.called_with().should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
volume.delete()
cm.exception.code.should.equal('InvalidVolume.NotFound')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2
@ -47,14 +52,23 @@ def test_volume_attach_and_detach():
volume.update()
volume.volume_state().should.equal('available')
volume.attach.when.called_with(
'i-1234abcd', "/dev/sdh").should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm1:
volume.attach('i-1234abcd', "/dev/sdh")
cm1.exception.code.should.equal('InvalidInstanceID.NotFound')
cm1.exception.status.should.equal(400)
cm1.exception.request_id.should_not.be.none
conn.detach_volume.when.called_with(
volume.id, instance.id, "/dev/sdh").should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm2:
conn.detach_volume(volume.id, instance.id, "/dev/sdh")
cm2.exception.code.should.equal('InvalidAttachment.NotFound')
cm2.exception.status.should.equal(400)
cm2.exception.request_id.should_not.be.none
conn.detach_volume.when.called_with(
volume.id, 'i-1234abcd', "/dev/sdh").should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm3:
conn.detach_volume(volume.id, 'i-1234abcd', "/dev/sdh")
cm3.exception.code.should.equal('InvalidInstanceID.NotFound')
cm3.exception.status.should.equal(400)
cm3.exception.request_id.should_not.be.none
@mock_ec2
@ -76,7 +90,11 @@ def test_create_snapshot():
conn.get_all_snapshots().should.have.length_of(1)
# Deleting something that was already deleted should throw an error
snapshot.delete.when.called_with().should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
snapshot.delete()
cm.exception.code.should.equal('InvalidSnapshot.NotFound')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2

76
tests/test_ec2/test_elastic_ip_addresses.py
View file

@ -3,6 +3,7 @@ import boto
from boto.exception import EC2ResponseError
import sure # noqa
from nose.tools import assert_raises
from moto import mock_ec2
@ -41,7 +42,11 @@ def test_eip_allocate_invalid_domain():
"""Allocate EIP invalid domain"""
conn = boto.connect_ec2('the_key', 'the_secret')
conn.allocate_address.when.called_with(domain="bogus").should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.allocate_address(domain="bogus")
cm.exception.code.should.equal('InvalidParameterValue')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2
@ -54,7 +59,13 @@ def test_eip_associate_classic():
eip = conn.allocate_address()
eip.instance_id.should.be.none
conn.associate_address.when.called_with(public_ip=eip.public_ip).should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.associate_address(public_ip=eip.public_ip)
cm.exception.code.should.equal('MissingParameter')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
conn.associate_address(instance_id=instance.id, public_ip=eip.public_ip)
eip = conn.get_all_addresses(addresses=[eip.public_ip])[0] # no .update() on address ):
eip.instance_id.should.be.equal(instance.id)
@ -77,7 +88,13 @@ def test_eip_associate_vpc():
eip = conn.allocate_address(domain='vpc')
eip.instance_id.should.be.none
conn.associate_address.when.called_with(allocation_id=eip.allocation_id).should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.associate_address(allocation_id=eip.allocation_id)
cm.exception.code.should.equal('MissingParameter')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
conn.associate_address(instance_id=instance.id, allocation_id=eip.allocation_id)
eip = conn.get_all_addresses(addresses=[eip.public_ip])[0] # no .update() on address ):
eip.instance_id.should.be.equal(instance.id)
@ -100,8 +117,15 @@ def test_eip_reassociate():
eip = conn.allocate_address()
conn.associate_address(instance_id=instance.id, public_ip=eip.public_ip)
conn.associate_address.when.called_with(instance_id=instance.id, public_ip=eip.public_ip, allow_reassociation=False).should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.associate_address(instance_id=instance.id, public_ip=eip.public_ip, allow_reassociation=False)
cm.exception.code.should.equal('Resource.AlreadyAssociated')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
conn.associate_address.when.called_with(instance_id=instance.id, public_ip=eip.public_ip, allow_reassociation=True).should_not.throw(EC2ResponseError)
eip.release()
eip = None
@ -116,7 +140,12 @@ def test_eip_associate_invalid_args():
instance = reservation.instances[0]
eip = conn.allocate_address()
conn.associate_address.when.called_with(instance_id=instance.id).should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.associate_address(instance_id=instance.id)
cm.exception.code.should.equal('MissingParameter')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
instance.terminate()
@ -125,27 +154,47 @@ def test_eip_associate_invalid_args():
def test_eip_disassociate_bogus_association():
"""Disassociate bogus EIP"""
conn = boto.connect_ec2('the_key', 'the_secret')
conn.disassociate_address.when.called_with(association_id="bogus").should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.disassociate_address(association_id="bogus")
cm.exception.code.should.equal('InvalidAssociationID.NotFound')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2
def test_eip_release_bogus_eip():
"""Release bogus EIP"""
conn = boto.connect_ec2('the_key', 'the_secret')
conn.release_address.when.called_with(allocation_id="bogus").should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.release_address(allocation_id="bogus")
cm.exception.code.should.equal('InvalidAllocationID.NotFound')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2
def test_eip_disassociate_arg_error():
"""Invalid arguments disassociate address"""
conn = boto.connect_ec2('the_key', 'the_secret')
conn.disassociate_address.when.called_with().should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.disassociate_address()
cm.exception.code.should.equal('MissingParameter')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2
def test_eip_release_arg_error():
"""Invalid arguments release address"""
conn = boto.connect_ec2('the_key', 'the_secret')
conn.release_address.when.called_with().should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.release_address()
cm.exception.code.should.equal('MissingParameter')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2
@ -186,9 +235,12 @@ def test_eip_describe():
@mock_ec2
def test_eip_describe_none():
"""Find nothing when seach for bogus IP"""
"""Error when search for bogus IP"""
conn = boto.connect_ec2('the_key', 'the_secret')
lookup_addresses = conn.get_all_addresses(addresses=["256.256.256.256"])
len(lookup_addresses).should.be.equal(0)
with assert_raises(EC2ResponseError) as cm:
conn.get_all_addresses(addresses=["256.256.256.256"])
cm.exception.code.should.equal('InvalidAddress.NotFound')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none

8
tests/test_ec2/test_general.py
View file

@ -1,6 +1,7 @@
import boto
from boto.exception import EC2ResponseError
import sure # noqa
from nose.tools import assert_raises
from moto import mock_ec2
@ -18,4 +19,9 @@ def test_console_output():
@mock_ec2
def test_console_output_without_instance():
conn = boto.connect_ec2('the_key', 'the_secret')
conn.get_console_output.when.called_with('i-1234abcd').should.throw(Exception)
with assert_raises(EC2ResponseError) as cm:
conn.get_console_output('i-1234abcd')
cm.exception.code.should.equal('InvalidInstanceID.NotFound')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none

10
tests/test_ec2/test_instances.py
View file

@ -4,6 +4,7 @@ import boto
from boto.ec2.instance import Reservation, InstanceAttribute
from boto.exception import EC2ResponseError
import sure # noqa
from nose.tools import assert_raises
from moto import mock_ec2
@ -72,10 +73,11 @@ def test_get_instances_by_id():
instance_ids.should.equal([instance1.id, instance2.id])
# Call get_all_instances with a bad id should raise an error
conn.get_all_instances.when.called_with(instance_ids=[instance1.id, "i-1234abcd"]).should.throw(
EC2ResponseError,
"The instance ID 'i-1234abcd' does not exist"
)
with assert_raises(EC2ResponseError) as cm:
conn.get_all_instances(instance_ids=[instance1.id, "i-1234abcd"])
cm.exception.code.should.equal('InvalidInstanceID.NotFound')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2

62
tests/test_ec2/test_internet_gateways.py
View file

@ -4,6 +4,7 @@ import boto
from boto.exception import EC2ResponseError
import sure # noqa
from nose.tools import assert_raises
from moto import mock_ec2
@ -41,7 +42,12 @@ def test_igw_attach_bad_vpc():
""" internet gateway fail to attach w/ bad vpc """
conn = boto.connect_vpc('the_key', 'the_secret')
igw = conn.create_internet_gateway()
conn.attach_internet_gateway.when.called_with(igw.id, BAD_VPC).should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.attach_internet_gateway(igw.id, BAD_VPC)
cm.exception.code.should.equal('InvalidVpcID.NotFound')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2
def test_igw_attach_twice():
@ -51,7 +57,12 @@ def test_igw_attach_twice():
vpc1 = conn.create_vpc(VPC_CIDR)
vpc2 = conn.create_vpc(VPC_CIDR)
conn.attach_internet_gateway(igw.id, vpc1.id)
conn.attach_internet_gateway.when.called_with(igw.id, vpc2.id).should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.attach_internet_gateway(igw.id, vpc2.id)
cm.exception.code.should.equal('Resource.AlreadyAssociated')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2
def test_igw_detach():
@ -65,20 +76,46 @@ def test_igw_detach():
igw.attachments.should.have.length_of(0)
@mock_ec2
def test_igw_detach_bad_vpc():
""" internet gateway fail to detach w/ bad vpc """
def test_igw_detach_wrong_vpc():
""" internet gateway fail to detach w/ wrong vpc """
conn = boto.connect_vpc('the_key', 'the_secret')
igw = conn.create_internet_gateway()
vpc1 = conn.create_vpc(VPC_CIDR)
vpc2 = conn.create_vpc(VPC_CIDR)
conn.attach_internet_gateway(igw.id, vpc1.id)
with assert_raises(EC2ResponseError) as cm:
conn.detach_internet_gateway(igw.id, vpc2.id)
cm.exception.code.should.equal('Gateway.NotAttached')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2
def test_igw_detach_invalid_vpc():
""" internet gateway fail to detach w/ invalid vpc """
conn = boto.connect_vpc('the_key', 'the_secret')
igw = conn.create_internet_gateway()
vpc = conn.create_vpc(VPC_CIDR)
conn.attach_internet_gateway(igw.id, vpc.id)
conn.detach_internet_gateway.when.called_with(igw.id, BAD_VPC).should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.detach_internet_gateway(igw.id, BAD_VPC)
cm.exception.code.should.equal('Gateway.NotAttached')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2
def test_igw_detach_unattached():
""" internet gateway fail to detach unattached """
conn = boto.connect_vpc('the_key', 'the_secret')
igw = conn.create_internet_gateway()
conn.detach_internet_gateway.when.called_with(igw.id, BAD_VPC).should.throw(EC2ResponseError)
vpc = conn.create_vpc(VPC_CIDR)
with assert_raises(EC2ResponseError) as cm:
conn.detach_internet_gateway(igw.id, vpc.id)
cm.exception.code.should.equal('Gateway.NotAttached')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2
def test_igw_delete():
@ -98,7 +135,12 @@ def test_igw_delete_attached():
igw = conn.create_internet_gateway()
vpc = conn.create_vpc(VPC_CIDR)
conn.attach_internet_gateway(igw.id, vpc.id)
conn.delete_internet_gateway.when.called_with(igw.id).should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.delete_internet_gateway(igw.id)
cm.exception.code.should.equal('DependencyViolation')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2
def test_igw_desribe():
@ -112,4 +154,8 @@ def test_igw_desribe():
def test_igw_desribe_bad_id():
""" internet gateway fail to fetch by bad id """
conn = boto.connect_vpc('the_key', 'the_secret')
conn.get_all_internet_gateways.when.called_with([BAD_IGW]).should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.get_all_internet_gateways([BAD_IGW])
cm.exception.code.should.equal('InvalidInternetGatewayID.NotFound')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none

22
tests/test_ec2/test_key_pairs.py
View file

@ -1,5 +1,6 @@
import boto
import sure # noqa
from nose.tools import assert_raises
from boto.exception import EC2ResponseError
from moto import mock_ec2
@ -11,6 +12,17 @@ def test_key_pairs_empty():
assert len(conn.get_all_key_pairs()) == 0
@mock_ec2
def test_key_pairs_invalid_id():
conn = boto.connect_ec2('the_key', 'the_secret')
with assert_raises(EC2ResponseError) as cm:
conn.get_all_key_pairs('foo')
cm.exception.code.should.equal('InvalidKeyPair.NotFound')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2
def test_key_pairs_create():
conn = boto.connect_ec2('the_key', 'the_secret')
@ -42,10 +54,12 @@ def test_key_pairs_create_exist():
kp = conn.create_key_pair('foo')
assert kp.material.startswith('---- BEGIN RSA PRIVATE KEY ----')
assert len(conn.get_all_key_pairs()) == 1
conn.create_key_pair.when.called_with('foo').should.throw(
EC2ResponseError,
"The keypair 'foo' already exists."
)
with assert_raises(EC2ResponseError) as cm:
conn.create_key_pair('foo')
cm.exception.code.should.equal('InvalidKeyPair.Duplicate')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2

38
tests/test_ec2/test_security_groups.py
View file

@ -1,6 +1,7 @@
import boto
from boto.exception import EC2ResponseError
import sure # noqa
from nose.tools import assert_raises
from moto import mock_ec2
@ -14,7 +15,11 @@ def test_create_and_describe_security_group():
security_group.description.should.equal('this is a test security group')
# Trying to create another group with the same name should throw an error
conn.create_security_group.when.called_with('test security group', 'this is a test security group').should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.create_security_group('test security group', 'this is a test security group')
cm.exception.code.should.equal('InvalidGroup.Duplicate')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
all_groups = conn.get_all_security_groups()
all_groups.should.have.length_of(1)
@ -24,7 +29,12 @@ def test_create_and_describe_security_group():
@mock_ec2
def test_create_security_group_without_description_raises_error():
conn = boto.connect_ec2('the_key', 'the_secret')
conn.create_security_group.when.called_with('test security group', '').should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.create_security_group('test security group', '')
cm.exception.code.should.equal('MissingParameter')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2
@ -39,7 +49,11 @@ def test_create_and_describe_vpc_security_group():
security_group.description.should.equal('this is a test security group')
# Trying to create another group with the same name in the same VPC should throw an error
conn.create_security_group.when.called_with('test security group', 'this is a test security group', vpc_id).should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.create_security_group('test security group', 'this is a test security group', vpc_id)
cm.exception.code.should.equal('InvalidGroup.Duplicate')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
all_groups = conn.get_all_security_groups()
@ -74,7 +88,11 @@ def test_deleting_security_groups():
conn.get_all_security_groups().should.have.length_of(2)
# Deleting a group that doesn't exist should throw an error
conn.delete_security_group.when.called_with('foobar').should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.delete_security_group('foobar')
cm.exception.code.should.equal('InvalidGroup.NotFound')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
# Delete by name
conn.delete_security_group('test2')
@ -108,7 +126,11 @@ def test_authorize_ip_range_and_revoke():
security_group.rules[0].grants[0].cidr_ip.should.equal("123.123.123.123/32")
# Wrong Cidr should throw error
security_group.revoke.when.called_with(ip_protocol="tcp", from_port="22", to_port="2222", cidr_ip="123.123.123.122/32").should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
security_group.revoke(ip_protocol="tcp", from_port="22", to_port="2222", cidr_ip="123.123.123.122/32")
cm.exception.code.should.equal('InvalidPermission.NotFound')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
# Actually revoke
security_group.revoke(ip_protocol="tcp", from_port="22", to_port="2222", cidr_ip="123.123.123.123/32")
@ -132,7 +154,11 @@ def test_authorize_other_group_and_revoke():
security_group.rules[0].grants[0].group_id.should.equal(other_security_group.id)
# Wrong source group should throw error
security_group.revoke.when.called_with(ip_protocol="tcp", from_port="22", to_port="2222", src_group=wrong_group).should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
security_group.revoke(ip_protocol="tcp", from_port="22", to_port="2222", src_group=wrong_group)
cm.exception.code.should.equal('InvalidPermission.NotFound')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
# Actually revoke
security_group.revoke(ip_protocol="tcp", from_port="22", to_port="2222", src_group=other_security_group)

8
tests/test_ec2/test_subnets.py
View file

@ -1,6 +1,7 @@
import boto
from boto.exception import EC2ResponseError
import sure # noqa
from nose.tools import assert_raises
from moto import mock_ec2
@ -19,8 +20,11 @@ def test_subnets():
all_subnets = conn.get_all_subnets()
all_subnets.should.have.length_of(0)
conn.delete_subnet.when.called_with(
subnet.id).should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.delete_subnet(subnet.id)
cm.exception.code.should.equal('InvalidSubnetID.NotFound')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2

22
tests/test_ec2/test_vpc_peering.py
View file

@ -1,6 +1,7 @@
import boto
from boto.exception import EC2ResponseError
import sure # noqa
from nose.tools import assert_raises
from moto import mock_ec2
from tests.helpers import requires_boto_gte
@ -40,8 +41,11 @@ def test_vpc_peering_connections_accept():
vpc_pcx = conn.accept_vpc_peering_connection(vpc_pcx.id)
vpc_pcx._status.code.should.equal('active')
conn.reject_vpc_peering_connection.when.called_with(
vpc_pcx.id).should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.reject_vpc_peering_connection(vpc_pcx.id)
cm.exception.code.should.equal('InvalidStateTransition')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
all_vpc_pcxs = conn.get_all_vpc_peering_connections()
all_vpc_pcxs.should.have.length_of(1)
@ -57,8 +61,11 @@ def test_vpc_peering_connections_reject():
verdict = conn.reject_vpc_peering_connection(vpc_pcx.id)
verdict.should.equal(True)
conn.accept_vpc_peering_connection.when.called_with(
vpc_pcx.id).should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.accept_vpc_peering_connection(vpc_pcx.id)
cm.exception.code.should.equal('InvalidStateTransition')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
all_vpc_pcxs = conn.get_all_vpc_peering_connections()
all_vpc_pcxs.should.have.length_of(1)
@ -77,6 +84,9 @@ def test_vpc_peering_connections_delete():
all_vpc_pcxs = conn.get_all_vpc_peering_connections()
all_vpc_pcxs.should.have.length_of(0)
conn.delete_vpc_peering_connection.when.called_with(
"pcx-1234abcd").should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.delete_vpc_peering_connection("pcx-1234abcd")
cm.exception.code.should.equal('InvalidVPCPeeringConnectionId.NotFound')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none

8
tests/test_ec2/test_vpcs.py
View file

@ -1,6 +1,7 @@
import boto
from boto.exception import EC2ResponseError
import sure # noqa
from nose.tools import assert_raises
from moto import mock_ec2
@ -19,8 +20,11 @@ def test_vpcs():
all_vpcs = conn.get_all_vpcs()
all_vpcs.should.have.length_of(0)
conn.delete_vpc.when.called_with(
"vpc-1234abcd").should.throw(EC2ResponseError)
with assert_raises(EC2ResponseError) as cm:
conn.delete_vpc("vpc-1234abcd")
cm.exception.code.should.equal('InvalidVpcID.NotFound')
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
@mock_ec2