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

Port test suite from nose to pytest.

Browse source 
This just eliminates all errors on the tests collection. Elimination of
failures is left to the next commit.
This commit is contained in:
Matěj Cepl 2020-10-06 07:54:49 +02:00
commit 77dc60ea97
146 changed files with 1172 additions and 1277 deletions
Download patch file Download diff file Expand all files Collapse all files

42
tests/test_ec2/test_amis.py
View file

@ -7,7 +7,7 @@ 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 pytest
import sure # noqa
from moto import mock_ec2_deprecated, mock_ec2
@ -27,7 +27,7 @@ def test_ami_create_and_delete():
reservation = conn.run_instances("ami-1234abcd")
instance = reservation.instances[0]
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
image_id = conn.create_image(
instance.id, "test-ami", "this is a test ami", dry_run=True
)
@ -76,7 +76,7 @@ def test_ami_create_and_delete():
root_mapping.should_not.be.none
# Deregister
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
success = conn.deregister_image(image_id, dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -87,7 +87,7 @@ def test_ami_create_and_delete():
success = conn.deregister_image(image_id)
success.should.be.true
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.deregister_image(image_id)
cm.exception.code.should.equal("InvalidAMIID.NotFound")
cm.exception.status.should.equal(400)
@ -112,7 +112,7 @@ def test_ami_copy():
# Boto returns a 'CopyImage' object with an image_id attribute here. Use
# the image_id to fetch the full info.
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
copy_image_ref = conn.copy_image(
source_image.region.name,
source_image.id,
@ -152,7 +152,7 @@ def test_ami_copy():
)
# Copy from non-existent source ID.
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.copy_image(
source_image.region.name,
"ami-abcd1234",
@ -164,7 +164,7 @@ def test_ami_copy():
cm.exception.request_id.should_not.be.none
# Copy from non-existent source region.
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
invalid_region = (
"us-east-1" if (source_image.region.name != "us-east-1") else "us-west-1"
)
@ -208,7 +208,7 @@ def test_ami_tagging():
conn.create_image(instance.id, "test-ami", "this is a test ami")
image = conn.get_all_images()[0]
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
image.add_tag("a key", "some value", dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -233,7 +233,7 @@ 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"]
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.create_image(*args)
cm.exception.code.should.equal("InvalidInstanceID.NotFound")
cm.exception.status.should.equal(400)
@ -353,7 +353,7 @@ def test_ami_filtering_via_tag():
def test_getting_missing_ami():
conn = boto.connect_ec2("the_key", "the_secret")
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.get_image("ami-missing")
cm.exception.code.should.equal("InvalidAMIID.NotFound")
cm.exception.status.should.equal(400)
@ -364,7 +364,7 @@ def test_getting_missing_ami():
def test_getting_malformed_ami():
conn = boto.connect_ec2("the_key", "the_secret")
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.get_image("foo-missing")
cm.exception.code.should.equal("InvalidAMIID.Malformed")
cm.exception.status.should.equal(400)
@ -399,7 +399,7 @@ def test_ami_attribute_group_permissions():
}
# Add 'all' group and confirm
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
conn.modify_image_attribute(**dict(ADD_GROUP_ARGS, **{"dry_run": True}))
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -678,7 +678,7 @@ def test_ami_attribute_error_cases():
image = conn.get_image(image_id)
# Error: Add with group != 'all'
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.modify_image_attribute(
image.id, attribute="launchPermission", operation="add", groups="everyone"
)
@ -687,7 +687,7 @@ def test_ami_attribute_error_cases():
cm.exception.request_id.should_not.be.none
# Error: Add with user ID that isn't an integer.
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.modify_image_attribute(
image.id,
attribute="launchPermission",
@ -699,7 +699,7 @@ def test_ami_attribute_error_cases():
cm.exception.request_id.should_not.be.none
# Error: Add with user ID that is > length 12.
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.modify_image_attribute(
image.id,
attribute="launchPermission",
@ -711,7 +711,7 @@ def test_ami_attribute_error_cases():
cm.exception.request_id.should_not.be.none
# Error: Add with user ID that is < length 12.
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.modify_image_attribute(
image.id,
attribute="launchPermission",
@ -724,7 +724,7 @@ def test_ami_attribute_error_cases():
# Error: Add with one invalid user ID among other valid IDs, ensure no
# partial changes.
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.modify_image_attribute(
image.id,
attribute="launchPermission",
@ -739,7 +739,7 @@ def test_ami_attribute_error_cases():
attributes.attrs.should.have.length_of(0)
# Error: Add with invalid image ID
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.modify_image_attribute(
"ami-abcd1234", attribute="launchPermission", operation="add", groups="all"
)
@ -748,7 +748,7 @@ def test_ami_attribute_error_cases():
cm.exception.request_id.should_not.be.none
# Error: Remove with invalid image ID
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.modify_image_attribute(
"ami-abcd1234",
attribute="launchPermission",
@ -765,11 +765,11 @@ def test_ami_describe_non_existent():
ec2 = boto3.resource("ec2", region_name="us-west-1")
# Valid pattern but non-existent id
img = ec2.Image("ami-abcd1234")
with assert_raises(ClientError):
with pytest.raises(ClientError):
img.load()
# Invalid ami pattern
img = ec2.Image("not_an_ami_id")
with assert_raises(ClientError):
with pytest.raises(ClientError):
img.load()

5
tests/test_ec2/test_customer_gateways.py
View file

@ -1,8 +1,7 @@
from __future__ import unicode_literals
import boto
import sure # noqa
from nose.tools import assert_raises
from nose.tools import assert_false
import pytest
from boto.exception import EC2ResponseError
from moto import mock_ec2_deprecated
@ -45,5 +44,5 @@ def test_delete_customer_gateways():
@mock_ec2_deprecated
def test_delete_customer_gateways_bad_id():
conn = boto.connect_vpc("the_key", "the_secret")
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.delete_customer_gateway("cgw-0123abcd")

23
tests/test_ec2/test_dhcp_options.py
View file

@ -1,8 +1,7 @@
from __future__ import unicode_literals
# Ensure 'assert_raises' context manager support for Python 2.6
import tests.backport_assert_raises
from nose.tools import assert_raises
import pytest
import boto3
import boto
@ -33,7 +32,7 @@ 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")
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.associate_dhcp_options("foo", vpc.id)
cm.exception.code.should.equal("InvalidDhcpOptionID.NotFound")
cm.exception.status.should.equal(400)
@ -46,7 +45,7 @@ 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)
with assert_raises(EC2ResponseError) as cm:
with pytest.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)
@ -64,7 +63,7 @@ def test_dhcp_options_delete_with_vpc():
rval = conn.associate_dhcp_options(dhcp_options_id, vpc.id)
rval.should.be.equal(True)
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.delete_dhcp_options(dhcp_options_id)
cm.exception.code.should.equal("DependencyViolation")
cm.exception.status.should.equal(400)
@ -72,7 +71,7 @@ def test_dhcp_options_delete_with_vpc():
vpc.delete()
with assert_raises(EC2ResponseError) as cm:
with pytest.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)
@ -100,13 +99,13 @@ def test_create_dhcp_options_invalid_options():
conn = boto.connect_vpc("the_key", "the_secret")
servers = ["f", "f", "f", "f", "f"]
with assert_raises(EC2ResponseError) as cm:
with pytest.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:
with pytest.raises(EC2ResponseError) as cm:
conn.create_dhcp_options(netbios_node_type="0")
cm.exception.code.should.equal("InvalidParameterValue")
cm.exception.status.should.equal(400)
@ -131,7 +130,7 @@ def test_describe_dhcp_options_invalid_id():
"""get error on invalid dhcp_option_id lookup"""
conn = boto.connect_vpc("the_key", "the_secret")
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.get_all_dhcp_options(["1"])
cm.exception.code.should.equal("InvalidDhcpOptionID.NotFound")
cm.exception.status.should.equal(400)
@ -149,7 +148,7 @@ def test_delete_dhcp_options():
conn.delete_dhcp_options(dhcp_option.id) # .should.be.equal(True)
with assert_raises(EC2ResponseError) as cm:
with pytest.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)
@ -162,7 +161,7 @@ def test_delete_dhcp_options_invalid_id():
conn.create_dhcp_options()
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.delete_dhcp_options("dopt-abcd1234")
cm.exception.code.should.equal("InvalidDhcpOptionID.NotFound")
cm.exception.status.should.equal(400)
@ -175,7 +174,7 @@ def test_delete_dhcp_options_malformed_id():
conn.create_dhcp_options()
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.delete_dhcp_options("foo-abcd1234")
cm.exception.code.should.equal("InvalidDhcpOptionsId.Malformed")
cm.exception.status.should.equal(400)

69
tests/test_ec2/test_elastic_block_store.py
View file

@ -1,8 +1,7 @@
from __future__ import unicode_literals
# Ensure 'assert_raises' context manager support for Python 2.6
import tests.backport_assert_raises
from nose.tools import assert_raises
import pytest
from moto.ec2 import ec2_backends
import boto
@ -31,7 +30,7 @@ def test_create_and_delete_volume():
volume = current_volume[0]
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
volume.delete(dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -46,7 +45,7 @@ def test_create_and_delete_volume():
my_volume.should.have.length_of(0)
# Deleting something that was already deleted should throw an error
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
volume.delete()
cm.exception.code.should.equal("InvalidVolume.NotFound")
cm.exception.status.should.equal(400)
@ -95,7 +94,7 @@ def test_delete_attached_volume():
@mock_ec2_deprecated
def test_create_encrypted_volume_dryrun():
conn = boto.ec2.connect_to_region("us-east-1")
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
conn.create_volume(80, "us-east-1a", encrypted=True, dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -109,7 +108,7 @@ def test_create_encrypted_volume():
conn = boto.ec2.connect_to_region("us-east-1")
volume = conn.create_volume(80, "us-east-1a", encrypted=True)
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
conn.create_volume(80, "us-east-1a", encrypted=True, dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -134,7 +133,7 @@ def test_filter_volume_by_id():
vol2 = conn.get_all_volumes(volume_ids=[volume1.id, volume2.id])
vol2.should.have.length_of(2)
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.get_all_volumes(volume_ids=["vol-does_not_exist"])
cm.exception.code.should.equal("InvalidVolume.NotFound")
cm.exception.status.should.equal(400)
@ -259,7 +258,7 @@ def test_volume_attach_and_detach():
volume.update()
volume.volume_state().should.equal("available")
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
volume.attach(instance.id, "/dev/sdh", dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -275,7 +274,7 @@ def test_volume_attach_and_detach():
volume.attach_data.instance_id.should.equal(instance.id)
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
volume.detach(dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -288,19 +287,19 @@ def test_volume_attach_and_detach():
volume.update()
volume.volume_state().should.equal("available")
with assert_raises(EC2ResponseError) as cm1:
with pytest.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
with assert_raises(EC2ResponseError) as cm2:
with pytest.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
with assert_raises(EC2ResponseError) as cm3:
with pytest.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)
@ -312,7 +311,7 @@ def test_create_snapshot():
conn = boto.ec2.connect_to_region("us-east-1")
volume = conn.create_volume(80, "us-east-1a")
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
snapshot = volume.create_snapshot("a dryrun snapshot", dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -340,7 +339,7 @@ def test_create_snapshot():
conn.get_all_snapshots().should.have.length_of(num_snapshots)
# Deleting something that was already deleted should throw an error
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
snapshot.delete()
cm.exception.code.should.equal("InvalidSnapshot.NotFound")
cm.exception.status.should.equal(400)
@ -382,7 +381,7 @@ def test_filter_snapshot_by_id():
s.volume_id.should.be.within([volume2.id, volume3.id])
s.region.name.should.equal(conn.region.name)
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.get_all_snapshots(snapshot_ids=["snap-does_not_exist"])
cm.exception.code.should.equal("InvalidSnapshot.NotFound")
cm.exception.status.should.equal(400)
@ -484,7 +483,7 @@ def test_snapshot_attribute():
# Add 'all' group and confirm
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
conn.modify_snapshot_attribute(**dict(ADD_GROUP_ARGS, **{"dry_run": True}))
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -506,7 +505,7 @@ def test_snapshot_attribute():
)
# Remove 'all' group and confirm
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
conn.modify_snapshot_attribute(**dict(REMOVE_GROUP_ARGS, **{"dry_run": True}))
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -527,7 +526,7 @@ def test_snapshot_attribute():
).should_not.throw(EC2ResponseError)
# Error: Add with group != 'all'
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.modify_snapshot_attribute(
snapshot.id,
attribute="createVolumePermission",
@ -539,7 +538,7 @@ def test_snapshot_attribute():
cm.exception.request_id.should_not.be.none
# Error: Add with invalid snapshot ID
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.modify_snapshot_attribute(
"snapshot-abcd1234",
attribute="createVolumePermission",
@ -551,7 +550,7 @@ def test_snapshot_attribute():
cm.exception.request_id.should_not.be.none
# Error: Remove with invalid snapshot ID
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.modify_snapshot_attribute(
"snapshot-abcd1234",
attribute="createVolumePermission",
@ -740,7 +739,7 @@ def test_create_volume_from_snapshot():
volume = conn.create_volume(80, "us-east-1a")
snapshot = volume.create_snapshot("a test snapshot")
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
snapshot = volume.create_snapshot("a test snapshot", dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -786,7 +785,7 @@ def test_modify_attribute_blockDeviceMapping():
instance = reservation.instances[0]
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
instance.modify_attribute(
"blockDeviceMapping", {"/dev/sda1": True}, dry_run=True
)
@ -809,7 +808,7 @@ def test_volume_tag_escaping():
vol = conn.create_volume(10, "us-east-1a")
snapshot = conn.create_snapshot(vol.id, "Desc")
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
snapshot.add_tags({"key": "</closed>"}, dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -879,25 +878,25 @@ def test_copy_snapshot():
getattr(source, attrib).should.equal(getattr(dest, attrib))
# Copy from non-existent source ID.
with assert_raises(ClientError) as cm:
with pytest.raises(ClientError) as cm:
create_snapshot_error = ec2_client.create_snapshot(VolumeId="vol-abcd1234")
cm.exception.response["Error"]["Code"].should.equal("InvalidVolume.NotFound")
cm.exception.response["Error"]["Message"].should.equal(
"The volume 'vol-abcd1234' does not exist."
)
cm.exception.response["ResponseMetadata"]["RequestId"].should_not.be.none
cm.exception.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
cm.exception.response["Error"]["Code"].should.equal("InvalidVolume.NotFound")
cm.exception.response["Error"]["Message"].should.equal(
"The volume 'vol-abcd1234' does not exist."
)
cm.exception.response["ResponseMetadata"]["RequestId"].should_not.be.none
cm.exception.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
# Copy from non-existent source region.
with assert_raises(ClientError) as cm:
with pytest.raises(ClientError) as cm:
copy_snapshot_response = dest_ec2_client.copy_snapshot(
SourceSnapshotId=create_snapshot_response["SnapshotId"],
SourceRegion="eu-west-2",
)
cm.exception.response["Error"]["Code"].should.equal("InvalidSnapshot.NotFound")
cm.exception.response["Error"]["Message"].should.be.none
cm.exception.response["ResponseMetadata"]["RequestId"].should_not.be.none
cm.exception.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
cm.exception.response["Error"]["Code"].should.equal("InvalidSnapshot.NotFound")
cm.exception.response["Error"]["Message"].should.be.none
cm.exception.response["ResponseMetadata"]["RequestId"].should_not.be.none
cm.exception.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
@mock_ec2

39
tests/test_ec2/test_elastic_ip_addresses.py
View file

@ -1,8 +1,7 @@
from __future__ import unicode_literals
# Ensure 'assert_raises' context manager support for Python 2.6
import tests.backport_assert_raises
from nose.tools import assert_raises
import pytest
import boto
import boto3
@ -21,7 +20,7 @@ def test_eip_allocate_classic():
"""Allocate/release Classic EIP"""
conn = boto.connect_ec2("the_key", "the_secret")
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
standard = conn.allocate_address(dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -35,7 +34,7 @@ def test_eip_allocate_classic():
standard.instance_id.should.be.none
standard.domain.should.be.equal("standard")
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
standard.release(dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -52,7 +51,7 @@ def test_eip_allocate_vpc():
"""Allocate/release VPC EIP"""
conn = boto.connect_ec2("the_key", "the_secret")
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
vpc = conn.allocate_address(domain="vpc", dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -84,7 +83,7 @@ def test_eip_allocate_invalid_domain():
"""Allocate EIP invalid domain"""
conn = boto.connect_ec2("the_key", "the_secret")
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.allocate_address(domain="bogus")
cm.exception.code.should.equal("InvalidParameterValue")
cm.exception.status.should.equal(400)
@ -102,13 +101,13 @@ def test_eip_associate_classic():
eip = conn.allocate_address()
eip.instance_id.should.be.none
with assert_raises(EC2ResponseError) as cm:
with pytest.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
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
conn.associate_address(
instance_id=instance.id, public_ip=eip.public_ip, dry_run=True
)
@ -123,7 +122,7 @@ def test_eip_associate_classic():
eip = conn.get_all_addresses(addresses=[eip.public_ip])[0]
eip.instance_id.should.be.equal(instance.id)
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
conn.disassociate_address(public_ip=eip.public_ip, dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -153,7 +152,7 @@ def test_eip_associate_vpc():
eip = conn.allocate_address(domain="vpc")
eip.instance_id.should.be.none
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.associate_address(allocation_id=eip.allocation_id)
cm.exception.code.should.equal("MissingParameter")
cm.exception.status.should.equal(400)
@ -169,7 +168,7 @@ def test_eip_associate_vpc():
eip.instance_id.should.be.equal("")
eip.association_id.should.be.none
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
eip.release(dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -241,7 +240,7 @@ def test_eip_associate_network_interface():
eip = conn.allocate_address(domain="vpc")
eip.network_interface_id.should.be.none
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.associate_address(network_interface_id=eni.id)
cm.exception.code.should.equal("MissingParameter")
cm.exception.status.should.equal(400)
@ -276,7 +275,7 @@ def test_eip_reassociate():
conn.associate_address(instance_id=instance1.id, public_ip=eip.public_ip)
# Different ID detects resource association
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.associate_address(
instance_id=instance2.id, public_ip=eip.public_ip, allow_reassociation=False
)
@ -312,7 +311,7 @@ def test_eip_reassociate_nic():
conn.associate_address(network_interface_id=eni1.id, public_ip=eip.public_ip)
# Different ID detects resource association
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.associate_address(network_interface_id=eni2.id, public_ip=eip.public_ip)
cm.exception.code.should.equal("Resource.AlreadyAssociated")
cm.exception.status.should.equal(400)
@ -336,7 +335,7 @@ def test_eip_associate_invalid_args():
eip = conn.allocate_address()
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.associate_address(instance_id=instance.id)
cm.exception.code.should.equal("MissingParameter")
cm.exception.status.should.equal(400)
@ -350,7 +349,7 @@ def test_eip_disassociate_bogus_association():
"""Disassociate bogus EIP"""
conn = boto.connect_ec2("the_key", "the_secret")
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.disassociate_address(association_id="bogus")
cm.exception.code.should.equal("InvalidAssociationID.NotFound")
cm.exception.status.should.equal(400)
@ -362,7 +361,7 @@ def test_eip_release_bogus_eip():
"""Release bogus EIP"""
conn = boto.connect_ec2("the_key", "the_secret")
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.release_address(allocation_id="bogus")
cm.exception.code.should.equal("InvalidAllocationID.NotFound")
cm.exception.status.should.equal(400)
@ -374,7 +373,7 @@ def test_eip_disassociate_arg_error():
"""Invalid arguments disassociate address"""
conn = boto.connect_ec2("the_key", "the_secret")
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.disassociate_address()
cm.exception.code.should.equal("MissingParameter")
cm.exception.status.should.equal(400)
@ -386,7 +385,7 @@ def test_eip_release_arg_error():
"""Invalid arguments release address"""
conn = boto.connect_ec2("the_key", "the_secret")
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.release_address()
cm.exception.code.should.equal("MissingParameter")
cm.exception.status.should.equal(400)
@ -438,7 +437,7 @@ def test_eip_describe_none():
"""Error when search for bogus IP"""
conn = boto.connect_ec2("the_key", "the_secret")
with assert_raises(EC2ResponseError) as cm:
with pytest.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)

15
tests/test_ec2/test_elastic_network_interfaces.py
View file

@ -1,8 +1,7 @@
from __future__ import unicode_literals
# Ensure 'assert_raises' context manager support for Python 2.6
import tests.backport_assert_raises
from nose.tools import assert_raises
import pytest
import boto3
from botocore.exceptions import ClientError
@ -21,7 +20,7 @@ def test_elastic_network_interfaces():
vpc = conn.create_vpc("10.0.0.0/16")
subnet = conn.create_subnet(vpc.id, "10.0.0.0/18")
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
eni = conn.create_network_interface(subnet.id, dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -38,7 +37,7 @@ def test_elastic_network_interfaces():
eni.private_ip_addresses.should.have.length_of(1)
eni.private_ip_addresses[0].private_ip_address.startswith("10.").should.be.true
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
conn.delete_network_interface(eni.id, dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -51,7 +50,7 @@ def test_elastic_network_interfaces():
all_enis = conn.get_all_network_interfaces()
all_enis.should.have.length_of(0)
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.delete_network_interface(eni.id)
cm.exception.error_code.should.equal("InvalidNetworkInterfaceID.NotFound")
cm.exception.status.should.equal(400)
@ -62,7 +61,7 @@ def test_elastic_network_interfaces():
def test_elastic_network_interfaces_subnet_validation():
conn = boto.connect_vpc("the_key", "the_secret")
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.create_network_interface("subnet-abcd1234")
cm.exception.error_code.should.equal("InvalidSubnetID.NotFound")
cm.exception.status.should.equal(400)
@ -133,7 +132,7 @@ def test_elastic_network_interfaces_modify_attribute():
eni.groups.should.have.length_of(1)
eni.groups[0].id.should.equal(security_group1.id)
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
conn.modify_network_interface_attribute(
eni.id, "groupset", [security_group2.id], dry_run=True
)
@ -228,7 +227,7 @@ def test_elastic_network_interfaces_get_by_tag_name():
SubnetId=subnet.id, PrivateIpAddress="10.0.10.5"
)
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
eni1.create_tags(Tags=[{"Key": "Name", "Value": "eni1"}], DryRun=True)
ex.exception.response["Error"]["Code"].should.equal("DryRunOperation")
ex.exception.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)

23
tests/test_ec2/test_flow_logs.py
View file

@ -1,7 +1,6 @@
from __future__ import unicode_literals
import tests.backport_assert_raises # noqa
from nose.tools import assert_raises
import pytest
import boto3
@ -36,7 +35,7 @@ def test_create_flow_logs_s3():
CreateBucketConfiguration={"LocationConstraint": "us-west-1"},
)
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
client.create_flow_logs(
ResourceType="VPC",
ResourceIds=[vpc["VpcId"]],
@ -87,7 +86,7 @@ def test_create_flow_logs_cloud_watch():
vpc = client.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]
logs_client.create_log_group(logGroupName="test-group")
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
client.create_flow_logs(
ResourceType="VPC",
ResourceIds=[vpc["VpcId"]],
@ -243,7 +242,7 @@ def test_delete_flow_logs_delete_many():
def test_delete_flow_logs_non_existing():
client = boto3.client("ec2", region_name="us-west-1")
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
client.delete_flow_logs(FlowLogIds=["fl-1a2b3c4d"])
ex.exception.response["Error"]["Code"].should.equal("InvalidFlowLogId.NotFound")
ex.exception.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
@ -251,7 +250,7 @@ def test_delete_flow_logs_non_existing():
"These flow log ids in the input list are not found: [TotalCount: 1] fl-1a2b3c4d"
)
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
client.delete_flow_logs(FlowLogIds=["fl-1a2b3c4d", "fl-2b3c4d5e"])
ex.exception.response["Error"]["Code"].should.equal("InvalidFlowLogId.NotFound")
ex.exception.response["ResponseMetadata"]["HTTPStatusCode"].should.equal(400)
@ -304,7 +303,7 @@ def test_create_flow_logs_invalid_parameters():
CreateBucketConfiguration={"LocationConstraint": "us-west-1"},
)
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
client.create_flow_logs(
ResourceType="VPC",
ResourceIds=[vpc["VpcId"]],
@ -319,7 +318,7 @@ def test_create_flow_logs_invalid_parameters():
"Invalid Flow Log Max Aggregation Interval"
)
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
client.create_flow_logs(
ResourceType="VPC",
ResourceIds=[vpc["VpcId"]],
@ -332,7 +331,7 @@ def test_create_flow_logs_invalid_parameters():
"LogDestination can't be empty if LogGroupName is not provided."
)
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
client.create_flow_logs(
ResourceType="VPC",
ResourceIds=[vpc["VpcId"]],
@ -346,7 +345,7 @@ def test_create_flow_logs_invalid_parameters():
"LogDestination type must be cloud-watch-logs if LogGroupName is provided."
)
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
client.create_flow_logs(
ResourceType="VPC",
ResourceIds=[vpc["VpcId"]],
@ -368,7 +367,7 @@ def test_create_flow_logs_invalid_parameters():
)["FlowLogIds"]
response.should.have.length_of(1)
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
client.create_flow_logs(
ResourceType="VPC",
ResourceIds=[vpc["VpcId"]],
@ -391,7 +390,7 @@ def test_create_flow_logs_invalid_parameters():
)["FlowLogIds"]
response.should.have.length_of(1)
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
client.create_flow_logs(
ResourceType="VPC",
ResourceIds=[vpc["VpcId"]],

5
tests/test_ec2/test_general.py
View file

@ -1,8 +1,7 @@
from __future__ import unicode_literals
# Ensure 'assert_raises' context manager support for Python 2.6
import tests.backport_assert_raises
from nose.tools import assert_raises
import pytest
import boto
import boto3
@ -25,7 +24,7 @@ def test_console_output():
def test_console_output_without_instance():
conn = boto.connect_ec2("the_key", "the_secret")
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.get_console_output("i-1234abcd")
cm.exception.code.should.equal("InvalidInstanceID.NotFound")
cm.exception.status.should.equal(400)

37
tests/test_ec2/test_instances.py
View file

@ -3,8 +3,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
import pytest
import base64
import ipaddress
@ -52,7 +51,7 @@ def test_add_servers():
def test_instance_launch_and_terminate():
conn = boto.ec2.connect_to_region("us-east-1")
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
reservation = conn.run_instances("ami-1234abcd", dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -87,7 +86,7 @@ def test_instance_launch_and_terminate():
volume.attach_data.instance_id.should.equal(instance.id)
volume.status.should.equal("in-use")
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
conn.terminate_instances([instance.id], dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -290,7 +289,7 @@ 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
with assert_raises(EC2ResponseError) as cm:
with pytest.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)
@ -743,7 +742,7 @@ def test_instance_start_and_stop():
instance_ids = [instance.id for instance in instances]
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
stopped_instances = conn.stop_instances(instance_ids, dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -756,7 +755,7 @@ def test_instance_start_and_stop():
for instance in stopped_instances:
instance.state.should.equal("stopping")
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
started_instances = conn.start_instances([instances[0].id], dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -774,7 +773,7 @@ def test_instance_reboot():
reservation = conn.run_instances("ami-1234abcd")
instance = reservation.instances[0]
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
instance.reboot(dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -792,7 +791,7 @@ def test_instance_attribute_instance_type():
reservation = conn.run_instances("ami-1234abcd")
instance = reservation.instances[0]
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
instance.modify_attribute("instanceType", "m1.small", dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -820,7 +819,7 @@ def test_modify_instance_attribute_security_groups():
"test security group 2", "this is a test security group 2"
).id
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
instance.modify_attribute("groupSet", [sg_id, sg_id2], dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -843,7 +842,7 @@ def test_instance_attribute_user_data():
reservation = conn.run_instances("ami-1234abcd")
instance = reservation.instances[0]
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
instance.modify_attribute("userData", "this is my user data", dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -873,7 +872,7 @@ def test_instance_attribute_source_dest_check():
# Set to false (note: Boto converts bool to string, eg 'false')
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
instance.modify_attribute("sourceDestCheck", False, dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -919,7 +918,7 @@ def test_user_data_with_run_instance():
def test_run_instance_with_security_group_name():
conn = boto.connect_ec2("the_key", "the_secret")
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
group = conn.create_security_group("group1", "some description", dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -1196,7 +1195,7 @@ def test_instance_with_nic_attach_detach():
set([group.id for group in eni.groups]).should.equal(set([security_group2.id]))
# Attach
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
conn.attach_network_interface(eni.id, instance.id, device_index=1, dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -1223,7 +1222,7 @@ def test_instance_with_nic_attach_detach():
)
# Detach
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
conn.detach_network_interface(instance_eni.attachment.id, dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -1242,7 +1241,7 @@ def test_instance_with_nic_attach_detach():
set([group.id for group in eni.groups]).should.equal(set([security_group2.id]))
# Detach with invalid attachment ID
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.detach_network_interface("eni-attach-1234abcd")
cm.exception.code.should.equal("InvalidAttachmentID.NotFound")
cm.exception.status.should.equal(400)
@ -1410,7 +1409,7 @@ def test_describe_instance_status_with_instance_filter_deprecated():
all_status[0].id.should.equal(instance.id)
# Call get_all_instance_status with a bad id should raise an error
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.get_all_instance_status(instance_ids=[instance.id, "i-1234abcd"])
cm.exception.code.should.equal("InvalidInstanceID.NotFound")
cm.exception.status.should.equal(400)
@ -1537,7 +1536,7 @@ def test_get_instance_by_security_group():
security_group = conn.create_security_group("test", "test")
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
conn.modify_instance_attribute(
instance.id, "groupSet", [security_group.id], dry_run=True
)
@ -1661,7 +1660,7 @@ def test_describe_instance_attribute():
]
for invalid_instance_attribute in invalid_instance_attributes:
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
client.describe_instance_attribute(
InstanceId=instance_id, Attribute=invalid_instance_attribute
)

25
tests/test_ec2/test_internet_gateways.py
View file

@ -1,8 +1,7 @@
from __future__ import unicode_literals
# Ensure 'assert_raises' context manager support for Python 2.6
import tests.backport_assert_raises
from nose.tools import assert_raises
import pytest
import re
@ -28,7 +27,7 @@ def test_igw_create():
conn.get_all_internet_gateways().should.have.length_of(0)
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
igw = conn.create_internet_gateway(dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -51,7 +50,7 @@ def test_igw_attach():
igw = conn.create_internet_gateway()
vpc = conn.create_vpc(VPC_CIDR)
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
conn.attach_internet_gateway(igw.id, vpc.id, dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -71,7 +70,7 @@ def test_igw_attach_bad_vpc():
conn = boto.connect_vpc("the_key", "the_secret")
igw = conn.create_internet_gateway()
with assert_raises(EC2ResponseError) as cm:
with pytest.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)
@ -87,7 +86,7 @@ def test_igw_attach_twice():
vpc2 = conn.create_vpc(VPC_CIDR)
conn.attach_internet_gateway(igw.id, vpc1.id)
with assert_raises(EC2ResponseError) as cm:
with pytest.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)
@ -102,7 +101,7 @@ def test_igw_detach():
vpc = conn.create_vpc(VPC_CIDR)
conn.attach_internet_gateway(igw.id, vpc.id)
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
conn.detach_internet_gateway(igw.id, vpc.id, dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -124,7 +123,7 @@ def test_igw_detach_wrong_vpc():
vpc2 = conn.create_vpc(VPC_CIDR)
conn.attach_internet_gateway(igw.id, vpc1.id)
with assert_raises(EC2ResponseError) as cm:
with pytest.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)
@ -139,7 +138,7 @@ def test_igw_detach_invalid_vpc():
vpc = conn.create_vpc(VPC_CIDR)
conn.attach_internet_gateway(igw.id, vpc.id)
with assert_raises(EC2ResponseError) as cm:
with pytest.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)
@ -153,7 +152,7 @@ def test_igw_detach_unattached():
igw = conn.create_internet_gateway()
vpc = conn.create_vpc(VPC_CIDR)
with assert_raises(EC2ResponseError) as cm:
with pytest.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)
@ -169,7 +168,7 @@ def test_igw_delete():
igw = conn.create_internet_gateway()
conn.get_all_internet_gateways().should.have.length_of(1)
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
conn.delete_internet_gateway(igw.id, dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -189,7 +188,7 @@ def test_igw_delete_attached():
vpc = conn.create_vpc(VPC_CIDR)
conn.attach_internet_gateway(igw.id, vpc.id)
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.delete_internet_gateway(igw.id)
cm.exception.code.should.equal("DependencyViolation")
cm.exception.status.should.equal(400)
@ -209,7 +208,7 @@ def test_igw_desribe():
def test_igw_describe_bad_id():
""" internet gateway fail to fetch by bad id """
conn = boto.connect_vpc("the_key", "the_secret")
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.get_all_internet_gateways([BAD_IGW])
cm.exception.code.should.equal("InvalidInternetGatewayID.NotFound")
cm.exception.status.should.equal(400)

21
tests/test_ec2/test_key_pairs.py
View file

@ -1,8 +1,7 @@
from __future__ import unicode_literals
# Ensure 'assert_raises' context manager support for Python 2.6
import tests.backport_assert_raises
from nose.tools import assert_raises
import pytest
import boto
import sure # noqa
@ -56,7 +55,7 @@ def test_key_pairs_empty():
def test_key_pairs_invalid_id():
conn = boto.connect_ec2("the_key", "the_secret")
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.get_all_key_pairs("foo")
cm.exception.code.should.equal("InvalidKeyPair.NotFound")
cm.exception.status.should.equal(400)
@ -67,7 +66,7 @@ def test_key_pairs_invalid_id():
def test_key_pairs_create():
conn = boto.connect_ec2("the_key", "the_secret")
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
conn.create_key_pair("foo", dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -110,7 +109,7 @@ def test_key_pairs_create_exist():
conn.create_key_pair("foo")
assert len(conn.get_all_key_pairs()) == 1
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.create_key_pair("foo")
cm.exception.code.should.equal("InvalidKeyPair.Duplicate")
cm.exception.status.should.equal(400)
@ -130,7 +129,7 @@ def test_key_pairs_delete_exist():
conn = boto.connect_ec2("the_key", "the_secret")
conn.create_key_pair("foo")
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
r = conn.delete_key_pair("foo", dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -147,7 +146,7 @@ def test_key_pairs_delete_exist():
def test_key_pairs_import():
conn = boto.connect_ec2("the_key", "the_secret")
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
conn.import_key_pair("foo", RSA_PUBLIC_KEY_OPENSSH, dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -176,7 +175,7 @@ def test_key_pairs_import_exist():
assert kp.name == "foo"
assert len(conn.get_all_key_pairs()) == 1
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.create_key_pair("foo")
cm.exception.code.should.equal("InvalidKeyPair.Duplicate")
cm.exception.status.should.equal(400)
@ -187,19 +186,19 @@ def test_key_pairs_import_exist():
def test_key_pairs_invalid():
conn = boto.connect_ec2("the_key", "the_secret")
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
conn.import_key_pair("foo", b"")
ex.exception.error_code.should.equal("InvalidKeyPair.Format")
ex.exception.status.should.equal(400)
ex.exception.message.should.equal("Key is not in valid OpenSSH public key format")
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
conn.import_key_pair("foo", b"garbage")
ex.exception.error_code.should.equal("InvalidKeyPair.Format")
ex.exception.status.should.equal(400)
ex.exception.message.should.equal("Key is not in valid OpenSSH public key format")
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
conn.import_key_pair("foo", DSA_PUBLIC_KEY_OPENSSH)
ex.exception.error_code.should.equal("InvalidKeyPair.Format")
ex.exception.status.should.equal(400)

4
tests/test_ec2/test_launch_templates.py
View file

@ -1,7 +1,7 @@
import boto3
import sure # noqa
from nose.tools import assert_raises
import pytest
from botocore.client import ClientError
from moto import mock_ec2
@ -30,7 +30,7 @@ def test_launch_template_create():
lt["DefaultVersionNumber"].should.equal(1)
lt["LatestVersionNumber"].should.equal(1)
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
cli.create_launch_template(
LaunchTemplateName="test-template",
LaunchTemplateData={

4
tests/test_ec2/test_network_acls.py
View file

@ -2,7 +2,7 @@ from __future__ import unicode_literals
import boto
import boto3
import sure # noqa
from nose.tools import assert_raises
import pytest
from botocore.exceptions import ClientError
from moto import mock_ec2_deprecated, mock_ec2
@ -261,7 +261,7 @@ def test_duplicate_network_acl_entry():
RuleNumber=rule_number,
)
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
default_network_acl.create_entry(
CidrBlock="10.0.0.0/0",
Egress=egress,

25
tests/test_ec2/test_route_tables.py
View file

@ -1,8 +1,7 @@
from __future__ import unicode_literals
# Ensure 'assert_raises' context manager support for Python 2.6
import tests.backport_assert_raises
from nose.tools import assert_raises
import pytest
import boto
import boto3
@ -61,7 +60,7 @@ def test_route_tables_additional():
local_route.state.should.equal("active")
local_route.destination_cidr_block.should.equal(vpc.cidr_block)
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.delete_vpc(vpc.id)
cm.exception.code.should.equal("DependencyViolation")
cm.exception.status.should.equal(400)
@ -72,7 +71,7 @@ def test_route_tables_additional():
all_route_tables = conn.get_all_route_tables(filters={"vpc-id": vpc.id})
all_route_tables.should.have.length_of(1)
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.delete_route_table("rtb-1234abcd")
cm.exception.code.should.equal("InvalidRouteTableID.NotFound")
cm.exception.status.should.equal(400)
@ -197,7 +196,7 @@ def test_route_table_associations():
association_id_idempotent.should.equal(association_id)
# Error: Attempt delete associated route table.
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.delete_route_table(route_table.id)
cm.exception.code.should.equal("DependencyViolation")
cm.exception.status.should.equal(400)
@ -211,21 +210,21 @@ def test_route_table_associations():
route_table.associations.should.have.length_of(0)
# Error: Disassociate with invalid association ID
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.disassociate_route_table(association_id)
cm.exception.code.should.equal("InvalidAssociationID.NotFound")
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
# Error: Associate with invalid subnet ID
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.associate_route_table(route_table.id, "subnet-1234abcd")
cm.exception.code.should.equal("InvalidSubnetID.NotFound")
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
# Error: Associate with invalid route table ID
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.associate_route_table("rtb-1234abcd", subnet.id)
cm.exception.code.should.equal("InvalidRouteTableID.NotFound")
cm.exception.status.should.equal(400)
@ -293,7 +292,7 @@ def test_route_table_replace_route_table_association():
association_id_idempotent.should.equal(association_id2)
# Error: Replace association with invalid association ID
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.replace_route_table_association_with_assoc(
"rtbassoc-1234abcd", route_table1.id
)
@ -302,7 +301,7 @@ def test_route_table_replace_route_table_association():
cm.exception.request_id.should_not.be.none
# Error: Replace association with invalid route table ID
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.replace_route_table_association_with_assoc(association_id2, "rtb-1234abcd")
cm.exception.code.should.equal("InvalidRouteTableID.NotFound")
cm.exception.status.should.equal(400)
@ -389,7 +388,7 @@ def test_routes_additional():
]
new_routes.should.have.length_of(0)
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.delete_route(main_route_table.id, ROUTE_CIDR)
cm.exception.code.should.equal("InvalidRoute.NotFound")
cm.exception.status.should.equal(400)
@ -442,7 +441,7 @@ def test_routes_replace():
target_route.state.should.equal("active")
target_route.destination_cidr_block.should.equal(ROUTE_CIDR)
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.replace_route("rtb-1234abcd", ROUTE_CIDR, gateway_id=igw.id)
cm.exception.code.should.equal("InvalidRouteTableID.NotFound")
cm.exception.status.should.equal(400)
@ -571,7 +570,7 @@ def test_create_route_with_invalid_destination_cidr_block_parameter():
internet_gateway.reload()
destination_cidr_block = "1000.1.0.0/20"
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
route = route_table.create_route(
DestinationCidrBlock=destination_cidr_block, GatewayId=internet_gateway.id
)

59
tests/test_ec2/test_security_groups.py
View file

@ -4,8 +4,7 @@ import copy
import json
# Ensure 'assert_raises' context manager support for Python 2.6
import tests.backport_assert_raises # noqa
from nose.tools import assert_raises
import pytest
import boto3
import boto
@ -20,7 +19,7 @@ from moto import mock_ec2, mock_ec2_deprecated
def test_create_and_describe_security_group():
conn = boto.connect_ec2("the_key", "the_secret")
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
security_group = conn.create_security_group(
"test security group", "this is a test security group", dry_run=True
)
@ -38,7 +37,7 @@ 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
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.create_security_group(
"test security group", "this is a test security group"
)
@ -57,7 +56,7 @@ def test_create_and_describe_security_group():
def test_create_security_group_without_description_raises_error():
conn = boto.connect_ec2("the_key", "the_secret")
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.create_security_group("test security group", "")
cm.exception.code.should.equal("MissingParameter")
cm.exception.status.should.equal(400)
@ -87,7 +86,7 @@ def test_create_and_describe_vpc_security_group():
# Trying to create another group with the same name in the same VPC should
# throw an error
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.create_security_group(
"test security group", "this is a test security group", vpc_id
)
@ -146,14 +145,14 @@ def test_deleting_security_groups():
conn.get_all_security_groups().should.have.length_of(4)
# Deleting a group that doesn't exist should throw an error
with assert_raises(EC2ResponseError) as cm:
with pytest.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
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
conn.delete_security_group("test2", dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -184,7 +183,7 @@ def test_authorize_ip_range_and_revoke():
conn = boto.connect_ec2("the_key", "the_secret")
security_group = conn.create_security_group("test", "test")
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
success = security_group.authorize(
ip_protocol="tcp",
from_port="22",
@ -208,7 +207,7 @@ 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
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
security_group.revoke(
ip_protocol="tcp",
from_port="22",
@ -220,7 +219,7 @@ def test_authorize_ip_range_and_revoke():
cm.exception.request_id.should_not.be.none
# Actually revoke
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
security_group.revoke(
ip_protocol="tcp",
from_port="22",
@ -246,7 +245,7 @@ def test_authorize_ip_range_and_revoke():
"testegress", "testegress", vpc_id="vpc-3432589"
)
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
success = conn.authorize_security_group_egress(
egress_security_group.id,
"tcp",
@ -285,7 +284,7 @@ def test_authorize_ip_range_and_revoke():
).should.throw(EC2ResponseError)
# Actually revoke
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
conn.revoke_security_group_egress(
egress_security_group.id,
"tcp",
@ -335,7 +334,7 @@ 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
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
security_group.revoke(
ip_protocol="tcp", from_port="22", to_port="2222", src_group=wrong_group
)
@ -440,7 +439,7 @@ def test_get_all_security_groups():
resp.should.have.length_of(1)
resp[0].id.should.equal(sg1.id)
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.get_all_security_groups(groupnames=["does_not_exist"])
cm.exception.code.should.equal("InvalidGroup.NotFound")
cm.exception.status.should.equal(400)
@ -469,7 +468,7 @@ def test_get_all_security_groups():
def test_authorize_bad_cidr_throws_invalid_parameter_value():
conn = boto.connect_ec2("the_key", "the_secret")
security_group = conn.create_security_group("test", "test")
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
security_group.authorize(
ip_protocol="tcp", from_port="22", to_port="2222", cidr_ip="123.123.123.123"
)
@ -485,7 +484,7 @@ def test_security_group_tagging():
sg = conn.create_security_group("test-sg", "Test SG", vpc.id)
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
sg.add_tag("Test", "Tag", dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -534,7 +533,7 @@ def test_sec_group_rule_limit():
other_sg = ec2_conn.create_security_group("test_2", "test_other")
# INGRESS
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
ec2_conn.authorize_security_group(
group_id=sg.id,
ip_protocol="-1",
@ -556,13 +555,13 @@ def test_sec_group_rule_limit():
)
success.should.be.true
# verify that we cannot authorize past the limit for a CIDR IP
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
ec2_conn.authorize_security_group(
group_id=sg.id, ip_protocol="-1", cidr_ip=["100.0.0.0/0"]
)
cm.exception.error_code.should.equal("RulesPerSecurityGroupLimitExceeded")
# verify that we cannot authorize past the limit for a different sec group
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
ec2_conn.authorize_security_group(
group_id=sg.id, ip_protocol="-1", src_security_group_group_id=other_sg.id
)
@ -581,13 +580,13 @@ def test_sec_group_rule_limit():
group_id=sg.id, ip_protocol="-1", cidr_ip="{0}.0.0.0/0".format(i)
)
# verify that we cannot authorize past the limit for a CIDR IP
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
ec2_conn.authorize_security_group_egress(
group_id=sg.id, ip_protocol="-1", cidr_ip="101.0.0.0/0"
)
cm.exception.error_code.should.equal("RulesPerSecurityGroupLimitExceeded")
# verify that we cannot authorize past the limit for a different sec group
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
ec2_conn.authorize_security_group_egress(
group_id=sg.id, ip_protocol="-1", src_group_id=other_sg.id
)
@ -605,7 +604,7 @@ def test_sec_group_rule_limit_vpc():
other_sg = ec2_conn.create_security_group("test_2", "test", vpc_id=vpc.id)
# INGRESS
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
ec2_conn.authorize_security_group(
group_id=sg.id,
ip_protocol="-1",
@ -627,13 +626,13 @@ def test_sec_group_rule_limit_vpc():
)
# verify that we cannot authorize past the limit for a CIDR IP
success.should.be.true
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
ec2_conn.authorize_security_group(
group_id=sg.id, ip_protocol="-1", cidr_ip=["100.0.0.0/0"]
)
cm.exception.error_code.should.equal("RulesPerSecurityGroupLimitExceeded")
# verify that we cannot authorize past the limit for a different sec group
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
ec2_conn.authorize_security_group(
group_id=sg.id, ip_protocol="-1", src_security_group_group_id=other_sg.id
)
@ -652,13 +651,13 @@ def test_sec_group_rule_limit_vpc():
group_id=sg.id, ip_protocol="-1", cidr_ip="{0}.0.0.0/0".format(i)
)
# verify that we cannot authorize past the limit for a CIDR IP
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
ec2_conn.authorize_security_group_egress(
group_id=sg.id, ip_protocol="-1", cidr_ip="50.0.0.0/0"
)
cm.exception.error_code.should.equal("RulesPerSecurityGroupLimitExceeded")
# verify that we cannot authorize past the limit for a different sec group
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
ec2_conn.authorize_security_group_egress(
group_id=sg.id, ip_protocol="-1", src_group_id=other_sg.id
)
@ -689,7 +688,7 @@ def test_add_same_rule_twice_throws_error():
]
sg.authorize_ingress(IpPermissions=ip_permissions)
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
sg.authorize_ingress(IpPermissions=ip_permissions)
@ -761,7 +760,7 @@ def test_security_group_tagging_boto3():
sg = conn.create_security_group(GroupName="test-sg", Description="Test SG")
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
conn.create_tags(
Resources=[sg["GroupId"]],
Tags=[{"Key": "Test", "Value": "Tag"}],
@ -926,7 +925,7 @@ def test_get_all_security_groups_filter_with_same_vpc_id():
)
security_groups.should.have.length_of(1)
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.get_all_security_groups(group_ids=["does_not_exist"])
cm.exception.code.should.equal("InvalidGroup.NotFound")
cm.exception.status.should.equal(400)

6
tests/test_ec2/test_spot_instances.py
View file

@ -1,5 +1,5 @@
from __future__ import unicode_literals
from nose.tools import assert_raises
import pytest
import datetime
import boto
@ -31,7 +31,7 @@ def test_request_spot_instances():
start = iso_8601_datetime_with_milliseconds(start_dt)
end = iso_8601_datetime_with_milliseconds(end_dt)
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
request = conn.request_spot_instances(
SpotPrice="0.5",
InstanceCount=1,
@ -155,7 +155,7 @@ def test_cancel_spot_instance_request():
requests = conn.get_all_spot_instance_requests()
requests.should.have.length_of(1)
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
conn.cancel_spot_instance_requests([requests[0].id], dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)

19
tests/test_ec2/test_subnets.py
View file

@ -1,8 +1,7 @@
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 pytest
import boto3
import boto
@ -30,7 +29,7 @@ def test_subnets():
all_subnets = conn.get_all_subnets()
all_subnets.should.have.length_of(0 + len(ec2.get_all_zones()))
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.delete_subnet(subnet.id)
cm.exception.code.should.equal("InvalidSubnetID.NotFound")
cm.exception.status.should.equal(400)
@ -41,7 +40,7 @@ def test_subnets():
def test_subnet_create_vpc_validation():
conn = boto.connect_vpc("the_key", "the_secret")
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.create_subnet("vpc-abcd1234", "10.0.0.0/18")
cm.exception.code.should.equal("InvalidVpcID.NotFound")
cm.exception.status.should.equal(400)
@ -202,7 +201,7 @@ def test_modify_subnet_attribute_validation():
VpcId=vpc.id, CidrBlock="10.0.0.0/24", AvailabilityZone="us-west-1a"
)
with assert_raises(ParamValidationError):
with pytest.raises(ParamValidationError):
client.modify_subnet_attribute(
SubnetId=subnet.id, MapPublicIpOnLaunch={"Value": "invalid"}
)
@ -228,7 +227,7 @@ def test_subnet_get_by_id():
subnetA.id.should.be.within(subnets_by_id)
subnetB1.id.should.be.within(subnets_by_id)
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.get_all_subnets(subnet_ids=["subnet-does_not_exist"])
cm.exception.code.should.equal("InvalidSubnetID.NotFound")
cm.exception.status.should.equal(400)
@ -386,7 +385,7 @@ def test_create_subnet_with_invalid_availability_zone():
vpc = ec2.create_vpc(CidrBlock="10.0.0.0/16")
subnet_availability_zone = "asfasfas"
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
subnet = client.create_subnet(
VpcId=vpc.id,
CidrBlock="10.0.0.0/24",
@ -409,7 +408,7 @@ def test_create_subnet_with_invalid_cidr_range():
vpc.is_default.shouldnt.be.ok
subnet_cidr_block = "10.1.0.0/20"
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
subnet = ec2.create_subnet(VpcId=vpc.id, CidrBlock=subnet_cidr_block)
str(ex.exception).should.equal(
"An error occurred (InvalidSubnet.Range) when calling the CreateSubnet "
@ -444,7 +443,7 @@ def test_create_subnet_with_invalid_cidr_block_parameter():
vpc.is_default.shouldnt.be.ok
subnet_cidr_block = "1000.1.0.0/20"
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
subnet = ec2.create_subnet(VpcId=vpc.id, CidrBlock=subnet_cidr_block)
str(ex.exception).should.equal(
"An error occurred (InvalidParameterValue) when calling the CreateSubnet "
@ -503,7 +502,7 @@ def test_create_subnets_with_overlapping_cidr_blocks():
vpc.is_default.shouldnt.be.ok
subnet_cidr_block = "10.0.0.0/24"
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
subnet1 = ec2.create_subnet(VpcId=vpc.id, CidrBlock=subnet_cidr_block)
subnet2 = ec2.create_subnet(VpcId=vpc.id, CidrBlock=subnet_cidr_block)
str(ex.exception).should.equal(

24
tests/test_ec2/test_tags.py
View file

@ -1,5 +1,5 @@
from __future__ import unicode_literals
from nose.tools import assert_raises
import pytest
import itertools
@ -11,7 +11,7 @@ from boto.ec2.instance import Reservation
import sure # noqa
from moto import mock_ec2_deprecated, mock_ec2
from nose.tools import assert_raises
import pytest
@mock_ec2_deprecated
@ -20,7 +20,7 @@ def test_add_tag():
reservation = conn.run_instances("ami-1234abcd")
instance = reservation.instances[0]
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
instance.add_tag("a key", "some value", dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -51,7 +51,7 @@ def test_remove_tag():
tag.name.should.equal("a key")
tag.value.should.equal("some value")
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
instance.remove_tag("a key", dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -106,7 +106,7 @@ def test_create_tags():
"blank key": "",
}
with assert_raises(EC2ResponseError) as ex:
with pytest.raises(EC2ResponseError) as ex:
conn.create_tags(instance.id, tag_dict, dry_run=True)
ex.exception.error_code.should.equal("DryRunOperation")
ex.exception.status.should.equal(400)
@ -131,14 +131,14 @@ def test_tag_limit_exceeded():
for i in range(51):
tag_dict["{0:02d}".format(i + 1)] = ""
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.create_tags(instance.id, tag_dict)
cm.exception.code.should.equal("TagLimitExceeded")
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
instance.add_tag("a key", "a value")
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.create_tags(instance.id, tag_dict)
cm.exception.code.should.equal("TagLimitExceeded")
cm.exception.status.should.equal(400)
@ -157,7 +157,7 @@ def test_invalid_parameter_tag_null():
reservation = conn.run_instances("ami-1234abcd")
instance = reservation.instances[0]
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
instance.add_tag("a key", None)
cm.exception.code.should.equal("InvalidParameterValue")
cm.exception.status.should.equal(400)
@ -167,13 +167,13 @@ def test_invalid_parameter_tag_null():
@mock_ec2_deprecated
def test_invalid_id():
conn = boto.connect_ec2("the_key", "the_secret")
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.create_tags("ami-blah", {"key": "tag"})
cm.exception.code.should.equal("InvalidID")
cm.exception.status.should.equal(400)
cm.exception.request_id.should_not.be.none
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.create_tags("blah-blah", {"key": "tag"})
cm.exception.code.should.equal("InvalidID")
cm.exception.status.should.equal(400)
@ -449,7 +449,7 @@ def test_create_tag_empty_resource():
# create ec2 client in us-west-1
client = boto3.client("ec2", region_name="us-west-1")
# create tag with empty resource
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
client.create_tags(Resources=[], Tags=[{"Key": "Value"}])
ex.exception.response["Error"]["Code"].should.equal("MissingParameter")
ex.exception.response["Error"]["Message"].should.equal(
@ -462,7 +462,7 @@ def test_delete_tag_empty_resource():
# create ec2 client in us-west-1
client = boto3.client("ec2", region_name="us-west-1")
# delete tag with empty resource
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
client.delete_tags(Resources=[], Tags=[{"Key": "Value"}])
ex.exception.response["Error"]["Code"].should.equal("MissingParameter")
ex.exception.response["Error"]["Message"].should.equal(

15
tests/test_ec2/test_vpc_peering.py
View file

@ -1,8 +1,7 @@
from __future__ import unicode_literals
# Ensure 'assert_raises' context manager support for Python 2.6
import tests.backport_assert_raises
from nose.tools import assert_raises
import pytest
from moto.ec2.exceptions import EC2ClientError
from botocore.exceptions import ClientError
@ -49,7 +48,7 @@ def test_vpc_peering_connections_accept():
vpc_pcx = conn.accept_vpc_peering_connection(vpc_pcx.id)
vpc_pcx._status.code.should.equal("active")
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.reject_vpc_peering_connection(vpc_pcx.id)
cm.exception.code.should.equal("InvalidStateTransition")
cm.exception.status.should.equal(400)
@ -69,7 +68,7 @@ def test_vpc_peering_connections_reject():
verdict = conn.reject_vpc_peering_connection(vpc_pcx.id)
verdict.should.equal(True)
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.accept_vpc_peering_connection(vpc_pcx.id)
cm.exception.code.should.equal("InvalidStateTransition")
cm.exception.status.should.equal(400)
@ -93,7 +92,7 @@ def test_vpc_peering_connections_delete():
all_vpc_pcxs.should.have.length_of(1)
all_vpc_pcxs[0]._status.code.should.equal("deleted")
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.delete_vpc_peering_connection("pcx-1234abcd")
cm.exception.code.should.equal("InvalidVpcPeeringConnectionId.NotFound")
cm.exception.status.should.equal(400)
@ -129,7 +128,7 @@ def test_vpc_peering_connections_cross_region_fail():
ec2_apn1 = boto3.resource("ec2", region_name="ap-northeast-1")
vpc_apn1 = ec2_apn1.create_vpc(CidrBlock="10.20.0.0/16")
# create peering wrong region with no vpc
with assert_raises(ClientError) as cm:
with pytest.raises(ClientError) as cm:
ec2_usw1.create_vpc_peering_connection(
VpcId=vpc_usw1.id, PeerVpcId=vpc_apn1.id, PeerRegion="ap-northeast-2"
)
@ -253,7 +252,7 @@ def test_vpc_peering_connections_cross_region_accept_wrong_region():
# accept wrong peering from us-west-1 which will raise error
ec2_apn1 = boto3.client("ec2", region_name="ap-northeast-1")
ec2_usw1 = boto3.client("ec2", region_name="us-west-1")
with assert_raises(ClientError) as cm:
with pytest.raises(ClientError) as cm:
ec2_usw1.accept_vpc_peering_connection(VpcPeeringConnectionId=vpc_pcx_usw1.id)
cm.exception.response["Error"]["Code"].should.equal("OperationNotPermitted")
exp_msg = (
@ -278,7 +277,7 @@ def test_vpc_peering_connections_cross_region_reject_wrong_region():
# reject wrong peering from us-west-1 which will raise error
ec2_apn1 = boto3.client("ec2", region_name="ap-northeast-1")
ec2_usw1 = boto3.client("ec2", region_name="us-west-1")
with assert_raises(ClientError) as cm:
with pytest.raises(ClientError) as cm:
ec2_usw1.reject_vpc_peering_connection(VpcPeeringConnectionId=vpc_pcx_usw1.id)
cm.exception.response["Error"]["Code"].should.equal("OperationNotPermitted")
exp_msg = (

19
tests/test_ec2/test_vpcs.py
View file

@ -1,8 +1,7 @@
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 pytest
from moto.ec2.exceptions import EC2ClientError
from botocore.exceptions import ClientError
@ -31,7 +30,7 @@ def test_vpcs():
all_vpcs = conn.get_all_vpcs()
all_vpcs.should.have.length_of(1)
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.delete_vpc("vpc-1234abcd")
cm.exception.code.should.equal("InvalidVpcID.NotFound")
cm.exception.status.should.equal(400)
@ -114,7 +113,7 @@ def test_vpc_get_by_id():
vpc1.id.should.be.within(vpc_ids)
vpc2.id.should.be.within(vpc_ids)
with assert_raises(EC2ResponseError) as cm:
with pytest.raises(EC2ResponseError) as cm:
conn.get_all_vpcs(vpc_ids=["vpc-does_not_exist"])
cm.exception.code.should.equal("InvalidVpcID.NotFound")
cm.exception.status.should.equal(400)
@ -402,7 +401,7 @@ def test_associate_vpc_ipv4_cidr_block():
)
# Check error on adding 6th association.
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
response = ec2.meta.client.associate_vpc_cidr_block(
VpcId=vpc.id, CidrBlock="10.10.50.0/22"
)
@ -447,7 +446,7 @@ def test_disassociate_vpc_ipv4_cidr_block():
)
# Error attempting to delete a non-existent CIDR_BLOCK association
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
response = ec2.meta.client.disassociate_vpc_cidr_block(
AssociationId="vpc-cidr-assoc-BORING123"
)
@ -469,7 +468,7 @@ def test_disassociate_vpc_ipv4_cidr_block():
{},
)["AssociationId"]
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
response = ec2.meta.client.disassociate_vpc_cidr_block(
AssociationId=vpc_base_cidr_assoc_id
)
@ -549,7 +548,7 @@ def test_vpc_associate_ipv6_cidr_block():
ipv6_cidr_block_association_set["AssociationId"].should.contain("vpc-cidr-assoc")
# Test Fail on adding 2nd IPV6 association - AWS only allows 1 at this time!
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
response = ec2.meta.client.associate_vpc_cidr_block(
VpcId=vpc.id, AmazonProvidedIpv6CidrBlock=True
)
@ -657,7 +656,7 @@ def test_create_vpc_with_invalid_cidr_block_parameter():
ec2 = boto3.resource("ec2", region_name="us-west-1")
vpc_cidr_block = "1000.1.0.0/20"
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
vpc = ec2.create_vpc(CidrBlock=vpc_cidr_block)
str(ex.exception).should.equal(
"An error occurred (InvalidParameterValue) when calling the CreateVpc "
@ -672,7 +671,7 @@ def test_create_vpc_with_invalid_cidr_range():
ec2 = boto3.resource("ec2", region_name="us-west-1")
vpc_cidr_block = "10.1.0.0/29"
with assert_raises(ClientError) as ex:
with pytest.raises(ClientError) as ex:
vpc = ec2.create_vpc(CidrBlock=vpc_cidr_block)
str(ex.exception).should.equal(
"An error occurred (InvalidVpc.Range) when calling the CreateVpc "

8
tests/test_ec2/test_vpn_connections.py
View file

@ -1,11 +1,11 @@
from __future__ import unicode_literals
import boto
import boto3
from nose.tools import assert_raises
import pytest
import sure # noqa
from boto.exception import EC2ResponseError
from moto import mock_ec2_deprecated, mock_ec2
from moto import mock_ec2, mock_ec2_deprecated
@mock_ec2_deprecated
@ -35,7 +35,7 @@ def test_delete_vpn_connections():
@mock_ec2_deprecated
def test_delete_vpn_connections_bad_id():
conn = boto.connect_vpc("the_key", "the_secret")
with assert_raises(EC2ResponseError):
with pytest.raises(EC2ResponseError):
conn.delete_vpn_connection("vpn-0123abcd")